add altered accounts to boradcast

common/constants.go

@@ -38,7 +38,7 @@ const (
 	// BlockScrs defines the subscription event type for block scrs
 	BlockScrs string = "block_scrs"
 
-	AlteredAccountsEvent string = "altered_accounts"
+	AlteredAccountsEvent string = "mvx_altered_accounts"
 )
 
 const (

dispatcher/subscription.go

@@ -146,7 +146,6 @@ func getEventType(subEntry data.SubscriptionEntry) string {
 		subEntry.EventType == common.BlockTxs ||
 		subEntry.EventType == common.BlockScrs ||
 		subEntry.EventType == common.AlteredAccountsEvent ||
-		subEntry.EventType == "mvx_altered_accounts" ||
 		subEntry.EventType == common.BlockEvents {
 		return subEntry.EventType
 	}

process/eventsHandler.go

@@ -111,6 +111,13 @@ func (eh *eventsHandler) HandleSaveBlockEvents(allEvents data.ArgsSaveBlockData)
 	}
 	eh.handleBlockScrs(scrs)
 
+	alteredEvent := data.AlteredAccountsEvent{
+		Hash:      eventsData.Hash,
+		ShardID:   eventsData.Header.GetShardID(),
+		TimeStamp: eventsData.Header.GetTimeStamp(),
+		Accounts:  eventsData.AlteredAccounts,
+	}
+
 	txsWithOrder := data.BlockEventsWithOrder{
 		Hash:      eventsData.Hash,
 		ShardID:   eventsData.Header.GetShardID(),
@@ -120,7 +127,7 @@ func (eh *eventsHandler) HandleSaveBlockEvents(allEvents data.ArgsSaveBlockData)
 		Events:    eventsData.LogEvents,
 	}
 	eh.handleBlockEventsWithOrder(txsWithOrder)
-
+	eh.handleAlteredAccounts(alteredEvent)
 	return nil
 }
 
@@ -314,6 +321,27 @@ func (eh *eventsHandler) handleBlockEventsWithOrder(blockTxs data.BlockEventsWit
 	eh.metricsHandler.AddRequest(getRabbitOpID(common.BlockEvents), time.Since(t))
 }
 
+// handleBlockEventsWithOrder will handle full block events received from observer
+func (eh *eventsHandler) handleAlteredAccounts(alteredAccountsEvent data.AlteredAccountsEvent) {
+	if len(alteredAccountsEvent.Accounts) == 0 {
+		log.Warn("received empty accounts", "event", common.AlteredAccountsEvent,
+			"will process", false,
+		)
+		return
+	}
+
+	log.Info("received", "event", common.AlteredAccountsEvent,
+		"block hash", alteredAccountsEvent.Hash,
+	)
+
+	t := time.Now()
+
+	for _, publisher := range eh.publishers {
+		publisher.BroadcastAlteredAccounts(alteredAccountsEvent)
+	}
+	eh.metricsHandler.AddRequest(getRabbitOpID(common.AlteredAccountsEvent), time.Since(t))
+}
+
 func (eh *eventsHandler) tryCheckProcessedWithRetry(id, blockHash string) bool {
 	var err error
 	var setSuccessful bool

process/interface.go

@@ -24,6 +24,7 @@ type Publisher interface {
 	BroadcastFinalized(event data.FinalizedBlock)
 	BroadcastTxs(event data.BlockTxs)
 	BroadcastBlockEventsWithOrder(event data.BlockEventsWithOrder)
+	BroadcastAlteredAccounts(accounts data.AlteredAccountsEvent)
 	BroadcastScrs(event data.BlockScrs)
 	Close() error
 	IsInterfaceNil() bool

rabbitmq/interface.go

@@ -27,6 +27,7 @@ type PublisherService interface {
 	BroadcastTxs(event data.BlockTxs)
 	BroadcastScrs(event data.BlockScrs)
 	BroadcastBlockEventsWithOrder(event data.BlockEventsWithOrder)
+	BroadcastAlteredAccounts(accounts data.AlteredAccountsEvent)
 	Close() error
 	IsInterfaceNil() bool
 }