Merge pull request #122 from hyperledger/fix_delete_listeners

fix: deleting listeners from event stream pagination

Author: peterbroadhurst

pkg/fftm/stream_management.go

@@ -73,17 +73,16 @@ func (m *manager) restoreStreams() error {
 }
 
 func (m *manager) deleteAllStreamListeners(ctx context.Context, streamID *fftypes.UUID) error {
-	var lastInPage *fftypes.UUID
 	for {
-		listenerDefs, err := m.persistence.ListStreamListenersByCreateTime(ctx, lastInPage, startupPaginationLimit, txhandler.SortDirectionAscending, streamID)
+		// Do not specify after as we just delete everything
+		listenerDefs, err := m.persistence.ListStreamListenersByCreateTime(ctx, nil, startupPaginationLimit, txhandler.SortDirectionAscending, streamID)
 		if err != nil {
 			return err
 		}
 		if len(listenerDefs) == 0 {
 			break
 		}
 		for _, def := range listenerDefs {
-			lastInPage = def.ID
 			if err := m.persistence.DeleteListener(ctx, def.ID); err != nil {
 				return err
 			}

pkg/fftm/stream_management_test.go

@@ -194,6 +194,37 @@ func TestDeleteStartedListenerFail(t *testing.T) {
 	mp.AssertExpectations(t)
 }
 
+func TestDeleteStartedListenerWithPagination(t *testing.T) {
+
+	_, m, close := newTestManagerMockPersistence(t)
+	defer close()
+
+	esID := apitypes.NewULID()
+	lID := apitypes.NewULID()
+	secondID := apitypes.NewULID()
+	mp := m.persistence.(*persistencemocks.Persistence)
+	mp.On("ListStreamListenersByCreateTime", m.ctx, (*fftypes.UUID)(nil), startupPaginationLimit, txhandler.SortDirectionAscending, esID).Return(
+		[]*apitypes.Listener{
+			{ID: lID, StreamID: esID},
+			{ID: secondID, StreamID: esID},
+		}, nil).Once()
+	thirdID := apitypes.NewULID()
+	mp.On("ListStreamListenersByCreateTime", m.ctx, (*fftypes.UUID)(nil), startupPaginationLimit, txhandler.SortDirectionAscending, esID).Return(
+		[]*apitypes.Listener{
+			{ID: thirdID, StreamID: esID},
+		}, nil).Once()
+	mp.On("ListStreamListenersByCreateTime", m.ctx, (*fftypes.UUID)(nil), startupPaginationLimit, txhandler.SortDirectionAscending, esID).Return(
+		[]*apitypes.Listener{}, nil)
+	mp.On("DeleteListener", m.ctx, lID).Return(nil)
+	mp.On("DeleteListener", m.ctx, secondID).Return(nil)
+	mp.On("DeleteListener", m.ctx, thirdID).Return(nil)
+
+	err := m.deleteAllStreamListeners(m.ctx, esID)
+	assert.NoError(t, err)
+
+	mp.AssertExpectations(t)
+}
+
 func TestDeleteStreamBadID(t *testing.T) {
 
 	_, m, close := newTestManagerMockPersistence(t)