Function TestForceReconnect

test/reconnect_test.go:887–966

(t *testing.T)

}

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

	nc, err := nats.Connect(s.ClientURL(), nats.ReconnectWait(10*time.Second))
	if err != nil {
		t.Fatalf("Unexpected error on connect: %v", err)
	}

	statusCh := nc.StatusChanged(nats.RECONNECTING, nats.CONNECTED)
	defer close(statusCh)
	newStatus := make(chan nats.Status, 10)
	// non-blocking channel, so we need to be constantly listening
	go func() {
		for {
			s, ok := <-statusCh
			if !ok {
				return
			}
			newStatus <- s
		}
	}()

	sub, err := nc.SubscribeSync("foo")
	if err != nil {
		t.Fatalf("Error on subscribe: %v", err)
	}
	if err := nc.Publish("foo", []byte("msg")); err != nil {
		t.Fatalf("Error on publish: %v", err)
	}
	_, err = sub.NextMsg(time.Second)
	if err != nil {
		t.Fatalf("Error getting message: %v", err)
	}

	// Force a reconnect
	err = nc.ForceReconnect()
	if err != nil {
		t.Fatalf("Unexpected error on reconnect: %v", err)
	}

	WaitOnChannel(t, newStatus, nats.RECONNECTING)
	WaitOnChannel(t, newStatus, nats.CONNECTED)

	if err := nc.Publish("foo", []byte("msg")); err != nil {
		t.Fatalf("Error on publish: %v", err)
	}
	_, err = sub.NextMsg(time.Second)
	if err != nil {
		t.Fatalf("Error getting message: %v", err)
	}

	// shutdown server and then force a reconnect
	s.Shutdown()
	WaitOnChannel(t, newStatus, nats.RECONNECTING)
	_, err = sub.NextMsg(100 * time.Millisecond)
	if err == nil {
		t.Fatal("Expected error getting message")
	}