[events] Support ephemeral consumers

events/nats_config.go

@@ -55,7 +55,7 @@ type NATSConfig struct {
 
 // Configured checks whether the provider has been configured.
 func (c NATSConfig) Configured() bool {
-	return c.URL != "" || c.QueueGroup != ""
+	return c.URL != ""
 }
 
 // Validate ensures the configuration is valid.

events/nats_connection.go

@@ -121,6 +121,10 @@ func NewNATSConnection(config NATSConfig, options ...NATSOption) (*NATSConnectio
 		}
 	}
 
+	if config.QueueGroup == "" {
+		nc.logger.Warn("NATS QueueGroup is not set. Subscriptions will not be durable.")
+	}
+
 	conn, err := nats.Connect(config.URL, nc.connectOptions...)
 	if err != nil {
 		return nil, err
@@ -143,7 +147,12 @@ func NewNATSConnection(config NATSConfig, options ...NATSOption) (*NATSConnectio
 }
 
 // NATSConsumerDurableName is the generator function to create a new durable consumer name.
+// If queueGroup is empty, an empty durable name is returned to support ephemeral consumers.
 func NATSConsumerDurableName(queueGroup, subject string) string {
+	if queueGroup == "" {
+		return ""
+	}
+
 	hash := md5.Sum([]byte(subject))
 
 	return queueGroup + hex.EncodeToString(hash[:])

events/nats_test.go

@@ -20,55 +20,76 @@ var errTimeout = errors.New("timeout waiting for event")
 
 func TestNATSPublishAndSubscribe(t *testing.T) {
 	ctx := context.Background()
-	nats, err := eventtools.NewNatsServer()
-	require.NoError(t, err)
 
-	defer nats.Close()
+	testCases := []struct {
+		name       string
+		queueGroup string
+	}{
+		{
+			name:       "with ephemeral consumer",
+			queueGroup: "",
+		},
+		{
+			name:       "with durable consumer",
+			queueGroup: "testing-durable",
+		},
+	}
 
-	conn, err := events.NewNATSConnection(nats.Config.NATS)
-	require.NoError(t, err)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			nats, err := eventtools.NewNatsServer()
+			require.NoError(t, err)
 
-	defer conn.Shutdown(ctx) //nolint:errcheck // within test
+			defer nats.Close()
 
-	change := testCreateChange()
+			natsCfg := nats.Config.NATS
+			natsCfg.QueueGroup = tc.queueGroup
+			conn, err := events.NewNATSConnection(natsCfg)
+			require.NoError(t, err)
 
-	msg, err := conn.PublishChange(ctx, "test", change)
-	require.NoError(t, err)
-	require.Equal(t, change, msg.Message())
+			defer conn.Shutdown(ctx) //nolint:errcheck // within test
 
-	change2 := testCreateChange()
+			change := testCreateChange()
 
-	msg, err = conn.PublishChange(ctx, "test", change2)
-	require.NoError(t, err)
-	require.Equal(t, change2, msg.Message())
+			msg, err := conn.PublishChange(ctx, "test", change)
+			require.NoError(t, err)
+			require.Equal(t, change, msg.Message())
 
-	change3 := testCreateChange()
-	change3.ActorID = ""
+			change2 := testCreateChange()
 
-	msg, err = conn.PublishChange(ctx, "test", change3)
-	require.NoError(t, err)
-	require.NotEqual(t, change3, msg.Message())
+			msg, err = conn.PublishChange(ctx, "test", change2)
+			require.NoError(t, err)
+			require.Equal(t, change2, msg.Message())
 
-	messages, err := conn.SubscribeChanges(ctx, ">")
-	require.NoError(t, err)
+			change3 := testCreateChange()
+			change3.ActorID = ""
 
-	receivedMsg, err := getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.EqualValues(t, change, receivedMsg.Message())
+			msg, err = conn.PublishChange(ctx, "test", change3)
+			require.NoError(t, err)
+			require.NotEqual(t, change3, msg.Message())
 
-	receivedMsg, err = getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.EqualValues(t, change2, receivedMsg.Message())
-	assert.NoError(t, receivedMsg.Ack())
+			messages, err := conn.SubscribeChanges(ctx, ">")
+			require.NoError(t, err)
 
-	receivedMsg, err = getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.NotEqualValues(t, change3, receivedMsg.Message())
-	assert.Equal(t, "unknown-actor", receivedMsg.Message().ActorID.String())
-	assert.NoError(t, receivedMsg.Ack())
+			receivedMsg, err := getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.EqualValues(t, change, receivedMsg.Message())
+
+			receivedMsg, err = getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.EqualValues(t, change2, receivedMsg.Message())
+			assert.NoError(t, receivedMsg.Ack())
+
+			receivedMsg, err = getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.NotEqualValues(t, change3, receivedMsg.Message())
+			assert.Equal(t, "unknown-actor", receivedMsg.Message().ActorID.String())
+			assert.NoError(t, receivedMsg.Ack())
+		})
+	}
 }
 
 func TestNATSRequestReply(t *testing.T) {

testing/eventtools/nats_test.go

@@ -20,14 +20,17 @@ var errTimeout = errors.New("timeout waiting for event")
 func TestNats(t *testing.T) {
 	ctx := context.Background()
 
-	consumerName := events.NATSConsumerDurableName("", eventtools.Prefix+".changes.>")
+	consumerName := events.NATSConsumerDurableName("nats-testing", eventtools.Prefix+".changes.>")
 
 	nats, err := eventtools.NewNatsServer()
 	require.NoError(t, err)
 
 	defer nats.Close()
 
-	conn, err := events.NewNATSConnection(nats.Config.NATS)
+	natsCfg := nats.Config.NATS
+	natsCfg.QueueGroup = "nats-testing"
+
+	conn, err := events.NewNATSConnection(natsCfg)
 	require.NoError(t, err)
 
 	change1 := testCreateChange()