// LogStreamer =========================
558870
func newLogsStreamer(c *Context) (empire.LogsStreamer, error) {
	switch c.String(FlagLogsStreamer) {
	case "kinesis":
		return newKinesisLogsStreamer(c)
	default:
		log.Println("Streaming logs are disabled")
		return nil, nil
	}
}
423490
func Logger(logger *zap.Logger) TransportOption {
	return func(transportOptions *transportOptions) {
		transportOptions.logger = logger
	}
}
157850
func Logger(l *log.Logger) TokenLimiterOption {
	return func(tl *TokenLimiter) error {
		tl.log = l
		return nil
	}
}
1130379
func NewLogWatcher() *LogWatcher {
	return &LogWatcher{
		Msg:          make(chan *Message, logWatcherBufferSize),
		Err:          make(chan error, 1),
		producerGone: make(chan struct{}),
		consumerGone: make(chan struct{}),
	}
}
648421
func (l *logStream) putLogEvents(events []*cloudwatchlogs.InputLogEvent, sequenceToken *string) (*string, error) {
	input := &cloudwatchlogs.PutLogEventsInput{
		LogEvents:     events,
		SequenceToken: sequenceToken,
		LogGroupName:  aws.String(l.logGroupName),
		LogStreamName: aws.String(l.logStreamName),
	}
	resp, err := l.client.PutLogEvents(input)
	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			logrus.WithFields(logrus.Fields{
				"errorCode":     awsErr.Code(),
				"message":       awsErr.Message(),
				"origError":     awsErr.OrigErr(),
				"logGroupName":  l.logGroupName,
				"logStreamName": l.logStreamName,
			}).Error("Failed to put log events")
		}
		return nil, err
	}
	return resp.NextSequenceToken, nil
}
643823
func NewStreamHandler(
	name string, level LogLevelType, stream Stream) *StreamHandler {

	object := &StreamHandler{
		BaseHandler: NewBaseHandler(name, level),
		stream:      stream,
	}
	Closer.AddHandler(object)
	return object
}
1133456
func (log Logger) Logc(lvl Level, closure func() string) {
	log.intLogc(lvl, closure)
}
802390
func (l *logStream) Close() error {
	l.lock.Lock()
	defer l.lock.Unlock()
	if !l.closed {
		close(l.messages)
	}
	l.closed = true
	return nil
}
107707
func (logger *Logger) watcher() {
	var buf bytes.Buffer
	for {
		timeout := time.After(time.Millisecond * logger.timeInterval)

		for i := 0; i < logger.bufferSize; i++ {
			select {
			case msg := <-logger.queue:
				fmt.Fprintln(&buf, msg)
			case req := <-logger.request:
				logger.flushReq(&buf, &req)
			case <-timeout:
				i = logger.bufferSize
			case <-logger.flush:
				logger.flushBuf(&buf)
				logger.finish <- true
				i = logger.bufferSize
			case <-logger.quit:
				// If quit signal received, cleans the channel
				// and writes all of them to io.Writer.
				for {
					select {
					case msg := <-logger.queue:
						fmt.Fprintln(&buf, msg)
					case req := <-logger.request:
						logger.flushReq(&buf, &req)
					case <-logger.flush:
						// do nothing
					default:
						logger.flushBuf(&buf)
						logger.quit <- true
						return
					}
				}
			}
		}
		logger.flushBuf(&buf)
	}
}
896197
func (a *Adapter) Stream(logstream chan *router.Message) {
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		a.subAdapter.Stream(a.out)
		wg.Done()
	}()
	defer func() {
		for _, message := range a.buffers {
			a.out <- message
		}

		close(a.out)
		wg.Wait()
	}()

	for {
		select {
		case message, ok := <-logstream:
			if !ok {
				return
			}

			if !multilineContainer(message.Container, a.enableByDefault) {
				a.out <- message
				continue
			}

			cID := message.Container.ID
			old, oldExists := a.buffers[cID]
			if a.isFirstLine(message) {
				if oldExists {
					a.out <- old
				}

				a.buffers[cID] = message
			} else {
				isLastLine := a.isLastLine(message)
				
				if oldExists {
					old.Data += a.separator + message.Data
					message = old
				}

				if isLastLine {
					a.out <- message
					if oldExists {
						delete(a.buffers, cID)
					}
				} else {
					a.buffers[cID] = message
				}
			}
		case <-a.nextCheck:
			now := time.Now()

			for key, message := range a.buffers {
				if message.Time.Add(a.flushAfter).After(now) {
					a.out <- message
					delete(a.buffers, key)
				}
			}

			a.nextCheck = time.After(a.checkInterval)
		}
	}
}
215175
func (l *Logger) AddStream(stream Stream) error {
	if err := stream.init(l.config); err != nil {
		return err
	}
	l.streams = append(l.streams, stream)
	return nil
}
229268
func (r *runner) hookStdoutStderr(ctx context.Context, logdogServ *logdogServer, stdout, stderr io.ReadCloser, streamNamePrefix string) error {
	tsNow, err := ptypes.TimestampProto(clock.Now(ctx))
	if err != nil {
		return err
	}

	hook := func(rc io.ReadCloser, name string, tee streamproto.TeeType) error {
		return logdogServ.AddStream(rc, &streamproto.Properties{
			LogStreamDescriptor: &logpb.LogStreamDescriptor{
				Name:        path.Join(streamNamePrefix, name),
				ContentType: "text/plain",
				Timestamp:   tsNow,
			},
			Tee: tee,
		})
	}

	if err := hook(stdout, "stdout", streamproto.TeeStdout); err != nil {
		return err
	}
	if err := hook(stderr, "stderr", streamproto.TeeStderr); err != nil {
		return err
	}
	return nil
}
303920
func Logger(l *log.Logger) CircuitBreakerOption {
	return func(c *CircuitBreaker) error {
		c.log = l
		return nil
	}
}
1299108
func (l *Log) Stream(header string) Stream {
	return &stream{
		header: header,
		log:    l,
		quiet:  l.quiet,
	}
}
91525
func (logger *StreamLogger) ServeLogs(url string, logf LogFormatter) {
	http.HandleFunc(url, func(w http.ResponseWriter, r *http.Request) {
		if err := acl.CheckAccessHTTP(r, acl.DEBUGGING); err != nil {
			acl.SendError(w, err)
			return
		}
		if err := r.ParseForm(); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
		}
		ch := logger.Subscribe("ServeLogs")
		defer logger.Unsubscribe(ch)

		// Notify client that we're set up. Helpful to distinguish low-traffic streams from connection issues.
		w.WriteHeader(http.StatusOK)
		w.(http.Flusher).Flush()

		for message := range ch {
			if err := logf(w, r.Form, message); err != nil {
				return
			}
			w.(http.Flusher).Flush()
		}
	})
	log.Infof("Streaming logs from %s at %v.", logger.Name(), url)
}
112311
func (sinkConfig *SinkConfig) BuildLogger() (log.Logger, map[string]*loggers.CaptureLogger, error) {
	return BuildLoggerFromSinkConfig(sinkConfig, make(map[string]*loggers.CaptureLogger))
}
1155124
func Logger(logger *zap.Logger) Option {
	return optionFunc(func(opts *options) {
		opts.logger = logger
	})
}