grpc transport v2

Signed-off-by: Wei Liu <[email protected]>

Author: skeeey

pkg/cloudevents/constants/constants.go

@@ -4,3 +4,7 @@ const (
 	ConfigTypeMQTT = "mqtt"
 	ConfigTypeGRPC = "grpc"
 )
+
+// GRPCSubscriptionIDKey is the key for the gRPC subscription ID.
+// This ID is generated by the gRPC server after the client subscribes to it.
+const GRPCSubscriptionIDKey = "subscription-id"

pkg/cloudevents/generic/clients/baseclient.go

@@ -79,7 +79,7 @@ func (c *baseClient) connect(ctx context.Context) error {
 				// TODO enhance the cloudevents SKD to avoid wrapping the error type to distinguish the net connection
 				// errors
 				if err != nil {
-					// failed to reconnect, try agin
+					// failed to reconnect, try again
 					runtime.HandleErrorWithContext(ctx, err, "the cloudevents client reconnect failed")
 					<-wait.RealTimer(DelayFn()).C()
 					continue
@@ -89,14 +89,11 @@ func (c *baseClient) connect(ctx context.Context) error {
 				metrics.IncreaseClientReconnectedCounter(c.clientID)
 				c.setClientReady(true)
 				c.sendReceiverSignal(restartReceiverSignal)
-				c.sendReconnectedSignal()
 			}
 
 			select {
 			case <-ctx.Done():
-				if c.receiverChan != nil {
-					close(c.receiverChan)
-				}
+				c.closeChannels()
 				return
 			case err, ok := <-c.transport.ErrorChan():
 				if !ok {
@@ -164,14 +161,33 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 		return
 	}
 
+	// send subscription request before starting to receive events
+	if err := c.transport.Subscribe(ctx); err != nil {
+		runtime.HandleErrorWithContext(ctx, err, "failed to subscribe")
+		return
+	}
+
 	c.receiverChan = make(chan int)
 
 	// start a go routine to handle cloudevents subscription
 	go func() {
-		receiverCtx, receiverCancel := context.WithCancel(context.TODO())
+		receiverCtx, receiverCancel := context.WithCancel(ctx)
 		startReceiving := true
+		subscribed := true
 
 		for {
+			if !subscribed {
+				// resubscribe before restarting the receiver
+				if err := c.transport.Subscribe(ctx); err != nil {
+					runtime.HandleError(fmt.Errorf("failed to resubscribe, %v", err))
+					<-wait.RealTimer(DelayFn()).C()
+					continue
+				}
+				subscribed = true
+				// notify the client caller to resync the resources
+				c.sendReconnectedSignal(ctx)
+			}
+
 			if startReceiving {
 				go func() {
 					if err := c.transport.Receive(receiverCtx, func(evt cloudevents.Event) {
@@ -191,7 +207,7 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 			case <-ctx.Done():
 				receiverCancel()
 				return
-			case signal, ok := <-c.receiverChan:
+			case signal, ok := <-c.getReceiverChan():
 				if !ok {
 					// receiver channel is closed, stop the receiver
 					receiverCancel()
@@ -202,8 +218,9 @@ func (c *baseClient) subscribe(ctx context.Context, receive receiveFn) {
 				case restartReceiverSignal:
 					logger.V(2).Info("restart the cloudevents receiver")
 					// rebuild the receiver context and restart receiving
-					receiverCtx, receiverCancel = context.WithCancel(context.TODO())
+					receiverCtx, receiverCancel = context.WithCancel(ctx)
 					startReceiving = true
+					subscribed = false
 				case stopReceiverSignal:
 					logger.V(2).Info("stop the cloudevents receiver")
 					receiverCancel()
@@ -224,10 +241,32 @@ func (c *baseClient) sendReceiverSignal(signal int) {
 	}
 }
 
-func (c *baseClient) sendReconnectedSignal() {
+func (c *baseClient) closeChannels() {
+	c.Lock()
+	defer c.Unlock()
+
+	if c.receiverChan != nil {
+		close(c.receiverChan)
+		c.receiverChan = nil
+	}
+	if c.reconnectedChan != nil {
+		close(c.reconnectedChan)
+		c.reconnectedChan = nil
+	}
+}
+
+func (c *baseClient) sendReconnectedSignal(ctx context.Context) {
 	c.RLock()
 	defer c.RUnlock()
-	c.reconnectedChan <- struct{}{}
+	if c.reconnectedChan != nil {
+		select {
+		case c.reconnectedChan <- struct{}{}:
+			// Signal sent successfully
+		default:
+			// No receiver listening on reconnectedChan, that's okay - don't block
+			klog.FromContext(ctx).Info("reconnected signal not sent, no receiver listening")
+		}
+	}
 }
 
 func (c *baseClient) isClientReady() bool {
@@ -241,3 +280,9 @@ func (c *baseClient) setClientReady(ready bool) {
 	defer c.Unlock()
 	c.clientReady = ready
 }
+
+func (c *baseClient) getReceiverChan() chan int {
+	c.RLock()
+	defer c.RUnlock()
+	return c.receiverChan
+}

pkg/cloudevents/generic/options/builder/optionsbuilder.go

@@ -7,6 +7,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	grpcv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/v2/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -57,7 +58,7 @@ func BuildCloudEventsSourceOptions(config any,
 	case *mqtt.MQTTOptions:
 		return mqtt.NewSourceOptions(config, clientId, sourceId), nil
 	case *grpc.GRPCOptions:
-		return grpc.NewSourceOptions(config, sourceId, dataType), nil
+		return grpcv2.NewSourceOptions(config, sourceId, dataType), nil
 	default:
 		return nil, fmt.Errorf("unsupported client configuration type %T", config)
 	}
@@ -70,7 +71,7 @@ func BuildCloudEventsAgentOptions(config any,
 	case *mqtt.MQTTOptions:
 		return mqtt.NewAgentOptions(config, clusterName, clientId), nil
 	case *grpc.GRPCOptions:
-		return grpc.NewAgentOptions(config, clusterName, clientId, dataType), nil
+		return grpcv2.NewAgentOptions(config, clusterName, clientId, dataType), nil
 	default:
 		return nil, fmt.Errorf("unsupported client configuration type %T", config)
 	}

pkg/cloudevents/generic/options/builder/optionsbuilder_test.go

@@ -1,9 +1,8 @@
 package builder
 
 import (
-	"encoding/json"
 	"os"
-	"strings"
+	"reflect"
 	"testing"
 	"time"
 
@@ -31,10 +30,11 @@ url: grpc
 )
 
 type buildingCloudEventsOptionTestCase struct {
-	name            string
-	configType      string
-	configFile      *os.File
-	expectedOptions any
+	name                  string
+	configType            string
+	configFile            *os.File
+	expectedOptions       any
+	expectedTransportType string
 }
 
 func TestBuildCloudEventsSourceOptions(t *testing.T) {
@@ -56,6 +56,7 @@ func TestBuildCloudEventsSourceOptions(t *testing.T) {
 					Timeout:    60 * time.Second,
 				},
 			},
+			expectedTransportType: "*mqtt.mqttSourceTransport",
 		},
 		{
 			name:       "grpc config",
@@ -72,6 +73,7 @@ func TestBuildCloudEventsSourceOptions(t *testing.T) {
 					},
 				},
 			},
+			expectedTransportType: "*grpc.grpcTransport",
 		},
 	}
 
@@ -107,10 +109,9 @@ func assertOptions(t *testing.T, c buildingCloudEventsOptionTestCase) {
 		t.Errorf("unexpected error %v", err)
 	}
 
-	optionsRaw, _ := json.Marshal(options)
-	expectedRaw, _ := json.Marshal(c.expectedOptions)
+	tt := reflect.TypeOf(options.CloudEventsTransport)
 
-	if !strings.Contains(string(optionsRaw), string(expectedRaw)) {
-		t.Errorf("the results %v\n does not contain the original options %v\n", string(optionsRaw), string(expectedRaw))
+	if tt.String() != c.expectedTransportType {
+		t.Errorf("expected %s, but got %s", c.expectedTransportType, tt)
 	}
 }

pkg/cloudevents/generic/options/fake/fakeoptions.go

@@ -43,6 +43,10 @@ func (c *EventChan) Connect(ctx context.Context) error {
 	return nil
 }
 
+func (c *EventChan) Subscribe(ctx context.Context) error {
+	return nil
+}
+
 func (c *EventChan) Send(ctx context.Context, evt cloudevents.Event) error {
 	select {
 	case c.evtChan <- evt:

pkg/cloudevents/generic/options/grpc/agentoptions.go

@@ -20,6 +20,7 @@ type grpcAgentTransport struct {
 	dataType          types.CloudEventsDataType
 }
 
+// Deprecated: use v2.grpc.NewAgentOptions instead
 func NewAgentOptions(grpcOptions *GRPCOptions,
 	clusterName, agentID string, dataType types.CloudEventsDataType) *options.CloudEventsAgentOptions {
 	return &options.CloudEventsAgentOptions{
@@ -82,6 +83,12 @@ func (o *grpcAgentTransport) Send(ctx context.Context, evt cloudevents.Event) er
 	return nil
 }
 
+func (o *grpcAgentTransport) Subscribe(ctx context.Context) error {
+	// Subscription is handled by the cloudevents client during receiver startup.
+	// No action needed here.
+	return nil
+}
+
 func (o *grpcAgentTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
 	return o.cloudEventsClient.StartReceiver(ctx, fn)
 }

pkg/cloudevents/generic/options/grpc/sourceoptions.go

@@ -78,6 +78,12 @@ func (o *gRPCSourceTransport) Send(ctx context.Context, evt cloudevents.Event) e
 	return nil
 }
 
+func (o *gRPCSourceTransport) Subscribe(ctx context.Context) error {
+	// Subscription is handled by the cloudevents client during receiver startup.
+	// No action needed here.
+	return nil
+}
+
 func (o *gRPCSourceTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
 	return o.cloudEventsClient.StartReceiver(ctx, fn)
 }

pkg/cloudevents/generic/options/mqtt/agentoptions.go

@@ -138,6 +138,14 @@ func (o *mqttAgentTransport) Send(ctx context.Context, evt cloudevents.Event) er
 	return nil
 }
 
+func (o *mqttAgentTransport) Subscribe(ctx context.Context) error {
+	// Subscription is handled by the cloudevents client during receiver startup.
+	// No action needed here.
+	// TODO: consider implementing native subscription logic in v2 to decouple from
+	// the CloudEvents SDK.
+	return nil
+}
+
 func (o *mqttAgentTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
 	return o.cloudEventsClient.StartReceiver(ctx, fn)
 }

pkg/cloudevents/generic/options/mqtt/sourceoptions.go

@@ -132,6 +132,14 @@ func (o *mqttSourceTransport) Send(ctx context.Context, evt cloudevents.Event) e
 	return nil
 }
 
+func (o *mqttSourceTransport) Subscribe(ctx context.Context) error {
+	// Subscription is handled by the cloudevents client during receiver startup.
+	// No action needed here.
+	// TODO: consider implementing native subscription logic in v2 to decouple from
+	// the CloudEvents SDK.
+	return nil
+}
+
 func (o *mqttSourceTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
 	return o.cloudEventsClient.StartReceiver(ctx, fn)
 }

pkg/cloudevents/generic/options/options.go

@@ -21,13 +21,17 @@ type CloudEventTransport interface {
 	// Connect establishes a connection to the event transport.
 	// This method should be called before Send or Receive.
 	// Returns an error if the connection cannot be established.
-	// TODO remove the dataType
 	Connect(ctx context.Context) error
 
 	// Send transmits a CloudEvent through the transport.
 	// Returns an error if the event cannot be send.
 	Send(ctx context.Context, evt cloudevents.Event) error
 
+	// Subscribe sends a subscription request to the transport to subscribe topics/services.
+	// This is a non-blocking method that should be called after Connect and before Receive.
+	// Returns an error if the subscription request cannot be sent.
+	Subscribe(ctx context.Context) error
+
 	// Receive starts receiving events and invokes the provided handler for each event.
 	// This is a BLOCKING call that runs an event loop until the context is cancelled.
 	// The handler function is called synchronously for each received event.