avoid duplicate connecting event, better Exception msg if tokenGetter not set

Author: FZambia

centrifuge/src/main/java/io/github/centrifugal/centrifuge/Client.java

@@ -185,9 +185,12 @@ void processDisconnect(int code, String reason, Boolean shouldReconnect) {
             this.reconnectTask.cancel(true);
             this.reconnectTask = null;
         }
+        boolean needEvent;
         if (shouldReconnect) {
+            needEvent = previousState != ClientState.CONNECTING;
             this.setState(ClientState.CONNECTING);
         } else {
+            needEvent = previousState != ClientState.DISCONNECTED;
             this.setState(ClientState.DISCONNECTED);
         }
 
@@ -212,12 +215,14 @@ void processDisconnect(int code, String reason, Boolean shouldReconnect) {
             }
         }
 
-        if (shouldReconnect) {
-            ConnectingEvent event = new ConnectingEvent(code, reason);
-            this.listener.onConnecting(this, event);
-        } else {
-            DisconnectedEvent event = new DisconnectedEvent(code, reason);
-            this.listener.onDisconnected(this, event);
+        if (needEvent) {
+            if (shouldReconnect) {
+                ConnectingEvent event = new ConnectingEvent(code, reason);
+                this.listener.onConnecting(this, event);
+            } else {
+                DisconnectedEvent event = new DisconnectedEvent(code, reason);
+                this.listener.onDisconnected(this, event);
+            }
         }
 
         this.ws.close(NORMAL_CLOSURE_STATUS, "");
@@ -335,12 +340,15 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
         });
     }
 
-    private void handleConnectionOpen() {
+    private void handleConnectionOpen() throws Exception {
         if (this.getState() != ClientState.CONNECTING) {
             return;
         }
         if (this.refreshRequired || (this.token == null && this.opts.getTokenGetter() != null)) {
             ConnectionTokenEvent connectionTokenEvent = new ConnectionTokenEvent();
+            if (this.opts.getTokenGetter() == null) {
+                throw new Exception("tokenGetter function should be provided in Client options to handle token refresh, see Options.setTokenGetter");
+            }
             this.opts.getTokenGetter().getConnectionToken(connectionTokenEvent, (err, token) -> this.executor.submit(() -> {
                 if (Client.this.getState() != ClientState.CONNECTING) {
                     return;
@@ -597,6 +605,7 @@ private void handleConnectReply(Protocol.Reply reply) {
                 sub.resubscribeIfNecessary();
             }
         }
+
         for (Map.Entry<String, Protocol.SubscribeResult> entry : result.getSubsMap().entrySet()) {
             Protocol.SubscribeResult subResult = entry.getValue();
             String channel = entry.getKey();
@@ -656,6 +665,7 @@ private void handleConnectReply(Protocol.Reply reply) {
                 }
             }
         }
+
         this.connectCommands.clear();
 
         for (Map.Entry<Integer, Protocol.Command> entry : this.connectAsyncCommands.entrySet()) {

example/src/main/java/io/github/centrifugal/centrifuge/example/Main.java

@@ -50,11 +50,11 @@ public void onConnecting(Client client, ConnectingEvent event) {
             }
             @Override
             public void onDisconnected(Client client, DisconnectedEvent event) {
-                System.out.printf("disconnected %d %s", event.getCode(), event.getReason());
+                System.out.printf("disconnected %d %s%n", event.getCode(), event.getReason());
             }
             @Override
             public void onError(Client client, ErrorEvent event) {
-                System.out.println("There was a problem connecting!");
+                System.out.printf("There was a problem connecting: %s%n", event.getError().toString());
             }
             @Override
             public void onMessage(Client client, MessageEvent event) {
@@ -89,7 +89,7 @@ public void onLeave(Client client, ServerLeaveEvent event) {
         };
 
         Options opts = new Options();
-//        opts.setToken("eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0c3VpdGVfand0In0.hPmHsVqvtY88PvK4EmJlcdwNuKFuy3BGaF7dMaKdPlw");
+//        opts.setToken("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyMSIsImV4cCI6MTY1OTQ0MTY4MiwiaWF0IjoxNjU4ODM2ODgyfQ.tyd2_TVq29WZuhh5OBni6dq7Lqry2s8z2PHZavplr7A");
 //        opts.setConnectionTokenGetter(new ConnectionTokenGetter() {
 //            @Override
 //            public void getConnectionToken(ConnectionTokenEvent event, TokenCallback cb) {
@@ -106,7 +106,7 @@ public void onLeave(Client client, ServerLeaveEvent event) {
 //        });
 
         Client client = new Client(
-                "ws://localhost:8000/connection/websocket?cf_protocol=protobuf",
+                "ws://localhost:8000/connection/websocket",
                 opts,
                 listener
         );