code: func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event, prefixRegex *parsers.ExtRegexp) {
  	wg := sync.WaitGroup{}
  	for i := 0; i < numParsers; i++ {
  		wg.Add(1)
  		go func() {
  			for line := range lines {
  				line = strings.TrimSpace(line)
  				// take care of any headers on the line
  				var prefixFields map[string]string
  				if prefixRegex != nil {
  					var prefix string
  					prefix, prefixFields = prefixRegex.FindStringSubmatchMap(line)
  					line = strings.TrimPrefix(line, prefix)
  				}
  
  				values, err := p.lineParser.ParseLine(line)
  				// we get a bunch of errors from the parser on ArangoDB logs, skip em
  				if err == nil {
  					timestamp, err := p.parseTimestamp(values)
  					if err != nil {
  						logSkipped(line, "couldn't parse logline timestamp, skipping")
  						continue
  					}
  
  					// merge the prefix fields and the parsed line contents
  					for k, v := range prefixFields {
  						values[k] = v
  					}
  
  					logrus.WithFields(logrus.Fields{
  						"line":   line,
  						"values": values,
  					}).Debug("Successfully parsed line")
  
  					// we'll be putting the timestamp in the Event
  					// itself, no need to also have it in the Data
  					delete(values, timestampFieldName)
  
  					send <- event.Event{
  						Timestamp: timestamp,
  						Data:      values,
  					}
  				} else {
  					logSkipped(line, "logline didn't parse, skipping.")
  				}
  			}
  			wg.Done()
  		}()
  	}
  	wg.Wait()
  	logrus.Debug("lines channel is closed, ending arangodb processor")
  }
docs: // ProcessLines method for Parser.