Problems/limitations with network busses

[usedbytes]

I want to bridge two busses over the network, giving the illusion of a single bus. This isn't currently possible as it deadlocks (and also you'll end up with an infinite loop of duplicate events).

The simple program below will show the deadlock:

package main

import (
    "fmt"
    "time"

    "github.com/asaskevich/EventBus"
)

func main() {

    networkBusA := EventBus.NewNetworkBus(":2035", "/_net_bus_A")
    networkBusA.Start()

    networkBusB := EventBus.NewNetworkBus(":2030", "/_net_bus_B")
    networkBusB.Start()

    networkBusA.Subscribe("topic-A", func(a int) { fmt.Println("A handler:", a) }, ":2030", "/_net_bus_B")

    networkBusB.Subscribe("topic-A", func(a int) { fmt.Println("B handler:", a) }, ":2035", "/_net_bus_A")

    fmt.Println("Publishing on A...")
    networkBusA.EventBus().Publish("topic-A", 10)
    fmt.Println("Done.")

    time.Sleep(2 * time.Second)

    networkBusA.Stop()
    networkBusB.Stop()
}

This is similar to #25, with a slightly different cause.

• networkBusA.EventBus().Publish("topic-A", 20) causes the networkBusA lock to be taken
• The rpc callback for networkBusB is called to publish the event there.
• This goes through networkBusB's normal Publish() path
  • Which will call the rpc callback for networkBusA, and now we're back trying to take the networkBusA lock again.

Avoiding the deadlock by spawning the Publish() in a goroutine shows the infinite loop (run the same program as above):

diff --git a/client.go b/client.go
index a9e9e69..831431a 100644
--- a/client.go
+++ b/client.go
@@ -116,7 +116,7 @@ type ClientService struct {

 // PushEvent - exported service to listening to remote events
 func (service *ClientService) PushEvent(arg *ClientArg, reply *bool) error {
-   service.client.eventBus.Publish(arg.Topic, arg.Args...)
+   go service.client.eventBus.Publish(arg.Topic, arg.Args...)
    *reply = true
    return nil
 }

I'm looking for any suggestions on how to fix this, or perhaps just a definition of how network busses should work (e.g. if there's multiple network clients subscribed to a bus, should events received over the network be propagated to them?)

Just avoiding sending an event back to the client that published it seems like OK behaviour, but I'm not entirely sure how to plumb that in - Publish() probably needs to gain an understanding of where a publish request came from, and handlers need to know where they are going to

[usedbytes]

Having chewed on this a bit, I came up with the commit above. It doesn't commute events across chained network subscriptions, but is fairly neat/pragmatic. Looking for feedback if there is any :smile:

[bennAH]

hi @usedbytes will ponder over this in the next few days.

Just to get a better understanding, what is your use case for this?

[usedbytes]

I'm using the bus to communicate between subsystems on a robot I'm building. e.g. the sensor subsystem(s) can publish events with sensor readings.

It's useful to send telemetry data from the robot to my laptop, which is easily achieved by letting the laptop "sniff" the bus - it'll receive the internal traffic by subscribing to the bus over the network. That works fine with the existing code, as the laptop only subscribes, and doesn't publish.

As an extension, it's nice from a development perspective if the laptop can inject events into the robot's on-board bus (e.g. to "fake" sensor readings, or to develop a subsystem on the laptop without needing to deploy it to the 'bot) - that's what leads to the bidirectional subscriptions. These could theoretically be published on a different topic, but it's obviously nicer if it can be on the same topic as if it were a "local" subsystem - i.e. the robot code doesn't need to know whether the event came from something on-board, or over the network.