Function TestDrainSlowSubscriber

test/drain_test.go:196–237

(t *testing.T)

}

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

	received := int32(0)

	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
		atomic.AddInt32(&received, 1)
		time.Sleep(100 * time.Millisecond)
	})
	if err != nil {
		t.Fatalf("Error creating subscription; %v", err)
	}

	total := 10
	for i := 0; i < total; i++ {
		nc.Publish("foo", []byte("Slow Slow"))
	}
	nc.Flush()

	pmsgs, _, _ := sub.Pending()
	if pmsgs != total && pmsgs != total-1 {
		t.Fatalf("Expected most messages to be pending, but got %d vs %d", pmsgs, total)
	}
	sub.Drain()

	// Should take a second or so to drain away.
	waitFor(t, 2*time.Second, 100*time.Millisecond, func() error {
		// Wait for it to become invalid. Once drained it is unsubscribed.
		_, _, err := sub.Pending()
		if err != nats.ErrBadSubscription {
			return errors.New("Still valid")
		}
		r := int(atomic.LoadInt32(&received))
		if r != total {
			t.Fatalf("Did not receive all messages, got %d vs %d", r, total)
		}
		return nil
	})
}

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