use common hub with common publisher

Author: ssd04

common/errors.go

@@ -34,3 +34,6 @@ var ErrNilStatusMetricsHandler = errors.New("nil status metrics handler")
 
 // ErrWrongTypeAssertion signals a wrong type assertion
 var ErrWrongTypeAssertion = errors.New("wrong type assertion")
+
+// ErrLoopAlreadyStarted signals that a loop has already been started
+var ErrLoopAlreadyStarted = errors.New("loop already started")

disabled/disabledHub.go

@@ -9,33 +9,33 @@ import (
 type Hub struct {
 }
 
-// Run does nothing
-func (h *Hub) Run() error {
+// RegisterListener does nothing
+func (h *Hub) RegisterListener() error {
 	return nil
 }
 
-// Broadcast does nothing
-func (h *Hub) Broadcast(_ data.BlockEvents) {
+// Publish -
+func (h *Hub) Publish(events data.BlockEvents) {
 }
 
-// BroadcastRevert does nothing
-func (h *Hub) BroadcastRevert(_ data.RevertBlock) {
+// PublishRevert -
+func (h *Hub) PublishRevert(revertBlock data.RevertBlock) {
 }
 
-// BroadcastFinalized does nothing
-func (h *Hub) BroadcastFinalized(_ data.FinalizedBlock) {
+// PublishFinalized -
+func (h *Hub) PublishFinalized(finalizedBlock data.FinalizedBlock) {
 }
 
-// BroadcastTxs does nothing
-func (h *Hub) BroadcastTxs(_ data.BlockTxs) {
+// PublishTxs -
+func (h *Hub) PublishTxs(blockTxs data.BlockTxs) {
 }
 
-// BroadcastScrs does nothing
-func (h *Hub) BroadcastScrs(_ data.BlockScrs) {
+// PublishScrs -
+func (h *Hub) PublishScrs(blockScrs data.BlockScrs) {
 }
 
-// BroadcastBlockEventsWithOrder does nothing
-func (h *Hub) BroadcastBlockEventsWithOrder(_ data.BlockEventsWithOrder) {
+// PublishBlockEventsWithOrder -
+func (h *Hub) PublishBlockEventsWithOrder(blockTxs data.BlockEventsWithOrder) {
 }
 
 // RegisterEvent does nothing

dispatcher/hub/commonHub.go

@@ -22,20 +22,16 @@ type ArgsCommonHub struct {
 }
 
 type commonHub struct {
-	filter                        filters.EventFilter
-	subscriptionMapper            dispatcher.SubscriptionMapperHandler
-	mutDispatchers                sync.RWMutex
-	dispatchers                   map[uuid.UUID]dispatcher.EventDispatcher
-	register                      chan dispatcher.EventDispatcher
-	unregister                    chan dispatcher.EventDispatcher
-	broadcast                     chan data.BlockEvents
-	broadcastRevert               chan data.RevertBlock
-	broadcastFinalized            chan data.FinalizedBlock
-	broadcastTxs                  chan data.BlockTxs
-	broadcastBlockEventsWithOrder chan data.BlockEventsWithOrder
-	broadcastScrs                 chan data.BlockScrs
-	closeChan                     chan struct{}
-	cancelFunc                    func()
+	filter             filters.EventFilter
+	subscriptionMapper dispatcher.SubscriptionMapperHandler
+	mutDispatchers     sync.RWMutex
+	dispatchers        map[uuid.UUID]dispatcher.EventDispatcher
+	register           chan dispatcher.EventDispatcher
+	unregister         chan dispatcher.EventDispatcher
+
+	cancelFunc func()
+	closeChan  chan struct{}
+	mutState   sync.RWMutex
 }
 
 // NewCommonHub creates a new commonHub instance
@@ -46,19 +42,13 @@ func NewCommonHub(args ArgsCommonHub) (*commonHub, error) {
 	}
 
 	return &commonHub{
-		mutDispatchers:                sync.RWMutex{},
-		filter:                        args.Filter,
-		subscriptionMapper:            args.SubscriptionMapper,
-		dispatchers:                   make(map[uuid.UUID]dispatcher.EventDispatcher),
-		register:                      make(chan dispatcher.EventDispatcher),
-		unregister:                    make(chan dispatcher.EventDispatcher),
-		broadcast:                     make(chan data.BlockEvents),
-		broadcastRevert:               make(chan data.RevertBlock),
-		broadcastFinalized:            make(chan data.FinalizedBlock),
-		broadcastTxs:                  make(chan data.BlockTxs),
-		broadcastBlockEventsWithOrder: make(chan data.BlockEventsWithOrder),
-		broadcastScrs:                 make(chan data.BlockScrs),
-		closeChan:                     make(chan struct{}),
+		mutDispatchers:     sync.RWMutex{},
+		filter:             args.Filter,
+		subscriptionMapper: args.SubscriptionMapper,
+		dispatchers:        make(map[uuid.UUID]dispatcher.EventDispatcher),
+		register:           make(chan dispatcher.EventDispatcher),
+		unregister:         make(chan dispatcher.EventDispatcher),
+		closeChan:          make(chan struct{}),
 	}, nil
 }
 
@@ -73,42 +63,26 @@ func checkArgs(args ArgsCommonHub) error {
 	return nil
 }
 
-// Run is launched as a goroutine and listens for events on the exposed channels
-// TODO: use protection from triggering multiple times
-func (ch *commonHub) Run() error {
+// RegisterListener creates a goroutine and listens for WS register events
+func (ch *commonHub) RegisterListener() error {
+	ch.mutState.Lock()
+	defer ch.mutState.Unlock()
+
+	if ch.cancelFunc != nil {
+		return common.ErrLoopAlreadyStarted
+	}
+
 	var ctx context.Context
 	ctx, ch.cancelFunc = context.WithCancel(context.Background())
 
-	go ch.run(ctx)
+	go ch.registerListener(ctx)
 
 	return nil
 }
 
-func (ch *commonHub) run(ctx context.Context) {
+func (ch *commonHub) registerListener(ctx context.Context) {
 	for {
 		select {
-		case <-ctx.Done():
-			log.Debug("commonHub is stopping...")
-			return
-
-		case events := <-ch.broadcast:
-			ch.handleBroadcast(events)
-
-		case revertEvent := <-ch.broadcastRevert:
-			ch.handleRevertBroadcast(revertEvent)
-
-		case finalizedEvent := <-ch.broadcastFinalized:
-			ch.handleFinalizedBroadcast(finalizedEvent)
-
-		case txsEvent := <-ch.broadcastTxs:
-			ch.handleTxsBroadcast(txsEvent)
-
-		case txsEvent := <-ch.broadcastBlockEventsWithOrder:
-			ch.handleBlockEventsWithOrderBroadcast(txsEvent)
-
-		case scrsEvent := <-ch.broadcastScrs:
-			ch.handleScrsBroadcast(scrsEvent)
-
 		case dispatcherClient := <-ch.register:
 			ch.registerDispatcher(dispatcherClient)
 
@@ -123,59 +97,6 @@ func (ch *commonHub) Subscribe(event data.SubscribeEvent) {
 	ch.subscriptionMapper.MatchSubscribeEvent(event)
 }
 
-// Broadcast handles block events pushed by producers into the broadcast channel
-// Upon reading the channel, the hub notifies the registered dispatchers, if any
-func (ch *commonHub) Broadcast(events data.BlockEvents) {
-	select {
-	case ch.broadcast <- events:
-	case <-ch.closeChan:
-	}
-}
-
-// BroadcastRevert handles revert event pushed by producers into the broadcast channel
-// Upon reading the channel, the hub notifies the registered dispatchers, if any
-func (ch *commonHub) BroadcastRevert(event data.RevertBlock) {
-	select {
-	case ch.broadcastRevert <- event:
-	case <-ch.closeChan:
-	}
-}
-
-// BroadcastFinalized handles finalized event pushed by producers into the broadcast channel
-// Upon reading the channel, the hub notifies the registered dispatchers, if any
-func (ch *commonHub) BroadcastFinalized(event data.FinalizedBlock) {
-	select {
-	case ch.broadcastFinalized <- event:
-	case <-ch.closeChan:
-	}
-}
-
-// BroadcastTxs handles block txs event pushed by producers into the broadcast channel
-// Upon reading the channel, the hub notifies the registered dispatchers, if any
-func (ch *commonHub) BroadcastTxs(event data.BlockTxs) {
-	select {
-	case ch.broadcastTxs <- event:
-	case <-ch.closeChan:
-	}
-}
-
-// BroadcastScrs handles block scrs event pushed by producers into the broadcast channel
-// Upon reading the channel, the hub notifies the registered dispatchers, if any
-func (ch *commonHub) BroadcastScrs(event data.BlockScrs) {
-	select {
-	case ch.broadcastScrs <- event:
-	case <-ch.closeChan:
-	}
-}
-
-// BroadcastBlockEventsWithOrder handles full block events pushed by producers into the channel
-func (ch *commonHub) BroadcastBlockEventsWithOrder(event data.BlockEventsWithOrder) {
-	select {
-	case ch.broadcastBlockEventsWithOrder <- event:
-	case <-ch.closeChan:
-	}
-}
-
 // RegisterEvent will send event to a receive-only channel used to register dispatchers
 func (ch *commonHub) RegisterEvent(event dispatcher.EventDispatcher) {
 	select {
@@ -192,7 +113,8 @@ func (ch *commonHub) UnregisterEvent(event dispatcher.EventDispatcher) {
 	}
 }
 
-func (ch *commonHub) handleBroadcast(blockEvents data.BlockEvents) {
+// Publish will publish logs and events to dispatcher
+func (ch *commonHub) Publish(blockEvents data.BlockEvents) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	for _, subscription := range subscriptions {
@@ -220,7 +142,8 @@ func (ch *commonHub) handlePushBlockEvents(blockEvents data.BlockEvents, subscri
 	ch.mutDispatchers.RUnlock()
 }
 
-func (ch *commonHub) handleRevertBroadcast(revertBlock data.RevertBlock) {
+// PublishRevert will publish revert event to dispatcher
+func (ch *commonHub) PublishRevert(revertBlock data.RevertBlock) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	dispatchersMap := make(map[uuid.UUID]data.RevertBlock)
@@ -242,7 +165,8 @@ func (ch *commonHub) handleRevertBroadcast(revertBlock data.RevertBlock) {
 	}
 }
 
-func (ch *commonHub) handleFinalizedBroadcast(finalizedBlock data.FinalizedBlock) {
+// PublishFinalized will publish finalized event to dispatcher
+func (ch *commonHub) PublishFinalized(finalizedBlock data.FinalizedBlock) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	dispatchersMap := make(map[uuid.UUID]data.FinalizedBlock)
@@ -264,7 +188,8 @@ func (ch *commonHub) handleFinalizedBroadcast(finalizedBlock data.FinalizedBlock
 	}
 }
 
-func (ch *commonHub) handleTxsBroadcast(blockTxs data.BlockTxs) {
+// PublishTxs will publish txs event to dispatcher
+func (ch *commonHub) PublishTxs(blockTxs data.BlockTxs) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockTxs)
@@ -286,7 +211,8 @@ func (ch *commonHub) handleTxsBroadcast(blockTxs data.BlockTxs) {
 	}
 }
 
-func (ch *commonHub) handleBlockEventsWithOrderBroadcast(blockTxs data.BlockEventsWithOrder) {
+// PublishBlockEventsWithOrder will publish block events with order to dispatcher
+func (ch *commonHub) PublishBlockEventsWithOrder(blockTxs data.BlockEventsWithOrder) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockEventsWithOrder)
@@ -308,7 +234,7 @@ func (ch *commonHub) handleBlockEventsWithOrderBroadcast(blockTxs data.BlockEven
 	}
 }
 
-func (ch *commonHub) handleScrsBroadcast(blockScrs data.BlockScrs) {
+func (ch *commonHub) PublishScrs(blockScrs data.BlockScrs) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockScrs)

dispatcher/interface.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/multiversx/mx-chain-notifier-go/data"
+	"github.com/multiversx/mx-chain-notifier-go/process"
 )
 
 // EventDispatcher defines the behaviour of a event dispatcher component
@@ -20,20 +21,20 @@ type EventDispatcher interface {
 	ScrsEvent(event data.BlockScrs)
 }
 
-// Hub defines the behaviour of a hub component which should be able to register
-// and unregister dispatching events
+// Hub defines the behaviour of a component which should be able to receive events
+// and publish them to subscribers
 type Hub interface {
-	Run() error
-	Broadcast(events data.BlockEvents)
-	BroadcastRevert(event data.RevertBlock)
-	BroadcastFinalized(event data.FinalizedBlock)
-	BroadcastTxs(event data.BlockTxs)
-	BroadcastScrs(event data.BlockScrs)
-	BroadcastBlockEventsWithOrder(event data.BlockEventsWithOrder)
+	process.PublisherHandler
+	Dispatcher
+}
+
+// Dispatcher defines the behaviour of a dispatcher component which should be able to register
+// and unregister dispatching events
+type Dispatcher interface {
+	RegisterListener() error
 	RegisterEvent(event EventDispatcher)
 	UnregisterEvent(event EventDispatcher)
 	Subscribe(event data.SubscribeEvent)
-	Close() error
 	IsInterfaceNil() bool
 }
 

dispatcher/ws/errors.go

@@ -2,8 +2,8 @@ package ws
 
 import "errors"
 
-// ErrNilHubHandler signals that a nil hub handler has been provided
-var ErrNilHubHandler = errors.New("nil hub handler")
+// ErrNilDispatcher signals that a nil dispatcher has been provided
+var ErrNilDispatcher = errors.New("nil dispatcher")
 
 // ErrNilWSUpgrader signals that a nil websocket upgrader has been provided
 var ErrNilWSUpgrader = errors.New("nil websocket upgrader")

dispatcher/ws/export_test.go

@@ -8,7 +8,7 @@ type ArgsWSDispatcher struct {
 // NewTestWSDispatcher -
 func NewTestWSDispatcher(args ArgsWSDispatcher) (*websocketDispatcher, error) {
 	wsArgs := argsWebSocketDispatcher{
-		Hub:        args.Hub,
+		Dispatcher: args.Dispatcher,
 		Conn:       args.Conn,
 		Marshaller: args.Marshaller,
 	}

dispatcher/ws/wsDispatcher.go

@@ -32,7 +32,7 @@ var (
 
 // argsWebSocketDispatcher defines the arguments needed for ws dispatcher
 type argsWebSocketDispatcher struct {
-	Hub        dispatcher.Hub
+	Dispatcher dispatcher.Dispatcher
 	Conn       dispatcher.WSConnection
 	Marshaller marshal.Marshalizer
 }
@@ -42,14 +42,14 @@ type websocketDispatcher struct {
 	wg         sync.WaitGroup
 	send       chan []byte
 	conn       dispatcher.WSConnection
-	hub        dispatcher.Hub
+	dispatcher dispatcher.Dispatcher
 	marshaller marshal.Marshalizer
 }
 
 // newWebSocketDispatcher createa a new ws dispatcher instance
 func newWebSocketDispatcher(args argsWebSocketDispatcher) (*websocketDispatcher, error) {
-	if check.IfNil(args.Hub) {
-		return nil, ErrNilHubHandler
+	if check.IfNil(args.Dispatcher) {
+		return nil, ErrNilDispatcher
 	}
 	if args.Conn == nil {
 		return nil, ErrNilWSConn
@@ -62,7 +62,7 @@ func newWebSocketDispatcher(args argsWebSocketDispatcher) (*websocketDispatcher,
 		id:         uuid.New(),
 		send:       make(chan []byte, 256),
 		conn:       args.Conn,
-		hub:        args.Hub,
+		dispatcher: args.Dispatcher,
 		marshaller: args.Marshaller,
 	}, nil
 }
@@ -254,7 +254,7 @@ func (wd *websocketDispatcher) writePump() {
 // readPump listens for incoming events and reads the content from the socket stream
 func (wd *websocketDispatcher) readPump() {
 	defer func() {
-		wd.hub.UnregisterEvent(wd)
+		wd.dispatcher.UnregisterEvent(wd)
 		if err := wd.conn.Close(); err != nil {
 			log.Error("failed to close socket on defer", "err", err.Error())
 		}
@@ -293,7 +293,7 @@ func (wd *websocketDispatcher) trySendSubscribeEvent(eventBytes []byte) {
 		return
 	}
 	subscribeEvent.DispatcherID = wd.id
-	wd.hub.Subscribe(subscribeEvent)
+	wd.dispatcher.Subscribe(subscribeEvent)
 }
 
 func (wd *websocketDispatcher) setSocketWriteLimits() error {