Fix subscribe tests
@@ -785,13 +785,13 @@ func TestSubscribePartitionDeleted(t *testing.T) {
require.NoError(t, err)
req := &proto.SubscribeRequest{StartPosition: proto.StartPosition_NEW_ONLY}
- _, statusCh, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req, make(chan struct{}))
+ sub, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req)
require.Nil(t, status)
require.NoError(t, stream.Delete())
select {
- case status := <-statusCh:
+ case status := <-sub.Errors():
require.Equal(t, codes.NotFound, status.Code())
case <-time.After(5 * time.Second):
t.Fatal("Did not receive expected status")
@@ -819,14 +819,14 @@ func TestSubscribePartitionPaused(t *testing.T) {
require.NoError(t, err)
req := &proto.SubscribeRequest{StartPosition: proto.StartPosition_NEW_ONLY}
- _, statusCh, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req, make(chan struct{}))
+ sub, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req)
require.Nil(t, status)
_, err = stream.Pause(nil, true)
require.NoError(t, err)
select {
- case status := <-statusCh:
+ case status := <-sub.Errors():
require.Equal(t, codes.FailedPrecondition, status.Code())
case <-time.After(5 * time.Second):
t.Fatal("Did not receive expected status")
@@ -853,13 +853,13 @@ func TestSubscribePartitionClosed(t *testing.T) {
require.NoError(t, err)
req := &proto.SubscribeRequest{StartPosition: proto.StartPosition_NEW_ONLY}
- _, statusCh, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req, make(chan struct{}))
+ sub, status := api.subscribe(context.Background(), stream.GetPartitions()[0], req)
require.Nil(t, status)
require.NoError(t, stream.Close())
select {
- case status := <-statusCh:
+ case status := <-sub.Errors():
require.Equal(t, codes.Internal, status.Code())
case <-time.After(5 * time.Second):
t.Fatal("Did not receive expected status")
