func handleConn(ctx context.Context, w *data.Writer, conn net.Conn) (
	err error) {
	var wg sync.WaitGroup
	defer wg.Wait()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	wg.Add(1)
	go func() {
		defer wg.Done()
		<-ctx.Done()
		conn.Close()
	}()
	scanner := bufio.NewScanner(conn)
	for scanner.Scan() {
		err := handleLine(ctx, w, scanner.Bytes())
		if err != nil {
			external.Errorw("invalid graphite line",
				"line", scanner.Text(),
				"peer", conn.RemoteAddr().String(),
				"err", err.Error(),
			)
		}
	}
	return scanner.Err()
}
handleConn handles lines from the connection and adds them to the writer.
package graphite
import (
	"bufio"
	"bytes"
	"context"
	"net"
	"strconv"
	"sync"
	"github.com/vivint/rothko/data"
	"github.com/vivint/rothko/external"
	"github.com/zeebo/errs"
)
type Listener struct {
	address string
}
func New(address string) *Listener {
	return &Listener{
		address: address,
	}
}
func (l *Listener) Run(ctx context.Context, w *data.Writer) (err error) {
	lis, err := net.Listen("tcp", l.address)
	if err != nil {
		return errs.Wrap(err)
	}
	defer lis.Close()
	var wg sync.WaitGroup
	var errs = make(chan error, 1)
	wg.Add(1)
	go func() {
		defer wg.Done()
		errs <- handleListener(ctx, w, lis)
	}()
	select {
	case err := <-errs:
		return err
	case <-ctx.Done():
		lis.Close()
		wg.Wait()
		return nil
	}
}
func handleListener(ctx context.Context, w *data.Writer, lis net.Listener) (
	err error) {
	var wg sync.WaitGroup
	defer wg.Wait()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	for {
		conn, err := lis.Accept()
		if err != nil {
			return err
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			err := handleConn(ctx, w, conn)
			if err != nil {
				external.Errorw("graphite connection error",
					"err", err.Error(),
				)
			}
		}()
	}
}