Merge pull request #370 from streamdal/dselans/relay-name-update

include subscription id in the relay name
streamdal · Mar 26, 2024 · e4b9db4 · e4b9db4
2 parents 3f7a227 + 2819561
commit e4b9db4
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/backends/gcppubsub/relay.go b/backends/gcppubsub/relay.go
@@ -58,10 +58,20 @@ func (g *GCPPubSub) Relay(ctx context.Context, relayOpts *opts.RelayOptions, rel
 		if sc != nil {
 			g.log.Debug("Processing message via streamdal SDK")
 
+			operationName := "relay"
+
+			if relayOpts != nil && relayOpts.GcpPubsub != nil && relayOpts.GcpPubsub.GetArgs() != nil {
+				if relayOpts.GcpPubsub.GetArgs().SubscriptionId == "" {
+					operationName = "relay-unknown-subid"
+				} else {
+					operationName = "relay-" + relayOpts.GcpPubsub.GetArgs().SubscriptionId
+				}
+			}
+
 			resp := sc.Process(ctx, &sdk.ProcessRequest{
 				ComponentName: "gcp-pubsub",
 				OperationType: sdk.OperationTypeConsumer,
-				OperationName: "relay",
+				OperationName: operationName,
 				Data:          msg.Data,
 			})