Breaking: indicate root struct when establishing gRPC connection (#62)

Resolves #58

This will allow the same gRPC destination to accept STEF data
that uses multiple different schemas (one schema per connection).
A known use cases is Otel where we want to send metrics, traces,
logs, etc. This should go to the same gRPC endpoint for
simplicity but the destination needs to know what schema to use
for decoding.

The root struct is indicated in the first message from gRPC
sender to destination, before the STEF stream bytes are sent.

Author: tigrannajaryan

go/grpc/client.go

@@ -24,11 +24,12 @@ type ClientCallbacks struct {
 	OnAck func(ackId uint64) error
 }
 
+// Client is a client for communicating over STEF/gRPC protocol.
 type Client struct {
 	grpcClient   stef_proto.STEFDestinationClient
 	stream       stef_proto.STEFDestination_StreamClient
 	callbacks    ClientCallbacks
-	clientSchema *schema.WireSchema
+	clientSchema ClientSchema
 	logger       types.Logger
 
 	// Running state
@@ -84,15 +85,47 @@ func (w *grpcWriter) WriteChunk(header []byte, content []byte) error {
 
 var ErrServerInvalidResponse = errors.New("invalid server response")
 
+// ClientSettings contains configuration settings for creating a Client.
 type ClientSettings struct {
+	// Logger instance used for logging client operations.
 	Logger types.Logger
+
 	// gRPC stream to send data over.
-	GrpcClient   stef_proto.STEFDestinationClient
-	ClientSchema *schema.WireSchema
-	Callbacks    ClientCallbacks
+	GrpcClient stef_proto.STEFDestinationClient
+
+	// ClientSchema of the client.
+	ClientSchema ClientSchema
+
+	// Callbacks for handling events such as acknowledgments and disconnections.
+	Callbacks ClientCallbacks
 }
 
-func NewClient(settings ClientSettings) *Client {
+type ClientSchema struct {
+	// The name of the root struct of the schema.
+	RootStructName string
+
+	// The wire schema of the client.
+	WireSchema *schema.WireSchema
+}
+
+// NewClient creates a new instance of the Client.
+//
+// Requirements:
+// - The `RootStructName` in `ClientSchema` must not be empty.
+// - The `WireSchema` in `ClientSchema` must not be nil.
+//
+// Example:
+//
+//	clientSettings := stefgrpc.ClientSettings{
+//	    GrpcClient:   grpcClient,
+//	    ClientSchema: stefgrpc.ClientSchema{RootStructName: "Metrics", WireSchema: &schema},
+//	    Callbacks:    stefgrpc.ClientCallbacks{},
+//	}
+//	client, err := stefgrpc.NewClient(clientSettings)
+//	if err != nil {
+//	    log.Fatalf("Failed to create client: %v", err)
+//	}
+func NewClient(settings ClientSettings) (*Client, error) {
 	if settings.Logger == nil {
 		settings.Logger = internal.NopLogger{}
 	}
@@ -101,6 +134,13 @@ func NewClient(settings ClientSettings) *Client {
 		settings.Callbacks.OnDisconnect = func(err error) {}
 	}
 
+	if settings.ClientSchema.RootStructName == "" {
+		return nil, fmt.Errorf("client schema root struct name is empty")
+	}
+	if settings.ClientSchema.WireSchema == nil {
+		return nil, fmt.Errorf("client schema wire schema is nil")
+	}
+
 	client := &Client{
 		grpcClient:   settings.GrpcClient,
 		callbacks:    settings.Callbacks,
@@ -109,7 +149,7 @@ func NewClient(settings ClientSettings) *Client {
 		waitCh:       make(chan struct{}),
 	}
 
-	return client
+	return client, nil
 }
 
 func (c *Client) Connect(ctx context.Context) (pkg.ChunkWriter, pkg.WriterOptions, error) {
@@ -137,6 +177,19 @@ func (c *Client) Connect(ctx context.Context) (pkg.ChunkWriter, pkg.WriterOption
 	}
 	defer closeOnErr()
 
+	// Send the first message to the server, include the root struct name.
+	clientMsg := &stef_proto.STEFClientFirstMessage{
+		RootStructName: c.clientSchema.RootStructName,
+	}
+	err = stream.Send(
+		&stef_proto.STEFClientMessage{
+			FirstMessage: clientMsg,
+		},
+	)
+	if err != nil {
+		return nil, opts, fmt.Errorf("failed to send to server: %w", err)
+	}
+
 	// The server must send capabilities message.
 	message, err := stream.Recv()
 	if err != nil {
@@ -162,7 +215,7 @@ func (c *Client) Connect(ctx context.Context) (pkg.ChunkWriter, pkg.WriterOption
 	}
 
 	// Check if server schema is backward compatible with client schema.
-	compatibility, err := serverSchema.Compatible(c.clientSchema)
+	compatibility, err := serverSchema.Compatible(c.clientSchema.WireSchema)
 	switch compatibility {
 	case schema.CompatibilityExact:
 		// Schemas match exactly, nothing else is needed, can start sending data.
@@ -171,12 +224,12 @@ func (c *Client) Connect(ctx context.Context) (pkg.ChunkWriter, pkg.WriterOption
 		// ServerStream schema is superset of client schema. The client MUST specify its schema
 		// in the STEF header.
 		opts.IncludeDescriptor = true
-		opts.Schema = c.clientSchema
+		opts.Schema = c.clientSchema.WireSchema
 
 	case schema.CompatibilityIncompatible:
 		// It is neither exact match nor is server schema a superset, but server schema maybe subset.
 		// Check the opposite direction: if client schema is backward compatible with server schema.
-		compatibility, err = serverSchema.Compatible(c.clientSchema)
+		compatibility, err = serverSchema.Compatible(c.clientSchema.WireSchema)
 
 		if err != nil || compatibility == schema.CompatibilityIncompatible {
 			return nil, opts, fmt.Errorf("client and server schemas are incompatble: %w", err)
@@ -185,7 +238,7 @@ func (c *Client) Connect(ctx context.Context) (pkg.ChunkWriter, pkg.WriterOption
 		if compatibility == schema.CompatibilitySuperset {
 			// Client schema is superset of server schema. The client MUST downgrade its schema.
 			opts.IncludeDescriptor = true
-			opts.Schema = c.clientSchema
+			opts.Schema = c.clientSchema.WireSchema
 		}
 	}
 

go/grpc/proto/destination.proto

@@ -16,19 +16,38 @@ service STEFDestination {
 }
 
 message STEFClientMessage {
+  // The client MUST set first_message field in the first STEFClientMessage sent.
+  // All other fields MUST be unset when first_message is set.
+  // All subsequent messages MUST have first_message unset.
+  STEFClientFirstMessage first_message = 1;
+
   // The bytes of STEF stream. The recipient is responsible for assembling the
   // STEF data stream from a sequence of messages in the order the
   // messages are received and decoding the STEF data stream.
   //
   // See specification.md for specification of STEF stream.
-  bytes stef_bytes = 1;
+  bytes stef_bytes = 2;
 
   // Indicates that the last byte of tef_bytes is also an end of a chunk (a STEF header or
   // STEF frame). This can be used by recipients to accumulates bytes until the end of
   // the chunk is encountered and only then start decoding the chunk.
   // Clients MUST ensure they mark this field true at least once in a while otherwise
   // recipients may never start decoding the data.
-  bool is_end_of_chunk = 2;
+  bool is_end_of_chunk = 3;
+}
+
+// ClientFirstMessage is the first message sent by the client to the destination.
+// The client MUST send this message first. The destination MUST respond with
+// STEFDestinationCapabilities message. The client MUST NOT send any other
+// messages until it receives STEFDestinationCapabilities message from the
+// destination. The client MUST NOT send STEF data until it receives
+// STEFDestinationCapabilities message from the destination.
+message STEFClientFirstMessage {
+  // The name of the root struct of the client's schema. This is useful
+  // for destinations that accept multiple schemas and need to know which schema
+  // the client is using. The destination will use this information to
+  // determine the schema to use for decoding the data.
+  string root_struct_name = 1;
 }
 
 message STEFDestinationCapabilities {
@@ -70,6 +89,8 @@ message STEFDictionaryLimits {
 
 
 message STEFServerMessage {
+  // TODO: refactor this to avoid using oneof message to reduce
+  // allocations for the most common case of STEFDataResponse.
   oneof message {
     STEFDestinationCapabilities capabilities = 1;
     STEFDataResponse response = 2;

go/grpc/server.go

@@ -181,13 +181,27 @@ func NewStreamServer(settings ServerSettings) *StreamServer {
 }
 
 func (s *StreamServer) Stream(server stef_proto.STEFDestination_StreamServer) error {
+	// Receive the first message from the client.
+	clientMsg, err := server.Recv()
+	if err != nil {
+		return fmt.Errorf("failed to receive a message from the client: %w", err)
+	}
+	if clientMsg.FirstMessage == nil {
+		return fmt.Errorf("FirstMessage is nil")
+	}
+	if clientMsg.FirstMessage.RootStructName == "" {
+		return fmt.Errorf("RootStructName is unspecified")
+	}
+
+	// Send capabilities message to the client.
+
+	// Prepare the schema bytes.
 	var schemaBytes bytes.Buffer
-	err := s.serverSchema.Serialize(&schemaBytes)
+	err = s.serverSchema.Serialize(&schemaBytes)
 	if err != nil {
 		return fmt.Errorf("could not marshal server schema: %w", err)
 	}
 
-	// Send capabilities message to the client.
 	message := stef_proto.STEFServerMessage{
 		Message: &stef_proto.STEFServerMessage_Capabilities{
 			Capabilities: &stef_proto.STEFDestinationCapabilities{