migrate to eventbus, add support to macos

pkg/eventbus/bus.go

@@ -26,7 +26,7 @@ type Subscriber interface {
 	// SubscribeAny is Subscribe with empty channel. this function will subscribe to event from any channel
 	SubscribeAny(eventId string, handlerName string, fn HandlerFunc) error
 	// SubscribeOnce will run handler once, and delete handler internally
-	SubscribeOnce(eventId string, handlerName string, fn HandlerFunc) error
+	SubscribeOnce(channel string, eventId string, handlerName string, fn HandlerFunc) error
 	// Unsubscribe just remove handler for the bus
 	Unsubscribe(eventId string, handlerName string) error
 }
@@ -34,6 +34,8 @@ type Subscriber interface {
 type Publisher interface {
 	// Publish basically a wrapper to PublishEvent
 	Publish(eventId string, data interface{}) error
+	// PublishToChannel publish event to a specific channel, basically another wrapper to PublishEvent
+	PublishToChannel(channel string, eventId string, data interface{}) error
 	// PublishEvent publish an event
 	PublishEvent(event *Event) error
 }
@@ -41,6 +43,7 @@ type Publisher interface {
 // Caller is special usage of a Publisher
 type Caller interface {
 	Call(pubEvtId string, data interface{}, subEvtId string) (*Event, error)
+	Reply(req *Event, eventId string, data interface{}) error
 }
 
 type Controller interface {

pkg/eventbus/bus_impl.go

@@ -191,7 +191,7 @@ func (b *bus) SubscribeAny(eventId, handlerName string, fn HandlerFunc) error {
 	return nil
 }
 
-func (b *bus) SubscribeOnce(eventId, handlerName string, fn HandlerFunc) error {
+func (b *bus) SubscribeOnce(channel, eventId, handlerName string, fn HandlerFunc) error {
 	if eventId == "" || handlerName == "" || fn == nil {
 		return errors.New("invalid SubscribeOnce args")
 	}
@@ -202,7 +202,7 @@ func (b *bus) SubscribeOnce(eventId, handlerName string, fn HandlerFunc) error {
 		m = make(map[string]handlerRec)
 		b.handlers[eventId] = m
 	}
-	m[handlerName] = handlerRec{name: handlerName, fn: fn, once: true}
+	m[handlerName] = handlerRec{channel: channel, name: handlerName, fn: fn, once: true}
 	return nil
 }
 
@@ -225,6 +225,10 @@ func (b *bus) Publish(eventId string, data interface{}) error {
 	return b.PublishEvent(&Event{Id: eventId, Data: data})
 }
 
+func (b *bus) PublishToChannel(channel string, eventId string, data interface{}) error {
+	return b.PublishEvent(&Event{Id: eventId, Channel: channel, Data: data})
+}
+
 func (b *bus) PublishEvent(ev *Event) error {
 	if ev == nil || ev.Id == "" {
 		return errors.New("invalid PublishEvent args")
@@ -327,6 +331,15 @@ func (b *bus) Call(eventId string, data interface{}, subEvtId string) (*Event, e
 	}
 }
 
+func (b *bus) Reply(req *Event, eventId string, data interface{}) error {
+	return b.PublishEvent(&Event{
+		Id:      eventId,
+		Channel: req.Channel,
+		EchoId:  req.EchoId,
+		Data:    data,
+	})
+}
+
 func (b *bus) nextEchoId() string {
 	x := b.idCtr.Add(1)
 	return fmt.Sprintf("echo-%d-%d", time.Now().UnixNano(), x)