1 files changed, 44 insertions, 22 deletions
diff --git a/sca-java-2.x/contrib/modules/binding-websocket/src/main/java/org/apache/tuscany/sca/binding/websocket/runtime/WebsocketReferenceInvoker.java b/sca-java-2.x/contrib/modules/binding-websocket/src/main/java/org/apache/tuscany/sca/binding/websocket/runtime/WebsocketReferenceInvoker.java
index a9c2505ea8..84025491fe 100644
--- a/sca-java-2.x/contrib/modules/binding-websocket/src/main/java/org/apache/tuscany/sca/binding/websocket/runtime/WebsocketReferenceInvoker.java
+++ b/sca-java-2.x/contrib/modules/binding-websocket/src/main/java/org/apache/tuscany/sca/binding/websocket/runtime/WebsocketReferenceInvoker.java
@@ -23,6 +23,9 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.channels.SocketChannel;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import org.apache.tuscany.sca.assembly.EndpointReference;
 import org.apache.tuscany.sca.interfacedef.Operation;
@@ -33,6 +36,9 @@ import org.apache.websocket.WebSocketConnector;
 
 public class WebsocketReferenceInvoker implements Invoker {
 
+    // TODO add timeout mechanism for persistent connections
+    private static ConcurrentMap<String, WebSocket<SocketChannel>> persistentWebsockets = new ConcurrentHashMap<String, WebSocket<SocketChannel>>();
+
     protected Operation operation;
     protected EndpointReference endpoint;
 
@@ -43,41 +49,57 @@ public class WebsocketReferenceInvoker implements Invoker {
 
     public Message invoke(Message msg) {
         try {
-            return doInvoke(msg);
+            WebSocket<SocketChannel> websocket = initWebsocketConnection(endpoint.getBinding().getURI());
+            return doInvoke(msg, websocket);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
 
-    public Message doInvoke(Message msg) {
+    private WebSocket<SocketChannel> initWebsocketConnection(String uri) throws IOException, URISyntaxException {
+        WebSocket<SocketChannel> websocket = null;
+        synchronized (persistentWebsockets) {
+            websocket = persistentWebsockets.get(uri);
+            if (websocket == null) {
+                WebSocketConnector connector = new WebSocketConnector();
+                websocket = connector.connect(new URI(uri), null, "apache-tuscany", null);
+                persistentWebsockets.put(uri, websocket);
+            }
+        }
+        return websocket;
+    }
+
+    public Message doInvoke(Message msg, WebSocket<SocketChannel> websocket) throws IOException {
         String componentName = endpoint.getTargetEndpoint().getComponent().getName();
         String serviceName = endpoint.getTargetEndpoint().getService().getName();
         String operationName = operation.getName();
-        String uri = endpoint.getBinding().getURI() + "/" + componentName + "/" + serviceName + "/" + operationName;
-        String jsonParams = JSONUtil.encodeRequestParams((Object[]) msg.getBody());
-        String responseJSON = invokeWebSocketRequest(uri, jsonParams);
+        String uri = componentName + "/" + serviceName + "/" + operationName;
+        String payload = JSONUtil.encodeRequestParams((Object[]) msg.getBody());
+        WebSocketBindingRequest request = new WebSocketBindingRequest(UUID.randomUUID().toString(), uri, payload);
+
+        String operationResponse = invokeViaWebsocket(websocket, JSONUtil.encodeRequest(request));
+
+        WebSocketBindingResponse response = JSONUtil.decodeResponse(operationResponse);
         Class<?> returnType = operation.getOutputType().getLogical().get(0).getPhysical();
-        Object response = JSONUtil.decodeResponse(responseJSON, returnType);
-        msg.setBody(response);
+        Object invocationResponse = JSONUtil.decodeResponsePayload(response.getPayload(), returnType);
+        msg.setBody(invocationResponse);
         return msg;
     }
 
-    private String invokeWebSocketRequest(String uri, String jsonParams) {
-        try {
-            return doInvokeWebSocketRequest(uri, jsonParams);
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        } catch (URISyntaxException e) {
-            throw new RuntimeException(e);
-        }
+    private String invokeViaWebsocket(WebSocket<SocketChannel> websocket, String request) throws IOException {
+        websocket.sendText(request);
+        return websocket.receiveText();
     }
 
-    private String doInvokeWebSocketRequest(String uri, String jsonParams) throws IOException, URISyntaxException {
-        WebSocketConnector connector = new WebSocketConnector();
-        WebSocket<SocketChannel> websocket = connector.connect(new URI(uri), null, "apache-tuscany", null);
-        websocket.sendText(jsonParams);
-        String jsonResponse = websocket.receiveText();
-        websocket.close();
-        return jsonResponse;
+    public static void shutdown() {
+        for (WebSocket<SocketChannel> websocket : persistentWebsockets.values()) {
+            try {
+                websocket.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        persistentWebsockets.clear();
     }
+
 }