Function TestDrainUnSubs

test/drain_test.go:144–194

(t *testing.T)

}

func TestDrainUnSubs(t *testing.T) {
	s := RunDefaultServer()
	defer s.Shutdown()
	nc := NewDefaultConnection(t)
	defer nc.Close()

	num := 100
	subs := make([]*nats.Subscription, num)

	// Normal Unsubscribe
	for i := 0; i < num; i++ {
		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
		if err != nil {
			t.Fatalf("Error creating subscription; %v", err)
		}
		subs[i] = sub
	}

	if numSubs := nc.NumSubscriptions(); numSubs != num {
		t.Fatalf("Expected %d subscriptions, got %d", num, numSubs)
	}
	for i := 0; i < num; i++ {
		subs[i].Unsubscribe()
	}
	if numSubs := nc.NumSubscriptions(); numSubs != 0 {
		t.Fatalf("Expected no subscriptions, got %d", numSubs)
	}

	// Drain version
	for i := 0; i < num; i++ {
		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
		if err != nil {
			t.Fatalf("Error creating subscription; %v", err)
		}
		subs[i] = sub
	}

	if numSubs := nc.NumSubscriptions(); numSubs != num {
		t.Fatalf("Expected %d subscriptions, got %d", num, numSubs)
	}
	for i := 0; i < num; i++ {
		subs[i].Drain()
	}
	// Should happen quickly that we get to zero, so do not need to wait long.
	waitFor(t, 2*time.Second, 10*time.Millisecond, func() error {
		if numSubs := nc.NumSubscriptions(); numSubs != 0 {
			return fmt.Errorf("Expected no subscriptions, got %d", numSubs)
		}
		return nil
	})
}

func TestDrainSlowSubscriber(t *testing.T) {
	s := RunDefaultServer()