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 b6450845c36..bfcb1b879e9 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 @@ -1,8 +1,14 @@ package com.sap.sailing.domain.igtimiadapter.websocket; +import static org.junit.Assert.assertNotNull; + +import java.util.Collections; + import org.junit.Test; +import com.sap.sailing.domain.igtimiadapter.Account; import com.sap.sailing.domain.igtimiadapter.Client; +import com.sap.sailing.domain.igtimiadapter.IgtimiConnectionFactory; import com.sap.sailing.domain.igtimiadapter.impl.ClientImpl; import com.sap.sailing.domain.igtimiadapter.impl.IgtimiConnectionFactoryImpl; @@ -11,7 +17,10 @@ public class WebSocketTest { public void testWebSocketConnect() throws Exception { final Client client = new ClientImpl("7fcdd217e0aa16090edb4ad55b09ec43b2021090e209541fc9b7003c2a2b70c6", "aa569cf4909bdc7b0e04b11873f3c4ea20687421e010fcc25b771cca9e6f3f9a", "http://127.0.0.1:8888/igtimi/oauth/v1/authorizationcallback"); - WebSocketConnectionManager manager = new WebSocketConnectionManager(new IgtimiConnectionFactoryImpl(client)); - manager.connect(); + final IgtimiConnectionFactory connectionFactory = new IgtimiConnectionFactoryImpl(client); + Account account = connectionFactory.registerAccountForWhichClientIsAuthorized("3b6cbd0522423bb1ac274ddb9e7e579c4b3be6667622271086c4fdbf30634ba9"); + WebSocketConnectionManager manager = new WebSocketConnectionManager(connectionFactory, Collections.singleton("GA-EN-AAEJ"), account); + assertNotNull(manager); + manager.disconnect(); } } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/Account.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/Account.java index 17bbe05b9be..ecd7bfa77ba 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/Account.java +++ b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/Account.java @@ -1,5 +1,8 @@ package com.sap.sailing.domain.igtimiadapter; +/** + * Represents the authorization of a {@link Client} to access the data of some {@link #getUser}. + */ public interface Account { User getUser(); } 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 bc142e66a6b..9502c68d70f 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 @@ -8,6 +8,7 @@ import java.util.Map; import org.apache.http.client.ClientProtocolException; import org.apache.http.client.HttpClient; +import org.json.simple.JSONObject; import org.json.simple.parser.ParseException; import com.sap.sailing.domain.common.TimePoint; @@ -71,4 +72,17 @@ public interface IgtimiConnectionFactory { Iterable getWebsocketServers() throws IllegalStateException, ClientProtocolException, IOException, ParseException, URISyntaxException; HttpClient getHttpClient(); + + /** + * Retrieves the JSON object to send in its string-serialized form to a web socket connection in order to receive + * live data from the units whose IDs are specified by unitIds. The sending units are expected to + * belong to the user account to which this factory's application client has been granted permission. + * + * @param account + * represents this factory's client's permissions to access a user account's data + * @param unitIds + * IDs of the transmitting units expected to be visible to the account's {@link Account#getUser() user's} + * . + */ + JSONObject getWebSocketConfigurationMessage(Account account, Iterable unitIds); } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/LiveDataListener.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/LiveDataListener.java new file mode 100755 index 00000000000..6840a7f5cff --- /dev/null +++ b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/LiveDataListener.java @@ -0,0 +1,7 @@ +package com.sap.sailing.domain.igtimiadapter; + +import com.sap.sailing.domain.igtimiadapter.datatypes.Fix; + +public interface LiveDataListener { + void fixesReceived(Iterable fixes); +} diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/AccountImpl.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/AccountImpl.java index 7fea4d059d1..bb53a1fff94 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/AccountImpl.java +++ b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/AccountImpl.java @@ -16,4 +16,8 @@ public class AccountImpl implements Account { return user; } + @Override + public String toString() { + return getUser().toString(); + } } 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 45f020cefc7..b6dc2c75845 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 @@ -124,6 +124,18 @@ public class IgtimiConnectionFactoryImpl implements IgtimiConnectionFactory { return getApiV1BaseUrl()+"account?"+getAccessTokenUrlParameter(account); } + @Override + public JSONObject getWebSocketConfigurationMessage(Account account, Iterable unitIds) { + JSONObject result = new JSONObject(); + result.put("auth_token", getAccessTokenForAccount(account)); + JSONArray unitIdsJson = new JSONArray(); + result.put("units", unitIdsJson); + for (String unitId : unitIds) { + unitIdsJson.add(unitId); + } + return result; + } + @Override public String getUsersUrl(Account account) { return getApiV1BaseUrl()+"users?"+getAccessTokenUrlParameter(account); diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/UserImpl.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/UserImpl.java index b40560dde03..4770ef7d8c6 100755 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/UserImpl.java +++ b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/impl/UserImpl.java @@ -36,5 +36,10 @@ public class UserImpl implements User { public String getEmail() { return email; } + + @Override + public String toString() { + return getFirstName()+" "+getSurname()+" <"+getEmail()+">, ID "+getId(); + } } diff --git a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/LiveDataWebSocket.java b/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/LiveDataWebSocket.java deleted file mode 100755 index 37bc552c245..00000000000 --- a/java/com.sap.sailing.domain.igtimiadapter/src/com/sap/sailing/domain/igtimiadapter/websocket/LiveDataWebSocket.java +++ /dev/null @@ -1,38 +0,0 @@ -package com.sap.sailing.domain.igtimiadapter.websocket; - -import org.eclipse.jetty.websocket.api.Session; -import org.eclipse.jetty.websocket.api.WebSocketListener; -import org.eclipse.jetty.websocket.api.annotations.WebSocket; - -@WebSocket -public class LiveDataWebSocket implements WebSocketListener { - @Override - public void onWebSocketBinary(byte[] payload, int offset, int len) { - // TODO Auto-generated method stub - - } - - @Override - public void onWebSocketClose(int statusCode, String reason) { - // TODO Auto-generated method stub - - } - - @Override - public void onWebSocketConnect(Session session) { - // TODO Auto-generated method stub - - } - - @Override - public void onWebSocketError(Throwable cause) { - // TODO Auto-generated method stub - - } - - @Override - public void onWebSocketText(String message) { - // TODO Auto-generated method stub - - } -} 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 f332b1dc73f..88555987d59 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 @@ -2,44 +2,185 @@ package com.sap.sailing.domain.igtimiadapter.websocket; import java.io.IOException; import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.logging.Level; import java.util.logging.Logger; +import org.eclipse.jetty.websocket.api.Session; +import org.eclipse.jetty.websocket.api.WebSocketAdapter; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; import org.eclipse.jetty.websocket.client.WebSocketClient; +import org.json.simple.JSONArray; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; +import org.json.simple.parser.ParseException; +import com.sap.sailing.domain.common.TimePoint; +import com.sap.sailing.domain.common.impl.MillisecondsTimePoint; +import com.sap.sailing.domain.common.impl.Util; +import com.sap.sailing.domain.common.impl.Util.Pair; +import com.sap.sailing.domain.igtimiadapter.Account; import com.sap.sailing.domain.igtimiadapter.IgtimiConnectionFactory; +import com.sap.sailing.domain.igtimiadapter.LiveDataListener; +import com.sap.sailing.domain.igtimiadapter.datatypes.Fix; +import com.sap.sailing.domain.igtimiadapter.impl.FixFactory; -public class WebSocketConnectionManager { +public class WebSocketConnectionManager extends WebSocketAdapter { private static final Logger logger = Logger.getLogger(WebSocketConnectionManager.class.getName()); private final IgtimiConnectionFactory connectionFactory; + private static enum TargetState { OPEN, CLOSED }; + private TargetState targetState; + private final WebSocketClient client; + private final ClientUpgradeRequest request; + private final JSONObject configurationMessage; + private final Timer timer; + private final Iterable unitIds; + private final Account account; + private final FixFactory fixFactory; + private boolean receivedServerHeartbeatInInterval; + private final Set liveDataListeners; + private TimePoint igtimiServerTimepoint; + private TimePoint localTimepointWhenServerTimepointWasReceived; - public WebSocketConnectionManager(IgtimiConnectionFactory connectionFactory) { + 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; + this.account = account; + this.fixFactory = new FixFactory(); this.connectionFactory = connectionFactory; + this.liveDataListeners = new ConcurrentSkipListSet<>(); + client = new WebSocketClient(); + configurationMessage = connectionFactory.getWebSocketConfigurationMessage(account, unitIds); + request = new ClientUpgradeRequest(); + client.start(); + reconnect(); + } + + public void addLiveDataListener(LiveDataListener listener) { + liveDataListeners.add(listener); + } + + public void disconnect() throws Exception { + targetState = TargetState.CLOSED; + client.stop(); } - public void connect() throws Exception { - WebSocketClient client = new WebSocketClient(); - client.start(); - LiveDataWebSocket socket = new LiveDataWebSocket(); - ClientUpgradeRequest request = new ClientUpgradeRequest(); + @Override + public void onWebSocketConnect(Session sess) { + try { + getRemote().sendString(configurationMessage.toString()); + startClientHeartbeat(); + startListeningForServerHeartbeat(); + } catch (IOException e) { + logger.log(Level.SEVERE, "Could not send configuration package to Igtimi web socket server in "+this, e); + throw new RuntimeException(e); + } + } + + @Override + public void onWebSocketText(String message) { + if (message.equals("1")) { + receivedServerHeartbeatInInterval = true; + } else if (message.startsWith("[")) { + List fixes = new ArrayList<>(); + try { + JSONArray jsonArray = (JSONArray) new JSONParser().parse(message); + for (Object o : jsonArray) { + Util.addAll(fixFactory.createFixes((JSONObject) o), 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(); + } + } + + public Pair getIgtimiServerTimePointAndWhenItWasReceived() { + return new Pair(igtimiServerTimepoint, localTimepointWhenServerTimepointWasReceived); + } + + private void notifyListeners(List fixes) { + for (LiveDataListener listener : liveDataListeners) { + listener.fixesReceived(fixes); + } + } + + @Override + public void onWebSocketClose(int statusCode, String reason) { + if (targetState == TargetState.OPEN) { + try { + reconnect(); + } catch (Exception e) { + logger.log(Level.SEVERE, "Couldn't reconnect to Igtimi web socket in "+this, e); + } + } + } + + @Override + public void onWebSocketError(Throwable cause) { + logger.log(Level.SEVERE, "Error trying to open Igtimi web socket in "+this, cause); + } + + @Override + public String toString() { + return "Web Socket Connection Manager for units "+unitIds+" and account "+account+" with web socket session "+getSession(); + } + + private void startListeningForServerHeartbeat() { + timer.scheduleAtFixedRate(new TimerTask() { + @Override + public void run() { + try { + if (!receivedServerHeartbeatInInterval) { + logger.info("Didn't receive server heartbeat in interval for "+this+". Reconnecting..."); + reconnect(); + } + } catch (Exception e) { + logger.log(Level.WARNING, "Error with server heartbeat in "+this, e); + } + } + }, /* wait for 45s for first execution */ 45000l, /* and check again every 45s*/ 45000l); + } + + private void startClientHeartbeat() { + timer.scheduleAtFixedRate(new TimerTask() { + @Override + public void run() { + try { + getRemote().sendString("1"); + } catch (IOException e) { + logger.log(Level.WARNING, "Couldn't send heartbeat to Igtimi web socket session in "+this, e); + } + } + }, /* delay */ 0, /* send ping message every other minute; has to be at least every five minutes, so this should be safe */ 120000l); + } + + public void reconnect() throws IOException, IllegalStateException, ParseException, URISyntaxException { IOException lastException = null; for (URI uri : connectionFactory.getWebsocketServers()) { try { - logger.log(Level.INFO, "Trying to connect to "+uri); - client.connect(socket, uri, request); - logger.log(Level.INFO, "Successfully connected to "+uri); + 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, e); + logger.log(Level.INFO, "Couldn't connect to "+uri+" for "+this, e); lastException = e; } } if (lastException != null) { throw lastException; - } else { - - } + } } }