fix(fxgcppubsub): Fixed nack reactor (#24)

* fix(fxgcppubsub): Fixed nack reactor

* fix(fxgcppubsub): Fixed nack reactor

* fix(fxgcppubsub): Fixed nack reactor

* fix(fxgcppubsub): Fixed nack reactor

* fix(fxgcppubsub): Fixed nack reactor

Author: ekkinox

fxgcppubsub/reactor/ack/reactor.go

@@ -30,8 +30,8 @@ func (r *AckReactor) React(req any) (bool, any, error) {
 		r.supervisor.StopAckWaiter(ackReq.Subscription, ackReq.AckIds, nil)
 	}
 
-	if ackReq, ok := req.(*pubsubpb.ModifyAckDeadlineRequest); ok {
-		r.supervisor.StopNackWaiter(ackReq.Subscription, ackReq.AckIds, nil)
+	if modReq, ok := req.(*pubsubpb.ModifyAckDeadlineRequest); ok {
+		r.supervisor.StopNackWaiter(modReq.Subscription, modReq.AckIds, nil)
 	}
 
 	return false, nil, nil

fxgcppubsub/reactor/ack/reactor_test.go

@@ -62,7 +62,7 @@ func TestAckReactor(t *testing.T) {
 			assert.NoError(t, rErr)
 		}()
 
-		data, err := waiter.WaitMaxDuration(context.Background(), 1*time.Millisecond)
+		data, err := waiter.WaitMaxDuration(context.Background(), 5*time.Millisecond)
 		assert.NoError(t, err)
 		assert.Equal(t, []string{"test-id"}, data)
 	})
@@ -83,7 +83,7 @@ func TestAckReactor(t *testing.T) {
 			assert.NoError(t, rErr)
 		}()
 
-		data, err := waiter.WaitMaxDuration(context.Background(), 1*time.Millisecond)
+		data, err := waiter.WaitMaxDuration(context.Background(), 5*time.Millisecond)
 		assert.NoError(t, err)
 		assert.Equal(t, []string{"test-id"}, data)
 	})

fxgcppubsub/reactor/supervisor.go

@@ -44,5 +44,7 @@ func (s *DefaultWaiterSupervisor) StopWaiter(target string, data any, err error)
 
 	if waiter, found := s.waiters[target]; found {
 		waiter.Stop(data, err)
+
+		delete(s.waiters, target)
 	}
 }