Lines Matching refs:tail
4 package tail package
142 func (tail *Tail) Tell() (offset int64, err error) {
143 if tail.file == nil {
146 offset, err = tail.file.Seek(0, os.SEEK_CUR)
151 tail.lk.Lock()
152 defer tail.lk.Unlock()
153 if tail.reader == nil {
157 offset -= int64(tail.reader.Buffered())
162 func (tail *Tail) Stop() error {
163 tail.Kill(nil)
164 return tail.Wait()
168 func (tail *Tail) StopAtEOF() error {
169 tail.Kill(errStopAtEOF)
170 return tail.Wait()
175 func (tail *Tail) close() {
176 close(tail.Lines)
177 tail.closeFile()
180 func (tail *Tail) closeFile() {
181 if tail.file != nil {
182 tail.file.Close()
183 tail.file = nil
187 func (tail *Tail) reopen() error {
188 tail.closeFile()
191 tail.file, err = OpenFile(tail.Filename)
194 tail.Logger.Printf("Waiting for %s to appear...", tail.Filename)
195 if err := tail.watcher.BlockUntilExists(&tail.Tomb); err != nil {
199 return fmt.Errorf("Failed to detect creation of %s: %s", tail.Filename, err)
203 return fmt.Errorf("Unable to open file %s: %s", tail.Filename, err)
210 func (tail *Tail) readLine() (string, error) {
211 tail.lk.Lock()
212 line, err := tail.reader.ReadString('\n')
213 tail.lk.Unlock()
226 func (tail *Tail) tailFileSync() {
227 defer tail.Done()
228 defer tail.close()
230 if !tail.MustExist {
232 err := tail.reopen()
235 tail.Kill(err)
242 if tail.Location != nil {
243 _, err := tail.file.Seek(tail.Location.Offset, tail.Location.Whence)
244 tail.Logger.Printf("Seeked %s - %+v\n", tail.Filename, tail.Location)
246 tail.Killf("Seek error on %s: %s", tail.Filename, err)
251 tail.openReader()
259 if !tail.Pipe {
261 offset, err = tail.Tell()
263 tail.Kill(err)
268 line, err := tail.readLine()
272 cooloff := !tail.sendLine(line)
279 tail.Lines <- &Line{msg, time.Now(), fmt.Errorf(msg)}
282 case <-tail.Dying():
285 if err := tail.seekEnd(); err != nil {
286 tail.Kill(err)
291 if !tail.Follow {
293 tail.sendLine(line)
298 if tail.Follow && line != "" {
301 err := tail.seekTo(SeekInfo{Offset: offset, Whence: 0})
303 tail.Kill(err)
311 err := tail.waitForChanges()
314 tail.Kill(err)
320 tail.Killf("Error reading %s: %s", tail.Filename, err)
325 case <-tail.Dying():
326 if tail.Err() == errStopAtEOF {
338 func (tail *Tail) waitForChanges() error {
339 if tail.changes == nil {
340 pos, err := tail.file.Seek(0, os.SEEK_CUR)
344 tail.changes, err = tail.watcher.ChangeEvents(&tail.Tomb, pos)
351 case <-tail.changes.Modified:
353 case <-tail.changes.Deleted:
354 tail.changes = nil
355 if tail.ReOpen {
357 tail.Logger.Printf("Re-opening moved/deleted file %s ...", tail.Filename)
358 if err := tail.reopen(); err != nil {
361 tail.Logger.Printf("Successfully reopened %s", tail.Filename)
362 tail.openReader()
365 tail.Logger.Printf("Stopping tail as file no longer exists: %s", tail.Filename)
368 case <-tail.changes.Truncated:
370 tail.Logger.Printf("Re-opening truncated file %s ...", tail.Filename)
371 if err := tail.reopen(); err != nil {
374 tail.Logger.Printf("Successfully reopened truncated %s", tail.Filename)
375 tail.openReader()
377 case <-tail.Dying():
383 func (tail *Tail) openReader() {
384 if tail.MaxLineSize > 0 {
386 tail.reader = bufio.NewReaderSize(tail.file, tail.MaxLineSize+2)
388 tail.reader = bufio.NewReader(tail.file)
392 func (tail *Tail) seekEnd() error {
393 return tail.seekTo(SeekInfo{Offset: 0, Whence: os.SEEK_END})
396 func (tail *Tail) seekTo(pos SeekInfo) error {
397 _, err := tail.file.Seek(pos.Offset, pos.Whence)
399 return fmt.Errorf("Seek error on %s: %s", tail.Filename, err)
402 tail.reader.Reset(tail.file)
408 func (tail *Tail) sendLine(line string) bool {
413 if tail.MaxLineSize > 0 && len(line) > tail.MaxLineSize {
414 lines = util.PartitionString(line, tail.MaxLineSize)
418 tail.Lines <- &Line{line, now, nil}
421 if tail.Config.RateLimiter != nil {
422 ok := tail.Config.RateLimiter.Pour(uint16(len(lines)))
424 tail.Logger.Printf("Leaky bucket full (%v); entering 1s cooloff period.\n",
425 tail.Filename)
436 func (tail *Tail) Cleanup() {
437 watch.Cleanup(tail.Filename)