CNS-340 - add ability to publish to specific subject hierarchies

bahamut.go

@@ -101,6 +101,10 @@ func New(options ...Option) Server {
 		zap.L().Warn("Push server is enabled but neither dispatching or publishing is. Use bahamut.OptPushPublishHandler() and/or bahamut.OptPushDispatchHandler()")
 	}
 
+	if !c.pushServer.enabled && c.pushServer.subjectHierarchiesEnabled {
+		zap.L().Warn("Push server subject hierarchies have been enabled, but no push server has been configured. Use bahamut.OptPushServer to configure a Push server.")
+	}
+
 	if (c.restServer.enabled || c.pushServer.enabled) && len(c.model.modelManagers) == 0 {
 		zap.L().Warn("No elemental.ModelManager is defined. Use bahamut.OptModel()")
 	}

config.go

@@ -70,14 +70,15 @@ type config struct {
 	}
 
 	pushServer struct {
-		service         PubSubClient
-		topic           string
-		endpoint        string
-		dispatchHandler PushDispatchHandler
-		publishHandler  PushPublishHandler
-		enabled         bool
-		publishEnabled  bool
-		dispatchEnabled bool
+		service                   PubSubClient
+		topic                     string
+		endpoint                  string
+		dispatchHandler           PushDispatchHandler
+		publishHandler            PushPublishHandler
+		enabled                   bool
+		subjectHierarchiesEnabled bool
+		publishEnabled            bool
+		dispatchEnabled           bool
 	}
 
 	healthServer struct {

options.go

@@ -161,6 +161,23 @@ func OptPushServer(service PubSubClient, topic string) Option {
 	}
 }
 
+// OptPushServerEnableSubjectHierarchies will cause the push server to push to specific subject hierarchies under the configured
+// pub/sub topic you have chosen for your push server. This option has no effect if OptPushServer is not set.
+//
+// For example:
+//
+//   If the push server topic has been set to "global-events" and the server is about to push a "create" event w/ an identity
+//   value of "apples", enabling this option, would cause the push server to target a new publication to the subject
+//   "global-events.apples.create", INSTEAD OF "global-events". Consequently, as a result of this, any upstream push
+//   servers that are interested in receiving all events you publish to this topic would need to utilize subject wildcards.
+//
+//   See: https://docs.nats.io/nats-concepts/subjects#wildcards for more details.
+func OptPushServerEnableSubjectHierarchies() Option {
+	return func(c *config) {
+		c.pushServer.subjectHierarchiesEnabled = true
+	}
+}
+
 // OptPushEndpoint sets the endpoint to use for websocket channel.
 //
 // If unset, it fallsback to the default which is /events. This option

options_test.go

@@ -105,6 +105,11 @@ func TestBahamut_Options(t *testing.T) {
 		So(c.pushServer.publishHandler, ShouldEqual, h)
 	})
 
+	Convey("Calling OptPushServerEnableSubjectHierarchies should work", t, func() {
+		OptPushServerEnableSubjectHierarchies()(&c)
+		So(c.pushServer.subjectHierarchiesEnabled, ShouldEqual, true)
+	})
+
 	Convey("Calling OptHealthServer should work", t, func() {
 		h := func() error { return nil }
 		OptHealthServer("1.2.3.4:123", h)(&c)

websocket_server.go

@@ -172,7 +172,27 @@ func (n *pushServer) pushEvents(events ...*elemental.Event) {
 			}
 		}
 
-		publication := NewPublication(n.cfg.pushServer.topic)
+		// if subject hierarchies are enabled, for the benefit of subscribers interested in specific identities, we utilize
+		// a subject hierarchy here to publish to a specific subject under the configured topic of the push server.
+		//
+		// for example:
+		//
+		//   if the push server topic has been set to "global-events" and the server is about to push a "create" event w/ an identity
+		//   value of "apples", enabling this option, would cause the push server to target a new publication to the subject
+		//   "global-events.apples.create", INSTEAD OF "global-events".
+		//
+		//   consequently, clients interested in receiving events pertaining to the "apples" resource can then subscribe
+		//   on that specific topic, as opposed to ignoring events they don't care about. For clients interested in receiving
+		//   ALL events published to "global-events", they can utilize NATS wildcards and subscribe to "global-events.>"
+		//   ('>' targets all hierarchies) or "global-events.*" ('*' matching a single token).
+		//
+		// more details: https://docs.nats.io/nats-concepts/subjects#subject-hierarchies
+		topic := n.cfg.pushServer.topic
+		if n.cfg.pushServer.subjectHierarchiesEnabled {
+			topic = fmt.Sprintf("%s.%s.%s", topic, event.Identity, event.Type)
+		}
+
+		publication := NewPublication(topic)
 		if err = publication.Encode(event); err != nil {
 			zap.L().Error("Unable to encode event", zap.Error(err))
 			break
@@ -258,7 +278,18 @@ func (n *pushServer) start(ctx context.Context) {
 
 	if n.cfg.pushServer.service != nil {
 		errors := make(chan error, 24000)
-		defer n.cfg.pushServer.service.Subscribe(n.publications, errors, n.cfg.pushServer.topic)()
+		subTopic := n.cfg.pushServer.topic
+		// backwards compatibility: if the push server is using subject hierarchies when publishing events, we must by default
+		// listen to all child subjects of the configured topic via a wildcard '>'.
+		//
+		// see: https://docs.nats.io/nats-concepts/subjects#wildcards for more details.
+		//
+		// TODO: in the future, support to subscribing to specific subjects and/or wildcards may be added.
+		if n.cfg.pushServer.subjectHierarchiesEnabled {
+			subTopic = fmt.Sprintf("%s.>", subTopic)
+		}
+
+		defer n.cfg.pushServer.service.Subscribe(n.publications, errors, subTopic)()
 	}
 
 	zap.L().Debug("Websocket server started",

websocket_server_test.go

@@ -518,6 +518,41 @@ func TestWebsocketServer_pushEvents(t *testing.T) {
 				So(len(srv.publications), ShouldEqual, 0)
 			})
 		})
+
+		Convey("When I call pushEvents on a server w/ NATS subject hierarchies enabled", func() {
+
+			srv := &mockPubSubServer{}
+			h := &mockSessionHandler{}
+			h.shouldPublishOK = true
+
+			cfg := config{}
+			cfg.pushServer.service = srv
+			cfg.pushServer.enabled = true
+			cfg.pushServer.publishEnabled = true
+			cfg.pushServer.dispatchEnabled = true
+			cfg.pushServer.subjectHierarchiesEnabled = true
+			cfg.pushServer.topic = "events"
+			cfg.pushServer.publishHandler = h
+
+			wss := newPushServer(cfg, mux, pf)
+			testEvent := elemental.NewEvent(elemental.EventCreate, testmodel.NewList())
+			wss.pushEvents(testEvent)
+
+			Convey("Then I should find one publication sent to the correct topic", func() {
+
+				eventOut := elemental.NewEvent(elemental.EventCreate, testmodel.NewList())
+				eventOut.Timestamp = testEvent.Timestamp
+				r, err := elemental.Encode(elemental.EncodingTypeMSGPACK, eventOut)
+				So(err, ShouldBeNil)
+				So(len(srv.publications), ShouldEqual, 1)
+				pub := srv.publications[0]
+				So(string(pub.Data), ShouldResemble, string(r))
+				So(pub.Topic, ShouldEqual, fmt.Sprintf("%s.%s.%s",
+					cfg.pushServer.topic,
+					testmodel.ListIdentity.Name,
+					elemental.EventCreate))
+			})
+		})
 	})
 }
 
@@ -542,6 +577,7 @@ func TestWebsocketServer_start(t *testing.T) {
 		cfg.pushServer.enabled = true
 		cfg.pushServer.publishEnabled = true
 		cfg.pushServer.dispatchEnabled = true
+		cfg.pushServer.subjectHierarchiesEnabled = true
 		cfg.pushServer.dispatchHandler = pushHandler
 
 		wss := newPushServer(cfg, mux, pf)