Function testJetStream_ClusterReconnectDurablePushSubscriber

test/js_test.go:7525–7636

(t *testing.T, subject string, srvs ...*jsServer)

}

func testJetStream_ClusterReconnectDurablePushSubscriber(t *testing.T, subject string, srvs ...*jsServer) {
	var (
		srvA          = srvs[0]
		srvB          = srvs[1]
		srvC          = srvs[2]
		totalMsgs     = 20
		reconnected   = make(chan struct{})
		reconnectDone bool
	)
	nc, err := nats.Connect(srvA.ClientURL(),
		nats.ReconnectHandler(func(nc *nats.Conn) {
			reconnected <- struct{}{}

			// Bring back the server after the reconnect event.
			if !reconnectDone {
				reconnectDone = true
				srvA.Restart()
			}
		}),
	)
	if err != nil {
		t.Error(err)
	}

	// Drain to allow Ack responses to be published.
	defer nc.Drain()

	js, err := nc.JetStream()
	if err != nil {
		t.Error(err)
	}

	// Initial burst of messages.
	for i := 0; i < 10; i++ {
		payload := fmt.Sprintf("i:%d", i)
		js.Publish(subject, []byte(payload))
	}

	// For now just confirm that do receive all messages across restarts.
	ctx, done := context.WithTimeout(context.Background(), 10*time.Second)
	defer done()

	recvd := make(chan *nats.Msg, totalMsgs)
	expected := totalMsgs
	_, err = js.Subscribe(subject, func(m *nats.Msg) {
		recvd <- m

		if len(recvd) == expected {
			done()
		}
	}, nats.Durable("sd1"))
	if err != nil {
		t.Errorf("Unexpected error: %v", err)
	}

	timeout := time.Now().Add(3 * time.Second)
	for time.Now().Before(timeout) {
		if len(recvd) >= 2 {