class ExtensionWebSocketListener implements WebSocketListener {

    @Override
    public void onMessage(ResponseBody body) {
        // Extract the original message from the 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; // Do nothing if closed at this point
        }
        
        // Convert the data from a Json string/byte array to a map
        Map msg;
        try {
            msg = mapper.readValue(data, Map.class);
        }
        catch (Exception e) {
            log.warn("Failed to interpret WebSocket message as Map.", e);
            return;
        }
        
        //Check to see if we should use log with Debug, or with Error
        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);
        }
        
        // Now we figure out which handler should receive the message
        
        // The message received has no op, and thus is not an ExtensionServiceMethod
        // Since we're acting through the WebSocket interface, this means it should be a Http response
        if (msg.get("op") == null) {
            Response message = Response.fromMap(msg);
            log.trace("Http response received");
            if (client.isAuthed()) {
                // Is an error message before successful connection to the target source
                // This is most likely a failure related to a source connection request
                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 {
                // Resetting authFuture is also sync'd on client, so this way they won't interfere with each other
                synchronized (client) {
                    // Checking isOpen() while sync'd in case of concurrency problems
                    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);
                        // Prepare a response with an empty body, so that the query doesn't wait for a timeout
                        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) {
                        // Do not alert client of reconnect message if this listener has been closed
                        if (this.isClosed) {
                            return; 
                        }
                        client.sourceHasDisconnected(); // Resets to pre source connection state
                    }
                    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();
                    }
                    // Warn when cannot reconnect or know that the connection has failed 
                    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()) {
                // Resetting sourceFuture is also sync'd on client, so this way they won't interfere with each other
                synchronized (client) {
                    // Rechecking isAuthed() while sync'd in case of concurrency problems
                    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");
            }
        }
    }

    public  ExtensionWebSocketListener(ExtensionWebSocketClient client);

}