chore: update even unsubscribe to get called in same go routine

Author: blockchainluffy

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -61,7 +61,11 @@ func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) erro
 		return err
 	}
 	runner.Go(func() error {
-		return s.watchNewEonPubkey(ctx, subs.Err(), subs.Unsubscribe)
+		if err := s.watchNewEonPubkey(ctx, subs.Err()); err != nil {
+			s.Log.Error("error watching new eon pubkey", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }
@@ -125,7 +129,7 @@ func (s *EonPubKeySyncer) GetEonPubKeyForEon(ctx context.Context, opts *bind.Cal
 	}, nil
 }
 
-func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan error, unsubscribe func()) error {
+func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newEonKey, ok := <-s.keyBroadcastCh:
@@ -152,7 +156,6 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan
 				return err
 			}
 		case <-ctx.Done():
-			unsubscribe()
 			return ctx.Err()
 		}
 	}

rolling-shutter/medley/chainsync/syncer/keyperset.go

@@ -71,7 +71,11 @@ func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) erro
 		return err
 	}
 	runner.Go(func() error {
-		return s.watchNewKeypersService(ctx, subs.Err(), subs.Unsubscribe)
+		if err := s.watchNewKeypersService(ctx, subs.Err()); err != nil {
+			s.Log.Error("error watching new keypers", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }
@@ -203,7 +207,7 @@ func (s *KeyperSetSyncer) newEvent(
 	}, nil
 }
 
-func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-chan error, unsubscribe func()) error {
+func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newKeypers, ok := <-s.keyperAddedCh:
@@ -239,7 +243,6 @@ func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-
 				return err
 			}
 		case <-ctx.Done():
-			unsubscribe()
 			return ctx.Err()
 		}
 	}

rolling-shutter/medley/chainsync/syncer/shutterstate.go

@@ -61,7 +61,12 @@ func (s *ShutterStateSyncer) Start(ctx context.Context, runner service.Runner) e
 	}
 
 	runner.Go(func() error {
-		return s.watchPaused(ctx, subs.Err(), subsUnpaused.Err(), subs.Unsubscribe, subsUnpaused.Unsubscribe)
+		if err := s.watchPaused(ctx, subs.Err(), subsUnpaused.Err()); err != nil {
+			s.Log.Error("error watching paused", err.Error())
+		}
+		subs.Unsubscribe()
+		subsUnpaused.Unsubscribe()
+		return err
 	})
 	return nil
 }
@@ -85,13 +90,7 @@ func (s *ShutterStateSyncer) handle(ctx context.Context, ev *event.ShutterState)
 	}
 }
 
-func (s *ShutterStateSyncer) watchPaused(
-	ctx context.Context,
-	subsErr <-chan error,
-	subsErrUnpaused <-chan error,
-	unsubscribe func(),
-	unsubscribeUnpaused func(),
-) error {
+func (s *ShutterStateSyncer) watchPaused(ctx context.Context, subsErr <-chan error, subsErrUnpaused <-chan error) error {
 	isActive, err := s.pollIsActive(ctx)
 	if err != nil {
 		// XXX: this will fail everything, do we want that?
@@ -138,8 +137,6 @@ func (s *ShutterStateSyncer) watchPaused(
 				return err
 			}
 		case <-ctx.Done():
-			unsubscribe()
-			unsubscribeUnpaused()
 			return ctx.Err()
 		}
 	}

rolling-shutter/medley/chainsync/syncer/unsafehead.go

@@ -32,12 +32,16 @@ func (s *UnsafeHeadSyncer) Start(ctx context.Context, runner service.Runner) err
 		return err
 	}
 	runner.Go(func() error {
-		return s.watchLatestUnsafeHead(ctx, subs.Err(), subs.Unsubscribe)
+		if err := s.watchLatestUnsafeHead(ctx, subs.Err()); err != nil {
+			s.Log.Error("error watching latest unsafe head", err.Error())
+		}
+		subs.Unsubscribe()
+		return err
 	})
 	return nil
 }
 
-func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-chan error, unsubscribe func()) error {
+func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newHeader, ok := <-s.newLatestHeadCh:
@@ -63,7 +67,6 @@ func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-
 				return err
 			}
 		case <-ctx.Done():
-			unsubscribe()
 			return ctx.Err()
 		}
 	}