fixed listener stop partition handling

Author: rekby

examples/topic/topicreader/topicreader_handler.go

@@ -55,7 +55,7 @@ func (h *TopicEventsHandler) OnReadMessages(
 
 func (h *TopicEventsHandler) OnStartPartitionSessionRequest(
 	ctx context.Context,
-	event *topiclistener.StartPartitionSessionRequest,
+	event *topiclistener.StartPartitionSessionEvent,
 ) error {
 	lockID, offset, err := lockPartition(ctx, event.PartitionSession.TopicPath, event.PartitionSession.PartitionID)
 
@@ -75,7 +75,7 @@ func (h *TopicEventsHandler) OnStartPartitionSessionRequest(
 
 func (h *TopicEventsHandler) OnStopPartitionSessionRequest(
 	ctx context.Context,
-	event *topiclistener.StopPartitionSessionRequest,
+	event *topiclistener.StopPartitionSessionEvent,
 ) error {
 	h.m.Lock()
 	lockID := h.locks[event.PartitionSession.PartitionSessionID]

internal/topic/topiclistenerinternal/event_handler.go

@@ -62,7 +62,6 @@ func NewPublicStartPartitionSessionEvent(
 		PartitionSession: session,
 		CommittedOffset:  committedOffset,
 		PartitionOffsets: partitionOffsets,
-		confirm:          newConfirmStorage[PublicStartPartitionSessionConfirm](),
 	}
 }
 
@@ -126,12 +125,15 @@ type PublicStopPartitionSessionEvent struct {
 	confirm confirmStorage[empty.Struct]
 }
 
-func NewPublicStopPartitionSessionEvent(partitionSession topicreadercommon.PublicPartitionSession, graceful bool, committedOffset int64) *PublicStopPartitionSessionEvent {
+func NewPublicStopPartitionSessionEvent(
+	partitionSession topicreadercommon.PublicPartitionSession,
+	graceful bool,
+	committedOffset int64,
+) *PublicStopPartitionSessionEvent {
 	return &PublicStopPartitionSessionEvent{
 		PartitionSession: partitionSession,
 		Graceful:         graceful,
 		CommittedOffset:  committedOffset,
-		confirm:          newConfirmStorage[empty.Struct](),
 	}
 }
 

internal/topic/topiclistenerinternal/stream_listener.go

@@ -287,7 +287,7 @@ func (l *streamListener) onStartPartitionRequest(
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-event.confirm.DoneChan:
+	case <-event.confirm.Done():
 		userResp, _ = event.confirm.Get()
 	}
 
@@ -344,15 +344,15 @@ func (l *streamListener) onStopPartitionRequest(
 		// remove partition on the confirmation or on the listener closed
 		select {
 		case <-l.background.Done():
-		case <-event.confirm.DoneChan:
+		case <-event.confirm.Done():
 		}
 		_, _ = l.sessions.Remove(m.PartitionSessionID)
 	}()
 
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-event.confirm.DoneChan:
+	case <-event.confirm.Done():
 		// pass
 	}
 
@@ -398,27 +398,37 @@ func (l *streamListener) sendMessage(m rawtopicreader.ClientMessage) {
 }
 
 type confirmStorage[T any] struct {
-	DoneChan      empty.Chan
+	doneChan      empty.Chan
 	confirmed     atomic.Bool
 	val           T
 	confirmAction sync.Once
+	initAction    sync.Once
 }
 
-func newConfirmStorage[T any]() confirmStorage[T] {
-	return confirmStorage[T]{
-		DoneChan: make(empty.Chan),
-	}
+func (c *confirmStorage[T]) init() {
+	c.initAction.Do(func() {
+		c.doneChan = make(empty.Chan)
+	})
 }
 
 func (c *confirmStorage[T]) Set(val T) {
+	c.init()
 	c.confirmAction.Do(func() {
 		c.val = val
 		c.confirmed.Store(true)
-		close(c.DoneChan)
+		close(c.doneChan)
 	})
 }
 
+func (c *confirmStorage[T]) Done() empty.ChanReadonly {
+	c.init()
+
+	return c.doneChan
+}
+
 func (c *confirmStorage[T]) Get() (val T, ok bool) {
+	c.init()
+
 	if c.confirmed.Load() {
 		return c.val, true
 	}

tests/integration/topic_listener_test.go

@@ -51,8 +51,8 @@ type TestTopicListener_Handler struct {
 
 	listener         *topiclistener.TopicListener
 	readMessages     *topiclistener.ReadMessages
-	onPartitionStart *topiclistener.StartPartitionSessionRequest
-	onPartitionStop  *topiclistener.StopPartitionSessionRequest
+	onPartitionStart *topiclistener.StartPartitionSessionEvent
+	onPartitionStop  *topiclistener.StopPartitionSessionEvent
 	done             empty.Chan
 }
 
@@ -63,7 +63,7 @@ func (h *TestTopicListener_Handler) OnReaderCreated(event *topiclistener.ReaderR
 
 func (h *TestTopicListener_Handler) OnStartPartitionSessionRequest(
 	ctx context.Context,
-	event *topiclistener.StartPartitionSessionRequest,
+	event *topiclistener.StartPartitionSessionEvent,
 ) error {
 	h.onPartitionStart = event
 	event.Confirm()
@@ -72,7 +72,7 @@ func (h *TestTopicListener_Handler) OnStartPartitionSessionRequest(
 
 func (h *TestTopicListener_Handler) OnStopPartitionSessionRequest(
 	ctx context.Context,
-	event *topiclistener.StopPartitionSessionRequest,
+	event *topiclistener.StopPartitionSessionEvent,
 ) error {
 	h.onPartitionStop = event
 	event.Confirm()

topic/topiclistener/event_handler.go

@@ -47,7 +47,7 @@ func (b BaseHandler) OnReaderCreated(event *ReaderReady) error {
 
 func (b BaseHandler) OnStartPartitionSessionRequest(
 	ctx context.Context,
-	event *StartPartitionSessionRequest,
+	event *StartPartitionSessionEvent,
 ) error {
 	event.Confirm()
 
@@ -61,7 +61,7 @@ func (b BaseHandler) OnStartPartitionSessionRequest(
 // not
 func (b BaseHandler) OnStopPartitionSessionRequest(
 	ctx context.Context,
-	event *StopPartitionSessionRequest,
+	event *StopPartitionSessionEvent,
 ) error {
 	event.Confirm()
 
@@ -76,10 +76,8 @@ func (b BaseHandler) OnReadMessages(
 }
 
 type (
-	StartPartitionSessionRequest = topiclistenerinternal.PublicStartPartitionSessionEvent
-	StartPartitionSessionConfirm = topiclistenerinternal.PublicStartPartitionSessionConfirm
-	StopPartitionSessionRequest  = topiclistenerinternal.PublicStopPartitionSessionEvent
-	StopPartitionSessionConfirm  = topiclistenerinternal.PublicStopPartitionSessionConfirm
+	StartPartitionSessionEvent = topiclistenerinternal.PublicStartPartitionSessionEvent
+	StopPartitionSessionEvent  = topiclistenerinternal.PublicStopPartitionSessionEvent
 )
 
 type PartitionSession = topicreadercommon.PublicPartitionSession