update events to check connection if new update receved

Author: Eslam-Nawara

pkg/events/events.go

@@ -66,7 +66,8 @@ type Callback func(events *substrate.EventRecords)
 // Events processor receives all events starting from the given state
 // and for each set of events calls callback cb
 type Processor struct {
-	sub substrate.Manager
+	sub    substrate.Manager
+	update chan substrate.Manager
 
 	cb    Callback
 	state State
@@ -173,6 +174,15 @@ func (e *Processor) subscribe(ctx context.Context) error {
 			if err := e.state.Set(block.Number); err != nil {
 				return errors.Wrap(err, "failed to commit last block number")
 			}
+		case newSub := <-e.update:
+			// if new update is received check if the current manager is healty, if not:
+			// update the manager and return error to force resubscribe with the new manager
+			cl, _, err := e.sub.Raw()
+			if err != nil {
+				e.sub = newSub
+				return err
+			}
+			cl.Client.Close()
 		}
 	}
 }

pkg/events/redis.go

@@ -54,7 +54,7 @@ func NewRedisStream(sub substrate.Manager, address string, farm pkg.FarmID, node
 
 func (r *RedisStream) UpdateSubstrateManager(sub substrate.Manager) {
 	r.sub = sub
-	r.processor.sub = sub
+	r.processor.update <- sub
 }
 
 func (r *RedisStream) push(con redis.Conn, queue string, event interface{}) error {