Add skeleton of the proxy

mtglib/init.go

@@ -8,7 +8,19 @@ import (
 	"time"
 )
 
-var ErrSecretEmpty = errors.New("secret is empty")
+var (
+	ErrSecretEmpty                 = errors.New("secret is empty")
+	ErrSecretInvalid               = errors.New("secret is invalid")
+	ErrNetworkIsNotDefined         = errors.New("network is not defined")
+	ErrAntiReplayCacheIsNotDefined = errors.New("anti-replay cache is not defined")
+	ErrIPBlocklistIsNotDefined     = errors.New("ip blocklist is not defined")
+	ErrEventStreamIsNotDefined     = errors.New("event stream is not defined")
+	ErrLoggerIsNotDefined          = errors.New("logger is not defined")
+)
+
+const (
+	DefaultConcurrency = 4096
+)
 
 type Network interface {
 	Dial(network, address string) (net.Conn, error)

mtglib/proxy.go

@@ -0,0 +1,120 @@
+package mtglib
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/panjf2000/ants/v2"
+)
+
+type Proxy struct {
+	ctx             context.Context
+	ctxCancel       context.CancelFunc
+	streamWaitGroup sync.WaitGroup
+	workerPool      *ants.PoolWithFunc
+
+	secret          Secret
+	network         Network
+	antiReplayCache AntiReplayCache
+	ipBlocklist     IPBlocklist
+	eventStream     EventStream
+	logger          Logger
+}
+
+func (p *Proxy) ServeConn(conn net.Conn) {
+	ctx := newStreamContext(p.ctx, p.logger, conn)
+	defer ctx.Close()
+
+	p.eventStream.Send(ctx, EventStart{
+		CreatedAt: time.Now(),
+		ConnID:    ctx.connID,
+		RemoteIP:  ctx.ClientIP(),
+	})
+	ctx.logger.Info("Stream has been started")
+
+	defer func() {
+		p.eventStream.Send(ctx, EventFinish{
+			CreatedAt: time.Now(),
+			ConnID:    ctx.connID,
+		})
+		ctx.logger.Info("Stream has been finished")
+	}()
+}
+
+func (p *Proxy) Serve(listener net.Listener) error {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			return fmt.Errorf("cannot accept a new connection: %w", err)
+		}
+
+		err = p.workerPool.Invoke(conn)
+
+		switch {
+		case err == nil:
+		case errors.Is(err, ants.ErrPoolClosed):
+			return nil
+		case errors.Is(err, ants.ErrPoolOverload):
+			p.eventStream.Send(p.ctx, EventConcurrencyLimited{})
+		}
+	}
+}
+
+func (p *Proxy) Shutdown() {
+	p.ctxCancel()
+	p.streamWaitGroup.Wait()
+	p.workerPool.Release()
+}
+
+type antsLogger struct{}
+
+func (a antsLogger) Printf(msg string, args ...interface{}) {}
+
+func NewProxy(opts ProxyOpts) (*Proxy, error) {
+	switch {
+	case opts.Network == nil:
+		return nil, ErrNetworkIsNotDefined
+	case opts.AntiReplayCache == nil:
+		return nil, ErrAntiReplayCacheIsNotDefined
+	case opts.IPBlocklist == nil:
+		return nil, ErrIPBlocklistIsNotDefined
+	case opts.EventStream == nil:
+		return nil, ErrEventStreamIsNotDefined
+	case opts.Logger == nil:
+		return nil, ErrLoggerIsNotDefined
+	case !opts.Secret.Valid():
+		return nil, ErrSecretInvalid
+	}
+
+	concurrency := opts.Concurrency
+	if concurrency == 0 {
+		concurrency = DefaultConcurrency
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	proxy := &Proxy{
+		ctx:             ctx,
+		ctxCancel:       cancel,
+		secret:          opts.Secret,
+		network:         opts.Network,
+		antiReplayCache: opts.AntiReplayCache,
+		ipBlocklist:     opts.IPBlocklist,
+		eventStream:     opts.EventStream,
+		logger:          opts.Logger.Named("proxy"),
+	}
+
+	pool, err := ants.NewPoolWithFunc(int(concurrency), func(arg interface{}) {
+		proxy.ServeConn(arg.(net.Conn))
+	}, ants.WithLogger(antsLogger{}))
+	if err != nil {
+		return nil, fmt.Errorf("cannot initialize a pool: %w", err)
+	}
+
+	proxy.workerPool = pool
+
+	return proxy, nil
+}

mtglib/proxy_opts.go

@@ -0,0 +1,12 @@
+package mtglib
+
+type ProxyOpts struct {
+	Secret          Secret
+	Network         Network
+	AntiReplayCache AntiReplayCache
+	IPBlocklist     IPBlocklist
+	EventStream     EventStream
+	Logger          Logger
+
+	Concurrency uint
+}

mtglib/stream_context.go

@@ -0,0 +1,63 @@
+package mtglib
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/base64"
+	"net"
+	"time"
+)
+
+type streamContext struct {
+	ctx        context.Context
+	ctxCancel  context.CancelFunc
+	clientConn net.Conn
+	connID     string
+	logger     Logger
+}
+
+func (s *streamContext) Deadline() (time.Time, bool) {
+	return s.ctx.Deadline()
+}
+
+func (s *streamContext) Done() <-chan struct{} {
+	return s.ctx.Done()
+}
+
+func (s *streamContext) Err() error {
+	return s.ctx.Err()
+}
+
+func (s *streamContext) Value(key interface{}) interface{} {
+	return s.ctx.Value(key)
+}
+
+func (s *streamContext) Close() {
+	s.ctxCancel()
+	s.clientConn.Close()
+}
+
+func (s *streamContext) ClientIP() net.IP {
+	return s.clientConn.RemoteAddr().(*net.TCPAddr).IP
+}
+
+func newStreamContext(ctx context.Context, logger Logger, clientConn net.Conn) *streamContext {
+	connIDBytes := make([]byte, 16)
+
+	if _, err := rand.Read(connIDBytes); err != nil {
+		panic(err)
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	streamCtx := &streamContext{
+		ctx:        ctx,
+		ctxCancel:  cancel,
+		clientConn: clientConn,
+		connID:     base64.RawURLEncoding.EncodeToString(connIDBytes),
+	}
+	streamCtx.logger = logger.
+		BindStr("stream-id", streamCtx.connID).
+		BindStr("client-ip", streamCtx.ClientIP().String())
+
+	return streamCtx
+}