JMSCore { public void sendJMSMessage(ExtensionServiceMessage message) { Map<String, ?> request = (Map<String, ?>) message.getObject(); Map<String, Object> messageMap = new LinkedHashMap<String, Object>(); JMS localJMS; synchronized (SYNCH_LOCK) { localJMS = jms; } if (localJMS == null) { log.error("JMS connection closed before operation could complete"); } else { Map headers = null; Map properties = null; Object msg = null; String dest; boolean isQueue; if (request.get("message") instanceof String) { msg = (String) request.get("message"); } else if (request.get("message") instanceof Map) { msg = (Map) request.get("message"); } else { log.debug("No message was specified in the publish request, or the message was not a String/Map. " + "The message was set to its default value, null."); } if (request.get("queue") instanceof String) { dest = (String) request.get("queue"); isQueue = true; } else if (request.get("topic") instanceof String) { dest = (String) request.get("topic"); isQueue = false; } else { log.error("No destination was specified, or destination was not a String. Either a topic " + "or a queue must be included as a String in the publish request."); return; } if (request.get("headers") instanceof Map) { headers = (Map) request.get("headers"); } if (request.get("properties") instanceof Map) { properties = (Map) request.get("properties"); } messageMap.put("message", msg); messageMap.put("headers", headers); messageMap.put("properties", properties); try { localJMS.produceMessage(messageMap, dest, isQueue); } catch (JMSException e) { log.error("An error occured when attempting to send the given message.", e); } catch (DestinationNotConfiguredException e) { log.error("An error occured when attempting to send the given message. The source was not configured " + "to send messages to the following destination: " + dest + ".", e); } catch(UnsupportedJMSMessageTypeException e) { log.error("An error occured when attempting to send the given message. The provided JMS Message Type: " + e.getMessage() + " is either invalid or not currently supported."); } catch (Exception e) { log.error("An unexpected error occured when attempting to send the given message.", e); } } } 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 testIncorectPublishRequests() { 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.sendJMSMessage(msg); assertFalse("Core should not be closed", core.isClosed()); request = new LinkedHashMap<>(); request.put("publish", "jibberish"); msg.object = request; core.sendJMSMessage(msg); assertFalse("Core should not be closed", core.isClosed()); }