bury overwrite behavior boolean

events/events.go

@@ -41,7 +41,10 @@ type Stream interface {
 	// Publish publishes the message to the message broker.
 	//
 	// rollupSubject when set to true will cause any previous messages with the same subject to be overwritten by this new msg.
-	Publish(ctx context.Context, subject string, msg []byte, rollupSubject bool) error
+	Publish(ctx context.Context, subject string, msg []byte) error
+
+	// PublishOverwrite publishes the message to the message broker overwriting any existing message with that subject
+	PublishOverwrite(ctx context.Context, subject string, msg []byte) error
 
 	// Subscribe subscribes to one or more subjects on the stream returning a message channel for subscribers to read from.
 	Subscribe(ctx context.Context) (MsgCh, error)

events/nats.go

@@ -276,11 +276,18 @@ func (n *NatsJetstream) consumerConfigIsEqual(consumerInfo *nats.ConsumerInfo) b
 
 // Publish publishes an event onto the NATS Jetstream.
 // The caller is responsible for message addressing and data serialization.
-//
+func (n *NatsJetstream) Publish(ctx context.Context, subjectSuffix string, data []byte) error {
+	return n._publish(ctx, subjectSuffix, data, false)
+}
+
+// PublishOverwrite publishes an event and will overwrite any existing message with that subject in the queue
+func (n *NatsJetstream) PublishOverwrite(ctx context.Context, subjectSuffix string, data []byte) error {
+	return n._publish(ctx, subjectSuffix, data, true)
+}
+
 // rollupSubject when set to true will cause any previous messages with the same subject to be overwritten by this new msg.
-//
 // NOTE: The subject passed here will be prepended with the configured PublisherSubjectPrefix.
-func (n *NatsJetstream) Publish(ctx context.Context, subjectSuffix string, data []byte, rollupSubject bool) error {
+func (n *NatsJetstream) _publish(ctx context.Context, subjectSuffix string, data []byte, rollupSubject bool) error {
 	if n.jsctx == nil {
 		return errors.Wrap(ErrNatsJetstreamAddConsumer, "Jetstream context is not setup")
 	}

events/nats_test.go

@@ -83,7 +83,7 @@ func TestPublishAndSubscribe(t *testing.T) {
 	require.NoError(t, err)
 
 	payload := []byte("test data")
-	require.NoError(t, njs.Publish(context.TODO(), "test", payload, false))
+	require.NoError(t, njs.Publish(context.TODO(), "test", payload))
 
 	msgs, err := njs.PullMsg(context.TODO(), 1)
 	require.NoError(t, err)
@@ -129,9 +129,9 @@ func TestPublishAndSubscribe_WithRollup(t *testing.T) {
 	require.NoError(t, err)
 
 	payload := []byte("test data")
-	require.NoError(t, njs.Publish(context.TODO(), "test", payload, true))
+	require.NoError(t, njs.PublishOverwrite(context.TODO(), "test", payload))
 	payload2 := []byte("rollup")
-	require.NoError(t, njs.Publish(context.TODO(), "test", payload2, true))
+	require.NoError(t, njs.PublishOverwrite(context.TODO(), "test", payload2))
 
 	msgs, err := njs.PullMsg(context.TODO(), 1)
 	require.NoError(t, err)