ExtensionWebSocketClient { public void sendNotification(Object data) { if (data != null && (data.getClass().isArray() || data instanceof List)) { throw new IllegalArgumentException("Notifications cannot be lists or arrays."); } if (isConnected()) { Map<String,Object> m = new LinkedHashMap<>(); m.put("op", ExtensionServiceMessage.OP_NOTIFICATION); m.put("resourceId", sourceName); m.put("resourceName", ExtensionServiceMessage.RESOURCE_NAME_SOURCES); m.put("object", data); ExtensionServiceMessage msg = new ExtensionServiceMessage(""); msg.fromMap(m); try { outstandingNotifications.acquire(); this.send(msg); } catch (InterruptedException ie) { log.warn("Obtaining space to sent notifications was interrupted.", ie); } catch (Exception e) { outstandingNotifications.release(); throw e; } } } ExtensionWebSocketClient(String sourceName); ExtensionWebSocketListener getListener(); String getSourceName(); CompletableFuture<Boolean> initiateFullConnection(String url, String token); synchronized CompletableFuture<Boolean> initiateWebsocketConnection(String url); CompletableFuture<Boolean> getWebsocketConnectionFuture(); void sendNotification(Object data); void sendQueryResponseEmpty(String replyAddress); void sendQueryResponse(int httpCode, String replyAddress, Map body); void sendQueryResponse(int httpCode, String replyAddress, Map[] body); void sendQueryError(String replyAddress, String messageCode, String messageTemplate, Object[] parameters); void send(Object obj); synchronized CompletableFuture<Boolean> authenticate(String user, String pass); synchronized CompletableFuture<Boolean> authenticate(String token); CompletableFuture<Boolean> getAuthenticationFuture(); synchronized CompletableFuture<Boolean> connectToSource(); CompletableFuture<Boolean> getSourceConnectionFuture(); void sourceHasDisconnected(); void setAutoReconnect(boolean value); boolean isOpen(); boolean isAuthed(); boolean isConnected(); void close(); void stop(); void useHandlersFrom(ExtensionWebSocketListener listener); void useHandlersFrom(ExtensionWebSocketClient client); void setCloseHandler(Handler<ExtensionWebSocketClient> closeHandler); void setHttpHandler(Handler<Response> httpHandler); void setPublishHandler(Handler<ExtensionServiceMessage> publishHandler); void setQueryHandler(Handler<ExtensionServiceMessage> queryHandler); void setConfigHandler(Handler<ExtensionServiceMessage> configHandler); void setAuthHandler(Handler<Response> authHandler); void setReconnectHandler(Handler<ExtensionServiceMessage> reconnectHandler); static final int QUERY_CHUNK_CODE; static final int QUERY_DATA_CODE; static final int QUERY_NODATA_CODE; }
@Test public void testNotification() { markSourceConnected(true); Map<String,Object> m = new LinkedHashMap<>(); m.put("msg", "str"); client.sendNotification(m); assert socket.compareData("op", ExtensionServiceMessage.OP_NOTIFICATION); assert socket.compareData("object.msg", "str"); assert socket.compareData("resourceId", srcName); }
@Test public void testBadNotificationArguments() { markSourceConnected(true); Integer[] intArray = new Integer[3]; intArray[0] = 1; intArray[1] = 2; intArray[2] = 3; try { client.sendNotification(intArray); fail("Send of Array should fail."); } catch (IllegalArgumentException iae) { } List<String> stringList = new ArrayList<>(); stringList.add("This "); stringList.add("should "); stringList.add("not "); stringList.add("work."); try { client.sendNotification(stringList); fail("Send of List should fail."); } catch (IllegalArgumentException iae) { } }