func (p *Parser) handleEvents(rawEvents <-chan []string, send chan<- event.Event, wg *sync.WaitGroup) {
	defer wg.Done()
	for rawEvent := range rawEvents {
		ev := p.handleEvent(rawEvent)
		if ev != nil {
			send <- *ev
		}
	}
}
handleEvents receives sets of grouped log lines, each representing a single
log statement. It attempts to parse them, and sends the events it constructs
down the send channel.
package postgresql
import (
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/sirupsen/logrus"
	"github.com/honeycombio/honeytail/event"
	"github.com/honeycombio/honeytail/parsers"
	"github.com/honeycombio/mysqltools/query/normalizer"
)
const (
	timestampRe   = `\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[.0-9]* [A-Z]+`
	defaultPrefix = "%t [%p-%l] %u@%d"
	slowQueryHeader = `\s*(?P<level>[A-Z0-9]+):\s+duration: (?P<duration>[0-9\.]+) ms\s+(?:(statement)|(execute \S+)): `
)
var slowQueryHeaderRegex = &parsers.ExtRegexp{regexp.MustCompile(slowQueryHeader)}
type prefixField struct {
	Name    string
	Pattern string
}
func (p *prefixField) ReString() string {
	return fmt.Sprintf("(?P<%s>%s)", p.Name, p.Pattern)
}
var prefixValues = map[string]prefixField{
	"%a": prefixField{Name: "application", Pattern: "\\S+"},
	"%u": prefixField{Name: "user", Pattern: "\\S+"},
	"%d": prefixField{Name: "database", Pattern: "\\S+"},
	"%r": prefixField{Name: "host_port", Pattern: "\\S+"},
	"%h": prefixField{Name: "host", Pattern: "\\S+"},
	"%p": prefixField{Name: "pid", Pattern: "\\d+"},
	"%t": prefixField{Name: "timestamp", Pattern: timestampRe},
	"%m": prefixField{Name: "timestamp_millis", Pattern: timestampRe},
	"%n": prefixField{Name: "timestamp_unix", Pattern: "\\d+"},
	"%i": prefixField{Name: "command_tag", Pattern: "\\S+"},
	"%e": prefixField{Name: "sql_state", Pattern: "\\S+"},
	"%c": prefixField{Name: "session_id", Pattern: "\\d+"},
	"%l": prefixField{Name: "session_line_number", Pattern: "\\d+"},
	"%s": prefixField{Name: "session_start", Pattern: timestampRe},
	"%v": prefixField{Name: "virtual_transaction_id", Pattern: "\\S+"},
	"%x": prefixField{Name: "transaction_id", Pattern: "\\S+"},
}
type Options struct {
	LogLinePrefix string `long:"log_line_prefix" description:"Format string for PostgreSQL log line prefix"`
}
type Parser struct {
	pgPrefixRegex *parsers.ExtRegexp
}
func (p *Parser) Init(options interface{}) (err error) {
	var logLinePrefixFormat string
	conf, ok := options.(*Options)
	if !ok || conf.LogLinePrefix == "" {
		logLinePrefixFormat = defaultPrefix
	} else {
		logLinePrefixFormat = conf.LogLinePrefix
	}
	p.pgPrefixRegex, err = buildPrefixRegexp(logLinePrefixFormat)
	return err
}
func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event, prefixRegex *parsers.ExtRegexp) {
	rawEvents := make(chan []string)
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go p.handleEvents(rawEvents, send, wg)
	var groupedLines []string
	for line := range lines {
		if prefixRegex != nil {
			var prefix string
			prefix = prefixRegex.FindString(line)
			line = strings.TrimPrefix(line, prefix)
		}
		if !isContinuationLine(line) && len(groupedLines) > 0 {
			rawEvents <- groupedLines
			groupedLines = make([]string, 0, 1)
		}
		groupedLines = append(groupedLines, line)
	}
	rawEvents <- groupedLines
	close(rawEvents)
	wg.Wait()
}