Added event handling support for the server.

Author: TVolden

ocpp-common/src/main/java/eu/chargetime/ocpp/Communicator.java

@@ -114,29 +114,16 @@ public Communicator(Transmitter transmitter) {
      * @param   events  handler for call back events.
      */
     public void connect(String uri, CommunicatorEvents events) {
-        transmitter.connect(uri, new TransmitterEvents() {
-            @Override
-            public void connected() {
-                events.onConnected();
-            }
-
-            @Override
-            public void receivedMessage(String input) {
-                System.out.println(input);
-                Message message = parse(input);
-                if(message instanceof CallResultMessage) {
-                    events.onCallResult(message.getId(), message.getPayload());
-                } else if (message instanceof CallMessage) {
-                    CallMessage call = (CallMessage)message;
-                    events.onCall(call.getId(), call.getAction(), call.getPayload());
-                }
-            }
+        transmitter.connect(uri, new EventHandler(events));
+    }
 
-            @Override
-            public void disconnected() {
-                events.onDisconnected();
-            }
-        });
+    /**
+     * Use the injected {@link Transmitter} to accept a client.
+     *
+     * @param events handler for call back events.
+     */
+    public void accept(CommunicatorEvents events) {
+        transmitter.accept(new EventHandler(events));
     }
 
     /**
@@ -177,4 +164,34 @@ public void sendCallError(String uniqueId, String errorCode, String errorDescrip
     public void disconnect() {
         transmitter.disconnect();
     }
+
+    private class EventHandler implements TransmitterEvents {
+        private final CommunicatorEvents events;
+
+        public EventHandler(CommunicatorEvents events) {
+            this.events = events;
+        }
+
+        @Override
+        public void connected() {
+            events.onConnected();
+        }
+
+        @Override
+        public void receivedMessage(String input) {
+            System.out.println(input);
+            Message message = parse(input);
+            if (message instanceof CallResultMessage) {
+                events.onCallResult(message.getId(), message.getPayload());
+            } else if (message instanceof CallMessage) {
+                CallMessage call = (CallMessage) message;
+                events.onCall(call.getId(), call.getAction(), call.getPayload());
+            }
+        }
+
+        @Override
+        public void disconnected() {
+            events.onDisconnected();
+        }
+    }
 }

ocpp-common/src/main/java/eu/chargetime/ocpp/Server.java

@@ -42,17 +42,71 @@ public abstract class Server extends FeatureHandler {
 
     private ArrayList<Session> sessions;
 
+    /**
+     *
+     */
     public Server() {
         this.sessions = new ArrayList<>();
     }
 
+    /**
+     * Start listening for clients.
+     *
+     * @param listener     Inject the listener.
+     * @param serverEvents Callback handler for server specific events.
+     */
     public void open(Listener listener, ServerEvents serverEvents) {
         listener.open(session -> {
+            session.accept(new SessionEvents() {
+                @Override
+                public Feature findFeatureByAction(String action) {
+                    return findFeature(action);
+                }
+
+                @Override
+                public Feature findFeatureByRequest(Request request) {
+                    return findFeature(request);
+                }
+
+                @Override
+                public void handleConfirmation(String uniqueId, Confirmation confirmation) {
+                    getPromise(uniqueId).complete(confirmation);
+                }
+
+                @Override
+                public Confirmation handleRequest(Request request) {
+                    Feature feature = findFeatureByRequest(request);
+                    return feature.handleRequest(sessions.indexOf(session), request);
+                }
+
+                @Override
+                public void handleError(String uniqueId) {
+
+                }
+
+                @Override
+                public void handleConnectionClosed() {
+
+                }
+
+                @Override
+                public void handleConnectionOpened() {
+
+                }
+            });
             sessions.add(session);
             serverEvents.newSession(sessions.indexOf(session));
         });
     }
 
+    /**
+     * Send a message to a client.
+     *
+     * @param sessionIndex Session index of the client.
+     * @param request      Request for the client.
+     * @return Callback handler for when the client responds.
+     * @throws UnsupportedFeatureException Thrown if the feature isn't amoung the list of supported featured.
+     */
     public CompletableFuture<Confirmation> send(int sessionIndex, Request request) throws UnsupportedFeatureException {
         Feature feature = findFeature(request);
         if (feature == null)

ocpp-common/src/main/java/eu/chargetime/ocpp/Session.java

@@ -103,6 +103,11 @@ public void close() {
         communicator.disconnect();
     }
 
+    public void accept(SessionEvents eventHandler) {
+        this.events = eventHandler;
+        communicator.accept(new CommunicatorEventHandler());
+    }
+
     private class CommunicatorEventHandler implements CommunicatorEvents {
         @Override
         public void onCallResult(String id, String payload) {

ocpp-common/src/main/java/eu/chargetime/ocpp/Transmitter.java

@@ -51,4 +51,11 @@ public interface Transmitter
      * @param message    test message to send.
      */
     void send(String message);
+
+    /**
+     * Accept an incoming connection request.
+     *
+     * @param events connection related events.
+     */
+    void accept(TransmitterEvents events);
 }

ocpp-common/src/test/java/eu/chargetime/ocpp/test/ServerTest.java

@@ -61,7 +61,9 @@ public void setup() {
         request = () -> false;
         doReturn(request.getClass()).when(feature).getRequestType();
         doReturn(TestConfirmation.class).when(feature).getConfirmationType();
+        when(feature.getAction()).thenReturn(null);
         doAnswer(invocation -> listenerEvents = invocation.getArgumentAt(0, ListenerEvents.class)).when(listener).open(any());
+        doAnswer(invocation -> sessionEvents = invocation.getArgumentAt(0, SessionEvents.class)).when(session).accept(any());
 
         server = new Server() {
         };
@@ -70,6 +72,18 @@ public void setup() {
         server.addFeatureProfile(profile);
     }
 
+    @Test
+    public void newSession_serverIsListening_sessionIsAccepted() {
+        // Given
+        server.open(listener, serverEvents);
+
+        // When
+        listenerEvents.newSession(session);
+
+        // Then
+        verify(session, times(1)).accept(any());
+    }
+
     @Test
     public void newSession_serverIsListening_callbackWithIndex0() {
         // Given
@@ -99,4 +113,17 @@ public void send_aMessage_isCommunicated() throws Exception {
         verify(session, times(1)).sendRequest(anyString(), eq(request));
     }
 
+    @Test
+    public void handleRequest_callsFeatureHandleRequest() {
+        // Given
+        server.open(listener, serverEvents);
+        listenerEvents.newSession(session);
+
+        // When
+        sessionEvents.handleRequest(request);
+
+        // Then
+        verify(feature, times(1)).handleRequest(eq(0), eq(request));
+    }
+
 }

ocpp-common/src/test/java/eu/chargetime/ocpp/test/SessionTest.java

@@ -204,7 +204,25 @@ public void onCall_unknownAction_callSendCallError() {
         // When
         eventHandler.onCall(someId, null, null);
 
-        // then
+        // Then
         verify(communicator, times(1)).sendCallError(eq(someId), anyString(), anyString());
     }
+
+    @Test
+    public void onCall_sessionAccepted_callIsForwarded() throws Exception {
+        // Given
+        String someId = "Some id";
+        session.accept(sessionEvents);
+        when(communicator.unpackPayload(any(), any())).thenReturn(new TestRequest());
+
+        // When
+        eventHandler.onCall(someId, null, null);
+        try {
+            Thread.sleep(10);
+        } catch (Exception ex) {
+        } // TODO make async invoker injectable
+
+        // Then
+        verify(sessionEvents, times(1)).handleRequest(any());
+    }
 }

ocpp-v1_6/src/main/java/eu/chargetime/ocpp/WebSocketTransmitter.java

@@ -88,4 +88,9 @@ public void disconnect()
     public void send(String request) {
         client.send(request);
     }
+
+    @Override
+    public void accept(TransmitterEvents events) {
+
+    }
 }