feat(server/memory,client/orb_transport/memory): Implement streaming, closes #48

Author: jochumdev

client/orb/memory_transport.go

@@ -0,0 +1,3 @@
+module github.com/go-orb/plugins/client/orb_transport/memory
+
+go 1.23.6

client/orb_transport/memory/go.mod

@@ -0,0 +1,111 @@
+// Package memory implements a go-orb/plugins/client/orb compatible memory transport.
+package memory
+
+import (
+	"context"
+	"maps"
+
+	"github.com/go-orb/go-orb/client"
+	"github.com/go-orb/go-orb/log"
+	"github.com/go-orb/go-orb/util/metadata"
+	"github.com/go-orb/go-orb/util/orberrors"
+
+	"github.com/go-orb/plugins/client/orb"
+)
+
+// Name is the name of this transport.
+const Name = "memory"
+
+func init() {
+	orb.RegisterTransport(Name, NewTransport)
+}
+
+// Transport is a go-orb/plugins/client/orb compatible memory transport.
+type Transport struct {
+	logger log.Logger
+}
+
+// Start starts the transport.
+func (t *Transport) Start() error {
+	return nil
+}
+
+// Stop stop the transport.
+func (t *Transport) Stop(_ context.Context) error {
+	return nil
+}
+
+// Name returns the name of this transport.
+func (t *Transport) Name() string {
+	return Name
+}
+
+// Request does the actual rpc request to the server.
+func (t *Transport) Request(ctx context.Context, infos client.RequestInfos, req any, result any, opts *client.CallOptions) error {
+	server, err := client.ResolveMemoryServer(infos.Service)
+	if err != nil {
+		return orberrors.ErrInternalServerError.Wrap(err)
+	}
+
+	md := opts.Metadata
+	if md == nil {
+		md = map[string]string{}
+	}
+
+	ctx, outMd := metadata.WithOutgoing(ctx)
+	ctx, inMd := metadata.WithIncoming(ctx)
+
+	maps.Copy(inMd, md)
+	inMd["Content-Type"] = opts.ContentType
+
+	err = server.Request(ctx, infos, req, result, opts)
+
+	// Retrieve metadata from memory.
+	if opts.ResponseMetadata != nil {
+		for k, v := range outMd {
+			opts.ResponseMetadata[k] = v
+		}
+	}
+
+	return err
+}
+
+// Stream opens a bidirectional stream to the memory server.
+func (t *Transport) Stream(
+	ctx context.Context, infos client.RequestInfos,
+	opts *client.CallOptions,
+) (client.StreamIface[any, any], error) {
+	server, err := client.ResolveMemoryServer(infos.Service)
+	if err != nil {
+		return nil, orberrors.ErrInternalServerError.Wrap(err)
+	}
+
+	md := opts.Metadata
+	if md == nil {
+		md = map[string]string{}
+	}
+
+	ctx, outMd := metadata.WithOutgoing(ctx)
+	ctx, inMd := metadata.WithIncoming(ctx)
+
+	maps.Copy(inMd, md)
+	inMd["Content-Type"] = opts.ContentType
+
+	stream, err := server.Stream(ctx, infos, opts)
+
+	// Retrieve metadata from memory.
+	if opts.ResponseMetadata != nil {
+		for k, v := range outMd {
+			opts.ResponseMetadata[k] = v
+		}
+	}
+
+	return stream, err
+}
+
+// NewTransport creates a Transport.
+func NewTransport(logger log.Logger, _ *orb.Config) (orb.TransportType, error) {
+	return orb.TransportType{Transport: &Transport{
+		logger: logger,
+	}}, nil
+}

client/orb_transport/memory/memory.go

@@ -0,0 +1,89 @@
+package memory
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/go-orb/go-orb/log"
+	"github.com/go-orb/go-orb/registry"
+	"github.com/go-orb/go-orb/server"
+	"github.com/go-orb/go-orb/types"
+	"github.com/go-orb/plugins/client/tests"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/go-orb/plugins/server/memory"
+
+	echohandler "github.com/go-orb/plugins/client/tests/handler/echo"
+	echoproto "github.com/go-orb/plugins/client/tests/proto/echo"
+
+	filehandler "github.com/go-orb/plugins/client/tests/handler/file"
+	fileproto "github.com/go-orb/plugins/client/tests/proto/file"
+
+	// Blank imports here are fine.
+	_ "github.com/go-orb/plugins/codecs/json"
+	_ "github.com/go-orb/plugins/codecs/proto"
+	_ "github.com/go-orb/plugins/codecs/yaml"
+	_ "github.com/go-orb/plugins/log/slog"
+	_ "github.com/go-orb/plugins/registry/mdns"
+)
+
+func setupServer(sn string) (*tests.SetupData, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	setupData := &tests.SetupData{}
+
+	logger, err := log.New(log.WithLevel(log.LevelTrace))
+	if err != nil {
+		cancel()
+
+		return nil, err
+	}
+
+	reg, err := registry.New(nil, &types.Components{}, logger)
+	if err != nil {
+		cancel()
+
+		return nil, err
+	}
+
+	echoHInstance := new(echohandler.Handler)
+	echoHRegister := echoproto.RegisterStreamsHandler(echoHInstance)
+
+	fileHInstance := new(filehandler.Handler)
+	fileHRegister := fileproto.RegisterFileServiceHandler(fileHInstance)
+
+	options := []server.Option{
+		memory.WithHandlers(echoHRegister, fileHRegister),
+	}
+
+	ep, err := memory.New(sn, "", memory.NewConfig(options...), logger, reg)
+	if err != nil {
+		cancel()
+
+		return nil, err
+	}
+
+	setupData.Logger = logger
+	setupData.Registry = reg
+	setupData.Entrypoints = []server.Entrypoint{ep}
+	setupData.Ctx = ctx
+	setupData.Stop = cancel
+
+	return setupData, nil
+}
+
+func newSuite() *tests.TestSuite {
+	s := tests.NewSuite(setupServer, []string{Name})
+	// s.Debug = true
+	return s
+}
+
+func TestSuite(t *testing.T) {
+	if os.Getenv("CI") != "" {
+		t.Skip("Skipping testing in CI environment")
+	}
+
+	// Run the tests.
+	suite.Run(t, newSuite())
+}

client/orb_transport/memory/memory_test.go

@@ -333,7 +333,7 @@ func (s *TestSuite) TestFileUpload() {
 			// Open a stream to the service
 			stream, err := fileClient.UploadFile(ctx, ServiceName, client.WithPreferredTransports(t))
 			if errors.Is(err, orberrors.ErrNotImplemented) {
-				// Transport does not support streaming.
+				s.T().Skip("Transport does not support streaming.")
 				return
 			}
 
@@ -377,6 +377,8 @@ func (s *TestSuite) TestFileUpload() {
 }
 
 // TestAuthorizedFileUpload tests the authorized client streaming functionality for file uploads.
+//
+//nolint:funlen
 func (s *TestSuite) TestAuthorizedFileUpload() {
 	// Create a file service client
 	fileClient := file.NewFileServiceClient(s.client)
@@ -389,7 +391,7 @@ func (s *TestSuite) TestAuthorizedFileUpload() {
 			// Open a stream to the service
 			stream, err := fileClient.AuthorizedUploadFile(ctx, ServiceName, client.WithPreferredTransports(t))
 			if errors.Is(err, orberrors.ErrNotImplemented) {
-				// Transport does not support streaming.
+				s.T().Skip("Transport does not support streaming.")
 				return
 			}
 
@@ -436,7 +438,7 @@ func (s *TestSuite) TestAuthorizedFileUpload() {
 				client.WithResponseMetadata(responseMd),
 			)
 			if errors.Is(err, orberrors.ErrNotImplemented) {
-				// Transport does not support streaming.
+				s.T().Skip("Transport does not support streaming.")
 				return
 			}