Function TestQueueSubsOnReconnect

test/reconnect_test.go:317–401

(t *testing.T)

}

func TestQueueSubsOnReconnect(t *testing.T) {
	ts := startReconnectServer(t)
	defer ts.Shutdown()

	opts := reconnectOpts

	// Allow us to block on reconnect complete.
	reconnectsDone := make(chan bool)
	opts.ReconnectedCB = func(nc *nats.Conn) {
		reconnectsDone <- true
	}

	// Create connection
	nc, err := opts.Connect()
	if err != nil {
		t.Fatalf("Should have connected ok: %v\n", err)
	}
	defer nc.Close()

	// To hold results.
	results := make(map[int]int)
	var mu sync.Mutex

	// Make sure we got what we needed, 1 msg only and all seqnos accounted for..
	checkResults := func(numSent int) {
		mu.Lock()
		defer mu.Unlock()

		for i := 0; i < numSent; i++ {
			if results[i] != 1 {
				t.Fatalf("Received incorrect number of messages, [%d] for seq: %d\n", results[i], i)
			}
		}

		// Auto reset results map
		results = make(map[int]int)
	}

	subj := "foo.bar"
	qgroup := "workers"

	cb := func(m *nats.Msg) {
		mu.Lock()
		defer mu.Unlock()
		seqno, err := strconv.Atoi(string(m.Data))
		if err != nil {
			t.Fatalf("Received an invalid sequence number: %v\n", err)
		}
		results[seqno] = results[seqno] + 1
	}

	// Create Queue Subscribers
	nc.QueueSubscribe(subj, qgroup, cb)
	nc.QueueSubscribe(subj, qgroup, cb)

	nc.Flush()

	// Helper function to send messages and check results.