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