new: Add SendBroadcastMessage() bus method

bus/eventbus.go

@@ -22,6 +22,7 @@ type EventBus interface {
 	GetChannelManager() ChannelManager
 	SendRequestMessage(channelName string, payload interface{}, destinationId *uuid.UUID) error
 	SendResponseMessage(channelName string, payload interface{}, destinationId *uuid.UUID) error
+	SendBroadcastMessage(channelName string, payload interface{}) error
 	SendErrorMessage(channelName string, err error, destinationId *uuid.UUID) error
 	ListenStream(channelName string) (MessageHandler, error)
 	ListenStreamForDestination(channelName string, destinationId *uuid.UUID) (MessageHandler, error)
@@ -186,6 +187,19 @@ func (bus *transportEventBus) SendResponseMessage(channelName string, payload in
 	return nil
 }
 
+// SendBroadcastMessage sends the payload as an outbound broadcast message to channelName. Since it is a broadcast,
+// the payload does not require a destination ID. Throws an error if the channel does not exist.
+func (bus *transportEventBus) SendBroadcastMessage(channelName string, payload interface{}) error {
+	channelObject, err := bus.ChannelManager.GetChannel(channelName)
+	if err != nil {
+		return err
+	}
+	config := buildConfig(channelName, payload, nil)
+	message := model.GenerateResponse(config)
+	sendMessageToChannel(channelObject, message)
+	return nil
+}
+
 // SendRequestMessage Send a RequestDir type message (outbound) message on Channel, with supplied Payload.
 // Throws error if the Channel does not exist.
 func (bus *transportEventBus) SendRequestMessage(channelName string, payload interface{}, destId *uuid.UUID) error {

bus/eventbus_test.go

@@ -87,6 +87,11 @@ func TestEventBus_SendRequestMessageNoChannel(t *testing.T) {
 	assert.NotNil(t, err)
 }
 
+func TestTransportEventBus_SendBroadcastMessageNoChannel(t *testing.T) {
+	err := evtBusTest.SendBroadcastMessage("Channel-not-here", "hello melody")
+	assert.NotNil(t, err)
+}
+
 func TestEventBus_ListenStream(t *testing.T) {
 	createTestChannel()
 	handler, err := evtBusTest.ListenStream(evtbusTestChannelName)
@@ -112,6 +117,30 @@ func TestEventBus_ListenStream(t *testing.T) {
 	destroyTestChannel()
 }
 
+func TestTransportEventBus_ListenStreamForBroadcast(t *testing.T) {
+	createTestChannel()
+	handler, err := evtBusTest.ListenStream(evtbusTestChannelName)
+	assert.Nil(t, err)
+	assert.NotNil(t, handler)
+	var count int32 = 0
+	handler.Handle(
+		func(msg *model.Message) {
+			assert.Equal(t, "hello melody", msg.Payload.(string))
+			inc(&count)
+		},
+		func(err error) {})
+
+	for i := 0; i < 3; i++ {
+		evtBusTest.SendBroadcastMessage(evtbusTestChannelName, "hello melody")
+
+		// send requests to make sure we're only getting requests
+		evtBusTest.SendRequestMessage(evtbusTestChannelName, 1, nil)
+	}
+	evtbusTestManager.WaitForChannel(evtbusTestChannelName)
+	assert.Equal(t, int32(3), count)
+	destroyTestChannel()
+}
+
 func TestTransportEventBus_ListenStreamForDestination(t *testing.T) {
 	createTestChannel()
 	id := uuid.New()
@@ -596,11 +625,11 @@ func TestChannelManager_TestConnectBroker(t *testing.T) {
 
 	// connect to broker
 	cf := &bridge.BrokerConnectorConfig{
-		Username:   "test",
-		Password:   "test",
-		UseWS:      true,
+		Username: "test",
+		Password: "test",
+		UseWS:    true,
 		WebSocketConfig: &bridge.WebSocketConfig{
-			WSPath:     "/",
+			WSPath: "/",
 		},
 		ServerAddr: "broker-url"}