Replace custom pubsub

api/api.go

@@ -3,17 +3,19 @@ package api
 import (
 	"fmt"
 
+	"github.com/cskr/pubsub"
 	"github.com/mesg-foundation/core/container"
 	"github.com/mesg-foundation/core/database"
 	"github.com/mesg-foundation/core/event"
 	"github.com/mesg-foundation/core/execution"
-	"github.com/mesg-foundation/core/pubsub"
 	"github.com/mesg-foundation/core/service"
 	uuid "github.com/satori/go.uuid"
 )
 
 // API exposes all functionalities of MESG core.
 type API struct {
+	ps *pubsub.PubSub
+
 	db        database.ServiceDB
 	execDB    database.ExecutionDB
 	container container.Container
@@ -22,6 +24,7 @@ type API struct {
 // New creates a new API with given options.
 func New(c container.Container, db database.ServiceDB, execDB database.ExecutionDB) *API {
 	return &API{
+		ps:        pubsub.New(0),
 		container: c,
 		db:        db,
 		execDB:    execDB,
@@ -94,7 +97,8 @@ func (a *API) EmitEvent(token, eventKey string, eventData map[string]interface{}
 	if err != nil {
 		return err
 	}
-	event.Publish()
+
+	go a.ps.Pub(event, s.EventSubscriptionChannel())
 	return nil
 }
 
@@ -126,7 +130,7 @@ func (a *API) ExecuteTask(serviceID, taskKey string, inputData map[string]interf
 	if err = a.execDB.Save(exec); err != nil {
 		return "", err
 	}
-	go pubsub.Publish(s.TaskSubscriptionChannel(), exec)
+	go a.ps.Pub(exec, s.TaskSubscriptionChannel())
 	return exec.ID, nil
 }
 

api/listen_event.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"github.com/mesg-foundation/core/event"
-	"github.com/mesg-foundation/core/pubsub"
 	"github.com/mesg-foundation/core/service"
 	"github.com/mesg-foundation/core/x/xstrings"
 )
@@ -73,9 +72,9 @@ func (l *EventListener) listen(serviceID string) error {
 }
 
 func (l *EventListener) listenLoop(service *service.Service) {
-	channel := service.EventSubscriptionChannel()
-	subscription := pubsub.Subscribe(channel)
-	defer pubsub.Unsubscribe(channel, subscription)
+	topic := service.EventSubscriptionChannel()
+	subscription := l.api.ps.Sub(topic)
+	defer l.api.ps.Unsub(subscription, topic)
 	close(l.listening)
 
 	for {

api/listen_result.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"github.com/mesg-foundation/core/execution"
-	"github.com/mesg-foundation/core/pubsub"
 	"github.com/mesg-foundation/core/service"
 	"github.com/mesg-foundation/core/x/xstrings"
 )
@@ -89,9 +88,9 @@ func (l *ResultListener) listen(serviceID string) error {
 }
 
 func (l *ResultListener) listenLoop(service *service.Service) {
-	channel := service.ResultSubscriptionChannel()
-	subscription := pubsub.Subscribe(channel)
-	defer pubsub.Unsubscribe(channel, subscription)
+	topic := service.ResultSubscriptionChannel()
+	subscription := l.api.ps.Sub(topic)
+	defer l.api.ps.Unsub(subscription, topic)
 	close(l.listening)
 
 	for {

api/listen_task.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"github.com/mesg-foundation/core/execution"
-	"github.com/mesg-foundation/core/pubsub"
 	"github.com/mesg-foundation/core/service"
 )
 
@@ -53,9 +52,9 @@ func (l *TaskListener) listen(token string) error {
 }
 
 func (l *TaskListener) listenLoop(service *service.Service) {
-	channel := service.TaskSubscriptionChannel()
-	subscription := pubsub.Subscribe(channel)
-	defer pubsub.Unsubscribe(channel, subscription)
+	topic := service.TaskSubscriptionChannel()
+	subscription := l.api.ps.Sub(topic)
+	defer l.api.ps.Unsub(subscription, topic)
 	close(l.listening)
 
 	for {

api/submit_result.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/mesg-foundation/core/database"
 	"github.com/mesg-foundation/core/execution"
-	"github.com/mesg-foundation/core/pubsub"
 )
 
 // SubmitResult submits results for executionID.
@@ -31,7 +30,7 @@ func (s *resultSubmitter) Submit(executionID string, outputKey string, outputDat
 	exec, stateChanged, err := s.processExecution(executionID, outputKey, outputData)
 	if stateChanged {
 		// only publish to listeners when the execution's state changed.
-		go pubsub.Publish(exec.Service.ResultSubscriptionChannel(), exec)
+		go s.api.ps.Pub(exec, exec.Service.ResultSubscriptionChannel())
 	}
 	// always return any error to the service.
 	return err

event/event.go

@@ -3,7 +3,6 @@ package event
 import (
 	"time"
 
-	"github.com/mesg-foundation/core/pubsub"
 	"github.com/mesg-foundation/core/service"
 )
 
@@ -31,9 +30,3 @@ func Create(s *service.Service, eventKey string, eventData map[string]interface{
 		CreatedAt: time.Now(),
 	}, nil
 }
-
-// Publish publishes an event for every listener.
-func (event *Event) Publish() {
-	channel := event.Service.EventSubscriptionChannel()
-	go pubsub.Publish(channel, event)
-}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/briandowns/spinner v0.0.0-20180822135157-9f016caa1359
 	github.com/cnf/structhash v0.0.0-20180104161610-62a607eb0224
 	github.com/containerd/continuity v0.0.0-20180712174259-0377f7d76720 // indirect
+	github.com/cskr/pubsub v1.0.2
 	github.com/docker/cli v0.0.0-20190129171106-b258f458cc8d
 	github.com/docker/distribution v0.0.0-20180720172123-0dae0957e5fe // indirect
 	github.com/docker/docker v0.0.0-20180803200506-eeea12db7a65

go.sum

@@ -16,6 +16,8 @@ github.com/cnf/structhash v0.0.0-20180104161610-62a607eb0224 h1:rnCKRrdSBqc061l0
 github.com/cnf/structhash v0.0.0-20180104161610-62a607eb0224/go.mod h1:pCxVEbcm3AMg7ejXyorUXi6HQCzOIBf7zEDVPtw0/U4=
 github.com/containerd/continuity v0.0.0-20180712174259-0377f7d76720 h1:T5LkgEMACPq7+VPRnkh51WhyV+Q0waOGGtp37Y82org=
 github.com/containerd/continuity v0.0.0-20180712174259-0377f7d76720/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
+github.com/cskr/pubsub v1.0.2 h1:vlOzMhl6PFn60gRlTQQsIfVwaPB/B/8MziK8FhEPt/0=
+github.com/cskr/pubsub v1.0.2/go.mod h1:/8MzYXk/NJAz782G8RPkFzXTZVu63VotefPnR9TIRis=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=