From c5c86e91d465eaf1503a33dd3166fd7c6a01f331 Mon Sep 17 00:00:00 2001 From: Axel Uhl Date: Fri, 29 Nov 2013 11:07:41 +0100 Subject: [PATCH] successful web socket Igtimi connection test with timestamp and heartbeat --- .../websocket/WebSocketTest.java | 12 +++- .../IgtimiConnectionFactory.java | 5 +- .../domain/igtimiadapter/impl/FixFactory.java | 2 +- .../impl/IgtimiConnectionFactoryImpl.java | 8 +-- .../websocket/WebSocketConnectionManager.java | 64 +++++++++++++++---- 5 files changed, 67 insertions(+), 24 deletions(-) diff --git a/java/com.sap.sailing.domain.igtimiadapter.test/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketTest.java b/java/com.sap.sailing.domain.igtimiadapter.test/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketTest.java index ee096f05c69..cb88475a517 100755 --- a/java/com.sap.sailing.domain.igtimiadapter.test/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketTest.java +++ b/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(); } } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/IgtimiConnectionFactory.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/IgtimiConnectionFactory.java index 9502c68d70f..29edff72a58 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/IgtimiConnectionFactory.java +++ b/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 unitIds); + JSONObject getWebSocketConfigurationMessage(Account account, Iterable deviceIds); } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/FixFactory.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/FixFactory.java index 7a0e41706b4..bb280c28022 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/FixFactory.java +++ b/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 valuesPerSubindex = new HashMap<>(); int i=1; JSONArray values; diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/IgtimiConnectionFactoryImpl.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/IgtimiConnectionFactoryImpl.java index d9b1bb5c0c4..8abfb96b9db 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/IgtimiConnectionFactoryImpl.java +++ b/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 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; } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketConnectionManager.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketConnectionManager.java index 88555987d59..c1376cd34e1 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/WebSocketConnectionManager.java +++ b/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 unitIds; + private final Iterable 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 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 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 true 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 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;