ExtensionWebSocketListener implements WebSocketListener { @Override public void onMessage(ResponseBody body) { byte[] data; try { if (body.contentType() == WebSocket.TEXT) { data = body.string().getBytes(); } else { data = body.bytes(); } } catch (IOException e) { log.error("Error trying to interpret WebSocket message", e); return; } body.close(); if (this.isClosed) { return; } Map msg; try { msg = mapper.readValue(data, Map.class); } catch (Exception e) { log.warn("Failed to interpret WebSocket message as Map.", e); return; } if (msg.containsKey("status")) { int statusCheck = (Integer) msg.get("status"); if (statusCheck >= 300) { log.error("Map of the received message: {}", msg); } else { log.trace("Map of the received message: {}", msg); } } else { log.debug("Map of the received message: {}", msg); } if (msg.get("op") == null) { Response message = Response.fromMap(msg); log.trace("Http response received"); if (client.isAuthed()) { if (!client.isConnected() && (Integer) message.getStatus() >= 300) { log.warn("Error occurred attempting to connect to source."); log.debug("Error message was: {}", message); client.sourceFuture.complete(false); } else { client.acknowledgeNotification(); } if (this.httpHandler != null) { try { this.httpHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the HTTP handler.", e); } } else { log.trace("Http response received with no handler set"); } } else { synchronized (client) { if (!client.isOpen() || this.isClosed) { return; } if ((int) message.getStatus() == 200 && !client.isAuthed()) { client.authFuture.complete(true); } else { client.authFuture.complete(false); log.warn("Error occurred attempting to authenticate"); } } if (authHandler != null) { try { this.authHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the authentication handler for source.", e); } } else { log.debug("Auth received with no handler set"); } } } else { ExtensionServiceMessage message = new ExtensionServiceMessage("").fromMap(msg); if (client.isConnected()) { log.debug("Message with op '{}' received", message.getOp()); log.debug("Map of ExtensionServiceMessage: {}", message); if (message.getOp().equals(ExtensionServiceMessage.OP_PUBLISH)) { if (this.publishHandler != null) { try { this.publishHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the Publish handler.", e); } } else { log.debug("Publish received with no handler set"); } } else if (message.getOp().equals(ExtensionServiceMessage.OP_QUERY)) { if (this.queryHandler != null && !isClosed) { try { this.queryHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the Query handler.", e); } } else { log.warn("Query received with no user-set handler"); log.debug("Full message: {}", message); client.sendQueryError(ExtensionServiceMessage.extractReplyAddress(msg), "io.vantiq.extjsdk.unsetQueryHandler", "Queries are not supported for source '{0}'. No handler has been set.", new Object[] {message.getSourceName()}); } } else if (message.getOp().equals(ExtensionServiceMessage.OP_RECONNECT_REQUIRED)) { synchronized (client) { if (this.isClosed) { return; } client.sourceHasDisconnected(); } if (this.reconnectHandler != null) { try { this.reconnectHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the Reconnect handler.", e); } } if (client.autoReconnect) { log.info("Automatically attempting to reconnect to source."); client.connectToSource(); } if (!client.autoReconnect && this.reconnectHandler == null) { log.warn("Reconnect received with no handler set and no autoconnect. Can no longer " + "communicate with source."); } } else { log.warn("ExtensionServiceMessage with unknown/unexpected op '{}'", msg.get("op")); } } else if (msg.get("op").equals(ExtensionServiceMessage.OP_CONFIGURE_EXTENSION) && client.isAuthed()) { synchronized (client) { if (!client.isAuthed()) { return; } client.sourceFuture.complete(true); log.info("Successful connection to {}", msg.get("resourceId").toString()); } if (this.configHandler != null) { try { this.configHandler.handleMessage(message); } catch (Exception e) { log.error("Error occurred when running the Configuration handler.", e); } } else { log.warn("Configuration received with no handler set"); } } else { log.warn("ExtensionServiceMessage received when not connected"); } } } ExtensionWebSocketListener(ExtensionWebSocketClient client); 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); void close(); boolean isClosed(); @Override void onOpen(WebSocket webSocket, okhttp3.Response response); @Override void onMessage(ResponseBody body); void useHandlersFromListener(ExtensionWebSocketListener listener); void useHandlersFromListener(ExtensionWebSocketClient client); @Override void onClose(int code, String reason); @Override void onPong(Buffer payload); @Override void onFailure(IOException e, okhttp3.Response response);  }
@Test public void testSourceConnectionFailFirst() { authenticate(true); listener.onMessage(TestListener.errorMessage()); assert !client.getSourceConnectionFuture().getNow(true); Map<String,Object> config = new LinkedHashMap<>(); String key = "key"; String val = "value"; config.put(key, val); connectToSource(srcName, config); assert client.isConnected(); assert cHandler.compareOp(ExtensionServiceMessage.OP_CONFIGURE_EXTENSION); assert cHandler.compareSourceName(srcName); assert cHandler.compareValue("config",config); }
@Test public void testSourceConnectionFailFirstAndAuthFailFirst() { authenticate(false); waitUntilTrue(5 * 1000, () -> client.getSourceConnectionFuture().isDone()); assert !client.getSourceConnectionFuture().getNow(true); authenticate(true); listener.onMessage(TestListener.errorMessage()); assert !client.getSourceConnectionFuture().getNow(true); Map<String,Object> config = new LinkedHashMap<>(); String key = "key"; String val = "value"; config.put(key, val); connectToSource(srcName, config); assert client.isConnected(); assert cHandler.compareOp(ExtensionServiceMessage.OP_CONFIGURE_EXTENSION); assert cHandler.compareSourceName(srcName); assert cHandler.compareValue("config", config); }
@Test public void testPublish() { connectToSource(srcName, null); Map<String,Object> publishMessage = new LinkedHashMap<>(); String key = "publish"; String val = "info"; publishMessage.put(key, val); ResponseBody body = TestListener.createPublishMessage(publishMessage, srcName); listener.onMessage(body); assert pHandler.compareOp(ExtensionServiceMessage.OP_PUBLISH); assert pHandler.compareSourceName(srcName); assert pHandler.compareValue( key, val); }
@Test public void testQuery() { connectToSource(srcName, null); Map<String,Object> queryMessage = new LinkedHashMap<>(); String key = "query"; String val = "request"; queryMessage.put(key, val); ResponseBody body = TestListener.createQueryMessage(queryMessage, srcName); listener.onMessage(body); assert qHandler.compareOp(ExtensionServiceMessage.OP_QUERY); assert qHandler.compareSourceName( srcName); assert qHandler.compareValue(key, val); }
@Test public void testHttp() { connectToSource(srcName, null); Response resp = new Response().status(200); ResponseBody body = TestListener.createHttpMessage(resp); listener.onMessage(body); assert hHandler.compareStatus(200); }
@Test public void testHttpHandlerOnFailedConnection() { authenticate(true); Response resp = new Response().status(403); ResponseBody body = TestListener.createHttpMessage(resp); listener.onMessage(body); assert !client.getSourceConnectionFuture().getNow(true); assert hHandler.compareStatus(403); }
@Test public void testReconnectHandler() { connectToSource(srcName, null); ResponseBody body = TestListener.createReconnectMessage(srcName); listener.onMessage(body); assert !client.isConnected(); assert rHandler.compareOp(ExtensionServiceMessage.OP_RECONNECT_REQUIRED); }
@Test public void testDefaultQuery() { listener = new ExtensionWebSocketListener(client); client.listener = listener; connectToSource(srcName, null); ResponseBody body = TestListener.createQueryMessage(new LinkedHashMap(), srcName); listener.onMessage(body); Response resp = null; try { resp = client.getLastMessageAsResponse(); } catch(Exception e) { e.printStackTrace(); fail("Could not interpret the default query handler response\n"); } assert resp.getBody() instanceof Map; Map<String,Object> m = (Map) resp.getBody(); assert "io.vantiq.extjsdk.unsetQueryHandler".equals(m.get("messageCode")); assert m.get("messageTemplate") instanceof String; assert m.get("parameters") instanceof List; }