Method handleConn

internal/tailer/logstream/socketstream.go:107–156

(ctx context.Context, wg *sync.WaitGroup, waker waker.Waker, c net.Conn)

}

func (ss *socketStream) handleConn(ctx context.Context, wg *sync.WaitGroup, waker waker.Waker, c net.Conn) {
	defer wg.Done()

	lr := NewLineReader(ss.sourcename, ss.lines, c, defaultReadBufferSize, ss.cancel)
	var total int
	defer func() {
		glog.V(2).Infof("stream(%s): read total %d bytes from %s", ss.sourcename, c, total)
		glog.V(2).Infof("stream(%s): closing connection, %v", ss.sourcename, c)
		err := c.Close()
		if err != nil {
			logErrors.Add(ss.address, 1)
			glog.Info(err)
		}
		lr.Finish(ctx)
		logCloses.Add(ss.address, 1)
	}()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	SetReadDeadlineOnDone(ctx, c)

	for {
		n, err := lr.ReadAndSend(ctx)
		glog.V(2).Infof("stream(%s): read %d bytes, err is %v", ss.sourcename, n, err)

		if n > 0 {
			total += n

			// No error implies more to read, so restart the loop.
			if err == nil && ctx.Err() == nil {
				continue
			}
		}

		if IsExitableError(err) {
			glog.V(2).Infof("stream(%s): exiting, conn has error %s", ss.sourcename, err)
			return
		}

		// Yield and wait
		glog.V(2).Infof("stream(%s): waiting", ss.sourcename)
		select {
		case <-ctx.Done():
			// Exit after next read attempt.
			glog.V(2).Infof("stream(%s:%s): context cancelled, exiting after next read timeout", ss.scheme, ss.address)
		case <-waker.Wake():
			// sleep until next Wake()
			glog.V(2).Infof("stream(%s): Wake received", ss.sourcename)
		}
	}
}