successful web socket Igtimi connection test with timestamp and heartbeat

java/com.sap.sailing.domain.igtimiadapter.test/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketTest.java

@@ -2,6 +2,7 @@ package com.sap.sailing.domain.igtimiadapter.websocket;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.net.URI;
@@ -13,12 +14,14 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
 import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
@@ -79,6 +82,11 @@ public class WebSocketTest {
             session.close(StatusCode.NORMAL, "I'm done");
         }
         
+        @OnWebSocketError
+        public void onError(Throwable cause) {
+            logger.log(Level.SEVERE, "Error with web socket connection", cause);
+        }
+        
         @OnWebSocketMessage
         public void onMessage(String msg) {
             logger.info("Received "+msg);
@@ -96,7 +104,7 @@ public class WebSocketTest {
     
     @Test
     public void simpleWebSockeEchoTest() throws Exception {
-        String destUri = "ws://echo.websocket.org";
+        String destUri = "ws://echo.websocket.org"; // wss currently doesn't seem to work with Jetty 9.0.4 WebSocket implementation
         WebSocketClient client = new WebSocketClient();
         SimpleEchoTestSocket socket = new SimpleEchoTestSocket();
         try {
@@ -137,7 +145,7 @@ public class WebSocketTest {
         Account account = connectionFactory.registerAccountForWhichClientIsAuthorized("3b6cbd0522423bb1ac274ddb9e7e579c4b3be6667622271086c4fdbf30634ba9");
         WebSocketConnectionManager manager = new WebSocketConnectionManager(connectionFactory, Collections.singleton("GA-EN-AAEJ"), account);
         assertNotNull(manager);
-        Thread.sleep(1000000);
+        assertTrue("Connection handshake not successful within 5s", manager.waitForConnection(5000));
         manager.disconnect();
     }
 }

java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/IgtimiConnectionFactory.java

@@ -80,9 +80,8 @@ public interface IgtimiConnectionFactory {
      * 
      * @param account
      *            represents this factory's client's permissions to access a user account's data
-     * @param unitIds
+     * @param deviceIds
      *            IDs of the transmitting units expected to be visible to the account's {@link Account#getUser() user's}
-     *            .
      */
-    JSONObject getWebSocketConfigurationMessage(Account account, Iterable<String> unitIds);
+    JSONObject getWebSocketConfigurationMessage(Account account, Iterable<String> deviceIds);
 }

java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/FixFactory.java

@@ -34,7 +34,7 @@ public class FixFactory {
                 JSONArray timePointsMillis = (JSONArray) fixesJson.get("t");
                 int fixIndex = 0;
                 for (Object timePointMillis : timePointsMillis) {
-                    TimePoint timePoint = new MillisecondsTimePoint((Long) timePointMillis);
+                    TimePoint timePoint = new MillisecondsTimePoint(((Number) timePointMillis).longValue());
                     Map<Integer, Object> valuesPerSubindex = new HashMap<>();
                     int i=1;
                     JSONArray values;

java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/IgtimiConnectionFactoryImpl.java

@@ -128,10 +128,10 @@ public class IgtimiConnectionFactoryImpl implements IgtimiConnectionFactory {
     public JSONObject getWebSocketConfigurationMessage(Account account, Iterable<String> deviceIds) {
         JSONObject result = new JSONObject();
         result.put("access_token", getAccessTokenForAccount(account));
-        JSONArray unitIdsJson = new JSONArray();
-        result.put("devices", unitIdsJson);
-        for (String unitId : deviceIds) {
-            unitIdsJson.add(unitId);
+        JSONArray deviceIdsJson = new JSONArray();
+        result.put("devices", deviceIdsJson);
+        for (String deviceId : deviceIds) {
+            deviceIdsJson.add(deviceId);
         }
         return result;
     }

java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketConnectionManager.java

@@ -40,7 +40,7 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
     private final ClientUpgradeRequest request;
     private final JSONObject configurationMessage;
     private final Timer timer;
-    private final Iterable<String> unitIds;
+    private final Iterable<String> deviceIds;
     private final Account account;
     private final FixFactory fixFactory;
     private boolean receivedServerHeartbeatInInterval;
@@ -48,20 +48,39 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
     private TimePoint igtimiServerTimepoint;
     private TimePoint localTimepointWhenServerTimepointWasReceived;
     
-    public WebSocketConnectionManager(IgtimiConnectionFactory connectionFactory, Iterable<String> unitIds, Account account) throws Exception {
-        this.timer = new Timer("Timer for WebSocketConnectionManager for units "+unitIds+" and account "+account);
-        this.unitIds = unitIds;
+    public WebSocketConnectionManager(IgtimiConnectionFactory connectionFactory, Iterable<String> deviceIds, Account account) throws Exception {
+        this.timer = new Timer("Timer for WebSocketConnectionManager for units "+deviceIds+" and account "+account);
+        this.deviceIds = deviceIds;
         this.account = account;
         this.fixFactory = new FixFactory();
         this.connectionFactory = connectionFactory;
         this.liveDataListeners = new ConcurrentSkipListSet<>();
         client = new WebSocketClient();
-        configurationMessage = connectionFactory.getWebSocketConfigurationMessage(account, unitIds);
+        configurationMessage = connectionFactory.getWebSocketConfigurationMessage(account, deviceIds);
         request = new ClientUpgradeRequest();
         client.start();
         reconnect();
     }
     
+    /**
+     * Waits until the web socket connection has been established and the initial configuration handshake has
+     * successfully completed. Technically, this means that the Igtimi server timestamp has successfully been received
+     * and parsed. {@link #getIgtimiServerTimePointAndWhenItWasReceived()} then holds a valid time point.
+     * 
+     * @param timeoutInMillis
+     *            use 0 to wait indefinitely
+     * @return <code>true</code> if the connection is established before the timeout occurred
+     */
+    public boolean waitForConnection(long timeoutInMillis) throws InterruptedException {
+        long startedToWait = System.currentTimeMillis();
+        synchronized (this) {
+            while (igtimiServerTimepoint == null && System.currentTimeMillis() - startedToWait < timeoutInMillis) {
+                wait(timeoutInMillis);
+            }
+            return igtimiServerTimepoint != null;
+        }
+    }
+    
     public void addLiveDataListener(LiveDataListener listener) {
         liveDataListeners.add(listener);
     }
@@ -72,7 +91,9 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
     }
 
     @Override
-    public void onWebSocketConnect(Session sess) {
+    public void onWebSocketConnect(Session session) {
+        super.onWebSocketConnect(session);
+        logger.info("received connection "+session+" for "+this);
         try {
             getRemote().sendString(configurationMessage.toString());
             startClientHeartbeat();
@@ -85,7 +106,11 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
 
     @Override
     public void onWebSocketText(String message) {
+        if (logger.isLoggable(Level.FINEST)) {
+            logger.finest("Received "+message+" in "+this);
+        }
         if (message.equals("1")) {
+            logger.fine("Received server heartbeat");
             receivedServerHeartbeatInInterval = true;
         } else if (message.startsWith("[")) {
             List<Fix> fixes = new ArrayList<>();
@@ -94,14 +119,21 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
                 for (Object o : jsonArray) {
                     Util.addAll(fixFactory.createFixes((JSONObject) o), fixes);
                 }
+                if (logger.isLoggable(Level.FINEST)) {
+                    logger.finest("Received fixes"+fixes);
+                }
                 notifyListeners(fixes);
             } catch (ParseException e) {
                 logger.log(Level.SEVERE, "Error trying to parse a web socket data package coming from Igtimi "+this, e);
             }
         } else {
             // try to parse server time stamp in response to the configuration message
-            igtimiServerTimepoint = new MillisecondsTimePoint(Long.valueOf(message));
-            localTimepointWhenServerTimepointWasReceived = MillisecondsTimePoint.now();
+            synchronized (this) {
+                igtimiServerTimepoint = new MillisecondsTimePoint(Long.valueOf(message));
+                localTimepointWhenServerTimepointWasReceived = MillisecondsTimePoint.now();
+                logger.info("Received server timestamp "+igtimiServerTimepoint);
+                notifyAll();
+            }
         }
     }
     
@@ -133,7 +165,7 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
 
     @Override
     public String toString() {
-        return "Web Socket Connection Manager for units "+unitIds+" and account "+account+" with web socket session "+getSession();
+        return "Web Socket Connection Manager for devices "+deviceIds+" and account "+account+" with web socket session "+getSession();
     }
 
     private void startListeningForServerHeartbeat() {
@@ -157,6 +189,7 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
             @Override
             public void run() {
                 try {
+                    logger.fine("Sending client heartbeat");
                     getRemote().sendString("1");
                 } catch (IOException e) {
                     logger.log(Level.WARNING, "Couldn't send heartbeat to Igtimi web socket session in "+this, e);
@@ -169,11 +202,14 @@ public class WebSocketConnectionManager extends WebSocketAdapter {
         IOException lastException = null;
         for (URI uri : connectionFactory.getWebsocketServers()) {
             try {
-                logger.log(Level.INFO, "Trying to connect to "+uri+" for "+this);
-                client.connect(this, uri, request);
-                logger.log(Level.INFO, "Successfully connected to "+uri+" for "+this);
-                lastException = null;
-                break; // successfully connected
+                if (uri.getScheme().equals("ws")) { // as Jetty 9.0.4 currently doesn't seem to support wss, explicitly
+                                                    // look for ws connectivity
+                    logger.log(Level.INFO, "Trying to connect to " + uri + " for " + this);
+                    client.connect(this, uri, request);
+                    logger.log(Level.INFO, "Successfully connected to " + uri + " for " + this);
+                    lastException = null;
+                    break; // successfully connected
+                }
             } catch (IOException e) {
                 logger.log(Level.INFO, "Couldn't connect to "+uri+" for "+this, e);
                 lastException = e;