Function TestExtendedReconnectFunctionality

test/reconnect_test.go:225–315

(t *testing.T)

}

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

	opts := reconnectOpts
	dch := make(chan bool, 2)
	opts.DisconnectedErrCB = func(_ *nats.Conn, _ error) {
		dch <- true
	}
	rch := make(chan bool, 1)
	opts.ReconnectedCB = func(_ *nats.Conn) {
		rch <- true
	}
	nc, err := opts.Connect()
	if err != nil {
		t.Fatalf("Should have connected ok: %v", err)
	}
	defer nc.Close()

	testString := "bar"
	received := int32(0)

	nc.Subscribe("foo", func(*nats.Msg) {
		atomic.AddInt32(&received, 1)
	})

	sub, _ := nc.Subscribe("foobar", func(*nats.Msg) {
		atomic.AddInt32(&received, 1)
	})

	nc.Publish("foo", []byte(testString))
	nc.Flush()

	ts.Shutdown()
	// server is stopped here..

	// wait for disconnect
	if e := WaitTime(dch, 2*time.Second); e != nil {
		t.Fatal("Did not receive a disconnect callback message")
	}

	// Sub while disconnected
	nc.Subscribe("bar", func(*nats.Msg) {
		atomic.AddInt32(&received, 1)
	})

	// Unsub foobar while disconnected
	sub.Unsubscribe()

	if err = nc.Publish("foo", []byte(testString)); err != nil {
		t.Fatalf("Received an error after disconnect: %v\n", err)
	}

	if err = nc.Publish("bar", []byte(testString)); err != nil {
		t.Fatalf("Received an error after disconnect: %v\n", err)
	}

	ts = startReconnectServer(t)