| 68 | } |
| 69 | |
| 70 | func Encode(stdout <-chan string, stderr <-chan string, writer io.Writer) error { |
| 71 | enc := json.NewEncoder(writer) |
| 72 | var encMu sync.Mutex |
| 73 | var wg sync.WaitGroup |
| 74 | wg.Add(2) |
| 75 | f := func(dataChan <-chan string, name string) { |
| 76 | defer wg.Done() |
| 77 | e := &Entry{ |
| 78 | Stream: name, |
| 79 | } |
| 80 | for logEntry := range dataChan { |
| 81 | e.Log = logEntry |
| 82 | e.Time = time.Now().UTC() |
| 83 | encMu.Lock() |
| 84 | encErr := enc.Encode(e) |
| 85 | encMu.Unlock() |
| 86 | if encErr != nil { |
| 87 | log.L.WithError(encErr).Errorf("failed to encode JSON") |
| 88 | return |
| 89 | } |
| 90 | } |
| 91 | } |
| 92 | go f(stdout, "stdout") |
| 93 | go f(stderr, "stderr") |
| 94 | wg.Wait() |
| 95 | return nil |
| 96 | } |
| 97 | |
| 98 | func writeEntry(e *Entry, stdout, stderr io.Writer, refTime time.Time, timestamps bool, since string, until string) error { |
| 99 | output := []byte{} |