Function TestDrainConnection

test/drain_test.go:239–340

(t *testing.T)

}

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

	done := make(chan bool)
	rdone := make(chan bool)

	closed := func(nc *nats.Conn) {
		done <- true
	}

	url := fmt.Sprintf("nats://127.0.0.1:%d", nats.DefaultPort)
	nc, err := nats.Connect(url, nats.ClosedHandler(closed))
	if err != nil {
		t.Fatalf("Failed to create default connection: %v", err)
	}
	defer nc.Close()

	nc2, err := nats.Connect(url)
	if err != nil {
		t.Fatalf("Failed to create default connection: %v", err)
	}
	defer nc2.Close()

	received := int32(0)
	responses := int32(0)
	expected := int32(50)
	sleep := 10 * time.Millisecond

	// Create the listener for responses on "bar"
	_, err = nc2.Subscribe("bar", func(_ *nats.Msg) {
		r := atomic.AddInt32(&responses, 1)
		if r == expected {
			rdone <- true
		}
	})
	if err != nil {
		t.Fatalf("Error creating subscription for responses: %v", err)
	}

	// Create a slow subscriber for the responder
	sub, err := nc.Subscribe("foo", func(m *nats.Msg) {
		time.Sleep(sleep)
		atomic.AddInt32(&received, 1)
		err := nc.Publish(m.Reply, []byte("Stop bugging me"))
		if err != nil {
			t.Errorf("Publisher received an error sending response: %v\n", err)
		}
	})
	if err != nil {
		t.Fatalf("Error creating subscription; %v", err)
	}

	// Publish some messages
	for i := int32(0); i < expected; i++ {
		nc.PublishRequest("foo", "bar", []byte("Slow Slow"))
	}