implement listener wrapper for use with http app

go.sum

@@ -86,8 +86,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
-github.com/caddyserver/caddy/v2 v2.6.2-0.20220930193937-fe91de67b62c h1:Fy/xDSXQ9i202+wEW03Oqe5DqvRrumio+YVUg6FPPzs=
-github.com/caddyserver/caddy/v2 v2.6.2-0.20220930193937-fe91de67b62c/go.mod h1:c7/x6iIgUIAuXv2gFovbn4zMqwQpEX0HrOq6UihOeNs=
+github.com/caddyserver/caddy/v2 v2.6.2 h1:wKoFIxpmOJLGl3QXoo6PNbYvGW4xLEgo32GPBEjWL8o=
+github.com/caddyserver/caddy/v2 v2.6.2/go.mod h1:ICM4D+OiSexKF077f92MzFRlbkmX4tu4TB8DJAG/lUk=
 github.com/caddyserver/certmagic v0.17.2 h1:o30seC1T/dBqBCNNGNHWwj2i5/I/FMjBbTAhjADP3nE=
 github.com/caddyserver/certmagic v0.17.2/go.mod h1:ouWUuC490GOLJzkyN35eXfV8bSbwMwSf4bdhkIxtdQE=
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=

layer4/connection.go

@@ -174,6 +174,9 @@ var (
 
 	// ReplacerCtxKey is the key used to store the replacer.
 	ReplacerCtxKey caddy.CtxKey = "replacer"
+
+	// listenerCtxKey is the key used to get the listener from a handler
+	listenerCtxKey caddy.CtxKey = "listener"
 )
 
 var bufPool = sync.Pool{

layer4/handlers.go

@@ -74,3 +74,10 @@ func (h HandlerFunc) Handle(cx *Connection) error { return h(cx) }
 type nopHandler struct{}
 
 func (nopHandler) Handle(_ *Connection) error { return nil }
+
+// listenerHandler is a connection handler that pipe incoming connection to channel as a listener wrapper
+type listenerHandler struct{}
+
+func (listenerHandler) Handle(conn *Connection) error {
+	return conn.Context.Value(listenerCtxKey).(*listener).pipeConnection(conn)
+}

layer4/listener.go

@@ -0,0 +1,157 @@
+package layer4
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"github.com/caddyserver/caddy/v2"
+	"go.uber.org/zap"
+	"net"
+	"runtime"
+	"sync"
+	"time"
+)
+
+func init() {
+	caddy.RegisterModule(ListenerWrapper{})
+}
+
+// ListenerWrapper is a Caddy module that wraps App as a listener wrapper, it doesn't support udp.
+type ListenerWrapper struct {
+	// Routes express composable logic for handling byte streams.
+	Routes RouteList `json:"routes,omitempty"`
+
+	compiledRoute Handler
+
+	logger *zap.Logger
+	ctx    caddy.Context
+}
+
+// CaddyModule returns the Caddy module information.
+func (ListenerWrapper) CaddyModule() caddy.ModuleInfo {
+	return caddy.ModuleInfo{
+		ID:  "caddy.listeners.layer4",
+		New: func() caddy.Module { return new(ListenerWrapper) },
+	}
+}
+
+// Provision sets up the ListenerWrapper.
+func (lw *ListenerWrapper) Provision(ctx caddy.Context) error {
+	lw.ctx = ctx
+	lw.logger = ctx.Logger()
+
+	err := lw.Routes.Provision(ctx)
+	if err != nil {
+		return err
+	}
+	lw.compiledRoute = lw.Routes.Compile(listenerHandler{}, lw.logger)
+
+	return nil
+}
+
+func (lw *ListenerWrapper) WrapListener(l net.Listener) net.Listener {
+	// TODO make channel capacity configurable
+	connChan := make(chan net.Conn, runtime.GOMAXPROCS(0))
+	li := &listener{
+		Listener:      l,
+		logger:        lw.logger,
+		compiledRoute: lw.compiledRoute,
+		connChan:      connChan,
+		wg:            new(sync.WaitGroup),
+	}
+	go li.loop()
+	return li
+}
+
+type listener struct {
+	net.Listener
+	logger        *zap.Logger
+	compiledRoute Handler
+
+	// closed when there is a non-recoverable error and all handle goroutines are done
+	connChan chan net.Conn
+	err      error
+
+	// count running handles
+	wg *sync.WaitGroup
+}
+
+// loop accept connection from underlying listener and pipe the connection if there are any
+func (l *listener) loop() {
+	for {
+		conn, err := l.Listener.Accept()
+		if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
+			l.logger.Error("temporary error accepting connection", zap.Error(err))
+			continue
+		}
+		if err != nil {
+			l.err = err
+			break
+		}
+
+		l.wg.Add(1)
+		go l.handle(conn)
+	}
+
+	// closing remaining conns in channel to release resources
+	go func() {
+		l.wg.Wait()
+		close(l.connChan)
+	}()
+	for conn := range l.connChan {
+		conn.Close()
+	}
+}
+
+// errHijacked is used when a handler takes over the connection, it's lifetime is not managed by handle
+var errHijacked = errors.New("hijacked connection")
+
+func (l *listener) handle(conn net.Conn) {
+	var err error
+	defer func() {
+		l.wg.Done()
+		if err != errHijacked {
+			conn.Close()
+		}
+	}()
+
+	buf := bufPool.Get().(*bytes.Buffer)
+	buf.Reset()
+	defer bufPool.Put(buf)
+
+	cx := WrapConnection(conn, buf)
+	cx.Context = context.WithValue(cx.Context, listenerCtxKey, l)
+
+	start := time.Now()
+	err = l.compiledRoute.Handle(cx)
+	duration := time.Since(start)
+	if err != nil && err != errHijacked {
+		l.logger.Error("handling connection", zap.Error(err))
+	}
+
+	l.logger.Debug("connection stats",
+		zap.String("remote", cx.RemoteAddr().String()),
+		zap.Uint64("read", cx.bytesRead),
+		zap.Uint64("written", cx.bytesWritten),
+		zap.Duration("duration", duration),
+	)
+}
+
+func (l *listener) Accept() (net.Conn, error) {
+	for conn := range l.connChan {
+		return conn, nil
+	}
+	return nil, l.err
+
+}
+
+func (l *listener) pipeConnection(conn net.Conn) error {
+	l.connChan <- conn
+	return errHijacked
+}
+
+// Interface guards
+var (
+	_ caddy.Module          = (*ListenerWrapper)(nil)
+	_ caddy.ListenerWrapper = (*ListenerWrapper)(nil)
+)