JMSCore { public void readQueueMessage(ExtensionServiceMessage message) { Map<String, ?> request = (Map<String, ?>) message.getObject(); String replyAddress = ExtensionServiceMessage.extractReplyAddress(message); JMS localJMS; synchronized (SYNCH_LOCK) { localJMS = jms; } if (localJMS == null) { if (client != null) { client.sendQueryError(replyAddress, this.getClass().getName() + ".closed", "JMS connection closed before operation could complete.", null); } else { log.error("JMS connection closed before operation could complete."); } } else { if (request.get("queue") instanceof String) { String queue = (String) request.get("queue"); int timeout = -1; if (request.get("timeout") instanceof Integer && (Integer) request.get("timeout") > 0) { timeout = (Integer) request.get("timeout"); } try { Map<String, Object> messageMap = localJMS.consumeMessage(queue, timeout); if (messageMap == null) { client.sendQueryError(replyAddress, this.getClass().getName() + ".invalidMessage", "The returned message was invalid. This is most likely because the MessageHandler did not format " + "the returned message, headers, properties, and queue name correctly.", null); } else { client.sendQueryResponse(200, replyAddress, messageMap); } } catch (JMSException e) { client.sendQueryError(replyAddress, JMSException.class.getCanonicalName(), "Failed to read message from the queue: " + queue + ". Error message was: " + e.getMessage(), null); } catch (DestinationNotConfiguredException e) { client.sendQueryError(replyAddress, DestinationNotConfiguredException.class.getCanonicalName(), "Failed to read message from the queue: " + queue + ". The source was not configured to read from " + "this queue.", null); } catch(UnsupportedJMSMessageTypeException e) { client.sendQueryError(replyAddress, UnsupportedJMSMessageTypeException.class.getCanonicalName(), "Failed to read message from the queue: " + queue + ". The incoming JMS Message Type was: " + e.getMessage() + ", which is not currently supported.", null); } catch (Exception e) { client.sendQueryError(replyAddress, Exception.class.getCanonicalName(), "An unexpected error occured when reading message from queue: " + queue + ". Error message was: " + e.getMessage(), null); } } else { client.sendQueryError(replyAddress, this.getClass().getName() + ".noQueue", "No queue was specified as a query parameter. Query cannot be completed.", null); } } } JMSCore(String sourceName, String authToken, String targetVantiqServer); String getSourceName(); boolean start(int timeout); void sendJMSMessage(ExtensionServiceMessage message); void readQueueMessage(ExtensionServiceMessage message); void close(); void stop(); boolean exitIfConnectionFails(ExtensionWebSocketClient client, int timeout); final Handler<ExtensionServiceMessage> reconnectHandler; final Handler<ExtensionWebSocketClient> closeHandler; }
@Test public void testIncorrectQueryRequests() { Map<String, Object> request; ExtensionServiceMessage msg = new ExtensionServiceMessage(""); Map<String, String> header = new LinkedHashMap<>(); header.put(ExtensionServiceMessage.ORIGIN_ADDRESS_HEADER, "queryAddress"); msg.messageHeaders = header; request = new LinkedHashMap<>(); msg.object = request; core.readQueueMessage(msg); assertFalse("Core should not be closed", core.isClosed()); request = new LinkedHashMap<>(); request.put("query", "jibberish"); msg.object = request; core.readQueueMessage(msg); assertFalse("Core should not be closed", core.isClosed()); }