Function TestReconnectToServerCallback

test/reconnect_test.go:1509–1729

(t *testing.T)

}

func TestReconnectToServerCallback(t *testing.T) {
	t.Run("select custom server", func(t *testing.T) {
		srv1 := RunServerOnPort(-1)
		defer srv1.Shutdown()
		srv2 := RunServerOnPort(-1)
		defer srv2.Shutdown()

		url1 := srv1.ClientURL()
		url2 := srv2.ClientURL()

		var callbackCalled atomic.Int32
		reconnectCh := make(chan bool, 1)

		nc, err := nats.Connect(
			url1+","+url2,
			nats.ReconnectWait(100*time.Millisecond),
			nats.MaxReconnects(10),
			nats.DontRandomize(),
			nats.ReconnectToServer(func(servers []nats.Server, info nats.ServerInfo) (*nats.Server, time.Duration) {
				callbackCalled.Add(1)
				for i := range servers {
					if servers[i].URL.String() == url2 {
						return &servers[i], 0
					}
				}
				return &servers[0], 0
			}),
			nats.ReconnectHandler(func(_ *nats.Conn) {
				select {
				case reconnectCh <- true:
				default:
				}
			}),
		)
		if err != nil {
			t.Fatalf("Failed to connect: %v", err)
		}
		defer nc.Close()

		if nc.ConnectedUrl() != url1 {
			t.Fatalf("Expected initial connection to %s, got %s", url1, nc.ConnectedUrl())
		}

		// try reconnecting a few times, we should end up on url2 each time
		for i := range 5 {
			if err := nc.ForceReconnect(); err != nil {
				t.Fatalf("Failed to force reconnect: %v", err)
			}

			select {
			case <-reconnectCh:
			case <-time.After(2 * time.Second):
				t.Fatal("Timed out waiting for reconnect")
			}

			if callbackCalled.Load() != int32(i+1) {
				t.Fatal("Expected ReconnectToServer callback to be called on each reconnect")
			}