add a more friendly method AddOnePath for DynamicStream

downstreamadapter/dispatcher/dispatcher.go

@@ -104,13 +104,13 @@ func NewDispatcher(tableSpan *common.TableSpan, sink sink.Sink, startTs uint64,
 
 	dispatcherEventsDynamicStream := appcontext.GetService[dynstream.DynamicStream[common.DispatcherID, *common.TxnEvent, *Dispatcher]](appcontext.DispatcherEventsDynamicStream)
 
-	err := dispatcherEventsDynamicStream.AddPath(dynstream.PathAndDest[common.DispatcherID, *Dispatcher]{Path: dispatcher.id, Dest: dispatcher})
+	err := dispatcherEventsDynamicStream.AddOnePath(dispatcher.id, dispatcher)
 	if err != nil {
 		log.Error("add dispatcher to dynamic stream failed", zap.Error(err))
 	}
 
 	dispatcherStatusDynamicStream := appcontext.GetService[dynstream.DynamicStream[common.DispatcherID, DispatcherStatusWithDispatcherID, *Dispatcher]](appcontext.DispatcherStatusDynamicStream)
-	err = dispatcherStatusDynamicStream.AddPath(dynstream.PathAndDest[common.DispatcherID, *Dispatcher]{Path: dispatcher.id, Dest: dispatcher})
+	err = dispatcherStatusDynamicStream.AddOnePath(dispatcher.id, dispatcher)
 	if err != nil {
 		log.Error("add dispatcher to dynamic stream failed", zap.Error(err))
 	}

downstreamadapter/dispatchermanager/heartbeat_collector.go

@@ -60,12 +60,12 @@ func NewHeartBeatCollector(serverId messaging.ServerId) *HeartBeatCollector {
 
 func (c *HeartBeatCollector) RegisterEventDispatcherManager(m *EventDispatcherManager) error {
 	m.SetHeartbeatRequestQueue(c.requestQueue)
-	err := c.heartBeatResponseDynamicStream.AddPath(dynstream.PathAndDest[model.ChangeFeedID, *EventDispatcherManager]{Path: m.changefeedID, Dest: m})
+	err := c.heartBeatResponseDynamicStream.AddOnePath(m.changefeedID, m)
 	if err != nil {
 		log.Error("heartBeatResponseDynamicStream Failed to add path", zap.Any("ChangefeedID", m.changefeedID))
 		return err
 	}
-	err = c.schedulerDispatcherRequestDynamicStream.AddPath(dynstream.PathAndDest[model.ChangeFeedID, *EventDispatcherManager]{Path: m.changefeedID, Dest: m})
+	err = c.schedulerDispatcherRequestDynamicStream.AddOnePath(m.changefeedID, m)
 	if err != nil {
 		log.Error("schedulerDispatcherRequestDynamicStream Failed to add path", zap.Any("ChangefeedID", m.changefeedID))
 		return err

utils/dynstream/dynamic_stream.go

@@ -194,6 +194,10 @@ func (d *dynamicStreamImpl[P, T, D]) AddPath(paths ...PathAndDest[P, D]) error {
 	return add.error
 }
 
+func (d *dynamicStreamImpl[P, T, D]) AddOnePath(path P, dest D) error {
+	return d.AddPath(PathAndDest[P, D]{Path: path, Dest: dest})
+}
+
 func (d *dynamicStreamImpl[P, T, D]) RemovePath(paths ...P) []error {
 	remove := &removePathCmd[P]{paths: paths}
 	cmd := &cmd{

utils/dynstream/interfaces.go

@@ -55,6 +55,7 @@ type DynamicStream[P Path, T Event, D Dest] interface {
 	// If some paths already exist, it will return an ErrorTypeDuplicate error. And no paths are added.
 	// If the stream is closed, it will return an ErrorTypeClosed error.
 	AddPath(paths ...PathAndDest[P, D]) error
+	AddOnePath(path P, dest D) error
 
 	// RemovePath removes the paths from the dynamic stream.
 	// After this call return, future events with the paths will be dropped, including events which are already in the stream.