Merge pull request #39 from kapetan-io/thrawn/nil-pointer-issue

Add InMemoryListener integration for daemon testing

Author: thrawn01

.github/workflows/lint.yaml

@@ -11,7 +11,7 @@ on:
       - main
 
 env:
-  GOLANGCI_LINT_VERSION: v1.61.0
+  GOLANGCI_LINT_VERSION: v2.2.1
 
 jobs:
   lint:
@@ -34,7 +34,7 @@ jobs:
         run: go mod tidy && git diff --exit-code
 
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v8
         with:
           version: ${{ env.GOLANGCI_LINT_VERSION }}
           skip-cache: true # cache/restore is done by actions/setup-go@v3 step

.gitignore

@@ -10,3 +10,4 @@ coverage.html
 .env
 docs/notes
 docs/plans
+testdata/

Makefile

@@ -1,6 +1,6 @@
 .DEFAULT_GOAL := build
 LINT = $(GOPATH)/bin/golangci-lint
-LINT_VERSION = v1.64.6
+LINT_VERSION = v2.2.1
 VERSION=$(shell git describe --tags --exact-match 2>/dev/null || echo "dev-build")
 
 .PHONY: install

client.go

@@ -14,6 +14,7 @@ import (
 	"github.com/kapetan-io/tackle/clock"
 	"github.com/kapetan-io/tackle/set"
 	"google.golang.org/protobuf/proto"
+	"net"
 	"net/http"
 )
 
@@ -400,6 +401,24 @@ func WithTLS(tls *tls.Config, address string) ClientConfig {
 	}
 }
 
+// WithConn returns ClientConfig suitable for use with a custom net.Conn connection
+func WithConn(conn net.Conn) ClientConfig {
+	return ClientConfig{
+		Endpoint: "http://memory",
+		Client: &http.Client{
+			Transport: &http.Transport{
+				Dial: func(network, addr string) (net.Conn, error) {
+					return conn, nil
+				},
+				MaxConnsPerHost:     1,
+				MaxIdleConns:        1,
+				MaxIdleConnsPerHost: 1,
+				IdleConnTimeout:     60 * clock.Second,
+			},
+		},
+	}
+}
+
 type ItemsWithIDs interface {
 	GetId() string
 }

common_test.go

@@ -59,7 +59,6 @@ func TestMain(m *testing.M) {
 	}
 
 	goleak.VerifyTestMain(m)
-	//os.Exit(m.Run())
 }
 
 // ---------------------------------------------------------------------

daemon/config.go

@@ -25,6 +25,12 @@ type Config struct {
 	// single `/queue.produce` request including the size of all fields in the marshalled protobuf.
 	// The default size is 1MB.
 	MaxProducePayloadSize int64
+
+	// InMemoryListener is true if daemon should ignore ListenAddress and use net.Pipe to listen for
+	// and handle new connections. When true, calls to Daemon.Client() and Daemon.MustClient() will return
+	// a new instance of the client bound to the client portion of a net.Pipe. This is useful for testing
+	// querator where access to the loop back is not allowed, or when using testing/synctest
+	InMemoryListener bool
 }
 
 func (c *Config) ClientTLS() *tls.Config {

daemon/daemon.go

@@ -80,7 +80,11 @@ func (d *Daemon) Start(ctx context.Context) error {
 	), d.conf.MaxProducePayloadSize, d.conf.Log)
 	registry.MustRegister(handler)
 
-	if d.conf.ServerTLS() != nil {
+	if d.conf.InMemoryListener {
+		if err := d.spawnInMemory(ctx, handler); err != nil {
+			return err
+		}
+	} else if d.conf.ServerTLS() != nil {
 		if err := d.spawnHTTPS(ctx, handler); err != nil {
 			return err
 		}
@@ -114,12 +118,34 @@ func (d *Daemon) Service() *querator.Service {
 func (d *Daemon) MustClient() *querator.Client {
 	c, err := d.Client()
 	if err != nil {
-		panic(fmt.Sprintf("[%s] failed to init daemon client - '%d'", d.conf.InstanceID, err))
+		panic(fmt.Sprintf("[%s] failed to init daemon client - '%s'", d.conf.InstanceID, err))
 	}
 	return c
 }
 
 func (d *Daemon) Client() (*querator.Client, error) {
+	if d.conf.InMemoryListener {
+		// Create a new net.Pipe for each client connection
+		clientConn, serverConn := net.Pipe()
+
+		// Serve the server side of the pipe through the InMemoryListener
+		if inMemListener, ok := d.Listener.(*InMemoryListener); ok {
+			if err := inMemListener.ServeConn(serverConn); err != nil {
+				_ = clientConn.Close()
+				_ = serverConn.Close()
+				return nil, fmt.Errorf("failed to serve connection: %w", err)
+			}
+		} else {
+			_ = clientConn.Close()
+			_ = serverConn.Close()
+			return nil, fmt.Errorf("InMemoryListener is enabled but listener is not of type *InMemoryListener")
+		}
+
+		// Create a new client using the client side of the pipe
+		return querator.NewClient(querator.WithConn(clientConn))
+	}
+
+	// Original logic for non-InMemoryListener clients
 	var err error
 	if d.client != nil {
 		return d.client, nil
@@ -201,3 +227,27 @@ func (d *Daemon) spawnHTTP(ctx context.Context, h http.Handler) error {
 	d.servers = append(d.servers, srv)
 	return nil
 }
+
+func (d *Daemon) spawnInMemory(ctx context.Context, h http.Handler) error {
+	srv := &http.Server{
+		ErrorLog: slog.NewLogLogger(d.conf.Log.Handler(), slog.LevelError),
+		Handler:  h,
+	}
+
+	d.Listener = NewInMemoryListener()
+	srv.Addr = d.Listener.Addr().String()
+
+	d.wg.Add(1)
+	go func() {
+		defer d.wg.Done()
+		if err := srv.Serve(d.Listener); err != nil {
+			if !errors.Is(err, http.ErrServerClosed) {
+				d.conf.Log.Error("while starting InMemory server", "error", err)
+			}
+		}
+	}()
+
+	d.conf.Log.Info("InMemory Server Started", "address", d.Listener.Addr().String())
+	d.servers = append(d.servers, srv)
+	return nil
+}

daemon/listener.go

@@ -0,0 +1,56 @@
+package daemon
+
+import (
+	"io"
+	"net"
+	"sync/atomic"
+)
+
+// InMemoryListener is intended to be used with testing/synctest
+type InMemoryListener struct {
+	closed   chan struct{}
+	connCh   chan net.Conn
+	isClosed atomic.Bool
+}
+
+func NewInMemoryListener() *InMemoryListener {
+	return &InMemoryListener{
+		connCh: make(chan net.Conn),
+		closed: make(chan struct{}),
+	}
+}
+
+func (l *InMemoryListener) ServeConn(conn net.Conn) error {
+	if l.isClosed.Load() {
+		return net.ErrClosed
+	}
+
+	l.connCh <- conn
+	return nil
+}
+
+func (l *InMemoryListener) Accept() (net.Conn, error) {
+	select {
+	case c := <-l.connCh:
+		return c, nil
+	case <-l.closed:
+		return nil, io.EOF
+	}
+}
+
+func (l *InMemoryListener) Close() error {
+	if !l.isClosed.CompareAndSwap(false, true) {
+		return nil
+	}
+	close(l.closed)
+	return nil
+}
+
+func (l *InMemoryListener) Addr() net.Addr {
+	return memAddr("memory-listener")
+}
+
+type memAddr string
+
+func (a memAddr) Network() string { return string(a) }
+func (a memAddr) String() string  { return string(a) }

daemon/listener_test.go

@@ -0,0 +1,115 @@
+package daemon_test
+
+import (
+	"context"
+	"fmt"
+	"github.com/kapetan-io/querator/daemon"
+	pb "github.com/kapetan-io/querator/proto"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestInMemoryListener(t *testing.T) {
+	listener := daemon.NewInMemoryListener()
+	server := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			_, _ = fmt.Fprintf(w, "Hello, %s!", r.URL.Path[1:])
+		}),
+	}
+	go func() { _ = server.Serve(listener) }()
+	defer func() { _ = server.Close() }()
+
+	clientCount := 3
+	var wg sync.WaitGroup
+	for i := 0; i < clientCount; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			// Each client gets its own net.Pipe
+			serverConn, clientConn := net.Pipe()
+			_ = listener.ServeConn(serverConn)
+
+			// Custom DialContext returns the clientConn for this request
+			dialOnce := sync.Once{}
+
+			client := &http.Client{
+				Transport: &http.Transport{
+					DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+						var c net.Conn
+						dialOnce.Do(func() { c = clientConn })
+						return c, nil
+					},
+				},
+				Timeout: 2 * time.Second,
+			}
+
+			url := fmt.Sprintf("http://inmemory/client%d", id)
+			resp, err := client.Get(url)
+			if err != nil {
+				t.Errorf("client %d error: %v", id, err)
+				return
+			}
+			defer func() { _ = resp.Body.Close() }()
+			body, _ := io.ReadAll(resp.Body)
+			expected := fmt.Sprintf("Hello, client%d!", id)
+			if !strings.Contains(string(body), expected) {
+				t.Errorf("client %d got unexpected body: %q", id, body)
+			}
+		}(i)
+	}
+	wg.Wait()
+	_ = listener.Close()
+}
+
+func TestDaemonInMemoryListener(t *testing.T) {
+	ctx := context.Background()
+
+	// Create daemon with InMemoryListener enabled
+	d, err := daemon.NewDaemon(ctx, daemon.Config{
+		InMemoryListener: true,
+	})
+	require.NoError(t, err)
+	defer func() { _ = d.Shutdown(ctx) }()
+
+	// Get a client - should create a new net.Pipe connection
+	{
+		client, err := d.Client()
+		require.NoError(t, err)
+
+		// Test QueuesList to verify the connection works
+		var resp pb.QueuesListResponse
+		err = client.QueuesList(ctx, &resp, nil)
+		require.NoError(t, err)
+
+		// Verify we got an empty list
+		assert.Nil(t, resp.Items)
+
+		// Should work a second time also
+		err = client.QueuesList(ctx, &resp, nil)
+		require.NoError(t, err)
+
+		// Verify we got an empty list
+		assert.Nil(t, resp.Items)
+	}
+
+	// Test getting multiple clients - each should work independently
+	{
+		client, err := d.Client()
+		require.NoError(t, err)
+
+		// Test QueuesList to verify the connection works
+		var resp pb.QueuesListResponse
+		err = client.QueuesList(ctx, &resp, nil)
+		require.NoError(t, err)
+
+		// Verify we got an empty list
+		assert.Nil(t, resp.Items)
+	}
+}