web socket framework that probably doesn't work yet

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();
     }
 }

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();
 }

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<URI> 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 <code>unitIds</code>. 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<String> unitIds);
 }

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<Fix> fixes);
+}

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();
+    }
 }

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<String> 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);

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();
+    }
 
 }

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

@@ -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
-        
-    }
-}

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<String> unitIds;
+    private final Account account;
+    private final FixFactory fixFactory;
+    private boolean receivedServerHeartbeatInInterval;
+    private final Set<LiveDataListener> liveDataListeners;
+    private TimePoint igtimiServerTimepoint;
+    private TimePoint localTimepointWhenServerTimepointWasReceived;
     
-    public WebSocketConnectionManager(IgtimiConnectionFactory connectionFactory) {
+    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;
+        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<Fix> 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<TimePoint, TimePoint> getIgtimiServerTimePointAndWhenItWasReceived() {
+        return new Pair<TimePoint, TimePoint>(igtimiServerTimepoint, localTimepointWhenServerTimepointWasReceived);
+    }
+    
+    private void notifyListeners(List<Fix> 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 {
-            
-        }
+        }        
     }
 }