code: func (p *Processor) RunStreams(streams []*Stream) error {
  	ingestMu := sync.Mutex{}
  	ingest := func(s *Stream, l string) error {
  		ingestMu.Lock()
  		defer ingestMu.Unlock()
  
  		return p.IngestLine(s, l)
  	}
  
  	// Read from all configured streams until they are finished.
  	return parallel.FanOutIn(func(taskC chan<- func() error) {
  		for _, s := range streams {
  			s := s
  			bufferSize := s.BufferSize
  			if bufferSize <= 0 {
  				bufferSize = DefaultBufferSize
  			}
  			taskC <- func() error {
  				lr := newLineReader(s.Reader, bufferSize)
  				for {
  					line, err := lr.readLine()
  					if err != nil {
  						if err == io.EOF {
  							return nil
  						}
  						return err
  					}
  					if err := ingest(s, line); err != nil {
  						log.Fields{
  							log.ErrorKey: err,
  							"stream":     s.Name,
  							"line":       line,
  						}.Errorf(p.ctx, "Failed to ingest line.")
  					}
  				}
  			}
  		}
  	})
  }
docs: // RunStreams executes the Processor, consuming data from its configured streams
  // and forwarding it to LogDog. Run will block until all streams have
  // terminated.
  //
  // If a stream terminates with an error, or if there is an error processing the
  // stream data, Run will return an error. If multiple Streams fail with errors,
  // an errors.MultiError will be returned. io.EOF does not count as an error.