feature: add stop loop watch function for with one or with batch event.

Author: breezelxp

src/storage/stream/loop/loop_watch.go

@@ -147,6 +147,11 @@ func (lw *LoopsWatch) watchRetry(cancel context.CancelFunc,
 		ctx := context.Background()
 
 		select {
+		case <-opts.StopNotifier:
+			cancel()
+			blog.Warnf("received stop %s loop watch job notify, stopping now.", opts.Name)
+			return
+
 		// wait for another retry
 		case <-retrySignal:
 			// wait for a well and then do the retry work.
@@ -266,6 +271,11 @@ func (lw *LoopsWatch) tryLoopWithBatch(ctxWithCancel context.Context,
 					// ticks, but no events received, loop next round to get events.
 					continue
 				}
+
+			case <-opts.StopNotifier:
+				ticker.Stop()
+				blog.Warnf("received stop %s loop watch job notify, stopping now.", opts.Name)
+				return
 			}
 
 			// break the for loop to handle event for now.
@@ -336,6 +346,11 @@ func (lw *LoopsWatch) tryLoopWithOne(ctxWithCancel context.Context,
 			blog.Warnf("%s job, received cancel loop watch %s signal, exit loop, exit loop", opts.Name,
 				opts.WatchOpt.Collection)
 			return
+
+		case <-opts.StopNotifier:
+			blog.Warnf("received stop %s loop watch job notify, stopping now.", opts.Name)
+			return
+
 		default:
 		}
 

src/storage/stream/types/types.go

@@ -350,6 +350,16 @@ type LoopOptions struct {
 	WatchOpt     *WatchOptions
 	TokenHandler TokenHandler
 	RetryOptions *RetryOptions
+
+	// StopNotifier is used when user need to stop loop events and release related resources.
+	// It's a optional option. when it's not set(as is nil), then the loop will not exit forever.
+	// Otherwise, user can use it to stop loop events.
+	// When a user want to stop the loop, the only thing that a user need to do is to just
+	// **close** this stop notifier channel.
+	// Attention:
+	// Close this notifier channel is the only way to stop loop correctly.
+	// Do not send data to this channel.
+	StopNotifier <-chan struct{}
 }
 
 type LoopOneOptions struct {
@@ -393,6 +403,11 @@ func (lo *LoopOneOptions) Validate() error {
 		}
 	}
 
+	if lo.LoopOptions.StopNotifier == nil {
+		// if not set, then set never stop loop as default
+		lo.LoopOptions.StopNotifier = make(<-chan struct{})
+	}
+
 	return nil
 }
 
@@ -449,6 +464,11 @@ func (lo *LoopBatchOptions) Validate() error {
 		}
 	}
 
+	if lo.LoopOptions.StopNotifier == nil {
+		// if not set, then set never stop loop as default
+		lo.LoopOptions.StopNotifier = make(<-chan struct{})
+	}
+
 	return nil
 }