UDPConfigHandler extends Handler<ExtensionServiceMessage> { @Override public void handleMessage(ExtensionServiceMessage message) { String sourceName = message.getSourceName(); Logger log = LoggerFactory.getLogger(this.getClass().getCanonicalName() + "#" + sourceName); Map srcConfig = (Map) ((Map)message.getObject()).get("config"); if (!(srcConfig.get("udpSourceConfig") instanceof Map)) { log.error("Unable to obtain source configuration."); return; } Map config = (Map) srcConfig.get("udpSourceConfig"); log.trace("Creating handlers"); Map general = null; if (config.get("general") instanceof Map) { general = (Map) config.get("general"); } int port = getListeningPort(general, sourceName); InetAddress address = getListeningAddress(general, sourceName); Map incoming = null; Map outgoing = null; if (config.get("incoming") instanceof Map) { incoming = (Map) config.get("incoming"); } if (config.get("outgoing") instanceof Map) { outgoing = (Map) config.get("outgoing"); } if (!isConfiguredToSend(outgoing) && !isConfiguredToReceive(incoming)) { log.error("Source is not configured to send or receive."); return; } DatagramSocket socket; synchronized (ConfigurableUDPSource.socketLock) { socket = ConfigurableUDPSource.createUDPSocket(port, address, sourceName); if (socket == null) { socket = ConfigurableUDPSource.listenOnUDPSocket(port, address, sourceName); } if (socket == null) { log.error("Failed to obtain UDP socket at address '{}' and port '{}'", address, port); return; } } if (isConfiguredToSend(outgoing)) { UDPPublishHandler handler = new UDPPublishHandler(outgoing, socket, sourceName); ConfigurableUDPSource.clients.get(sourceName).setPublishHandler(handler); log.debug("Publish handler created"); } if (isConfiguredToReceive(incoming)) { UDPNotificationHandler handler = new UDPNotificationHandler(incoming, ConfigurableUDPSource.clients.get(sourceName)); ConfigurableUDPSource.setNotificationHandler(handler, sourceName, incoming); log.debug("Notification handler created"); } log.info("Source setup and ready to go."); }  @Override void handleMessage(ExtensionServiceMessage message);  }
@Test public void testHandleMessage() throws UnknownHostException { ExtensionServiceMessage msg = new ExtensionServiceMessage(""); msg.resourceId = sourceName; Map<String,Object> object = new LinkedHashMap<>(); msg.object = object; Map<String,Object> config = new LinkedHashMap<>(); object.put("config", config); Map<String,Object> udpSourceConfig = new LinkedHashMap<>(); config.put("udpSourceConfig", udpSourceConfig); Map<String,Object> general = new LinkedHashMap<>(); Map<String,Object> incoming = new LinkedHashMap<>(); Map<String,Object> outgoing = new LinkedHashMap<>(); udpSourceConfig.put("type", "udp"); udpSourceConfig.put("general", general); udpSourceConfig.put("incoming", incoming); udpSourceConfig.put("outgoing", outgoing); udpConfig.handleMessage(msg); assert !ConfigurableUDPSource.notificationHandlers.containsKey(sourceName); assert ConfigurableUDPSource.udpSocketToSources.isEmpty(); int port = 10213; String addressName = "invalidAddress"; incoming.put("passPureMapIn", true); outgoing.put("targetPort", port); outgoing.put("targetAddress", addressName); InetAddress address = InetAddress.getLocalHost(); ConfigurableUDPSource.LISTENING_ADDRESS = address; assumeTrue("Cannot perform test. Address at port:" + port + " and IP:" + address + "already bound" , socketCanBind(port, address)); udpConfig.handleMessage(msg); assert ConfigurableUDPSource.notificationHandlers.containsKey(sourceName); DatagramSocket socket = (DatagramSocket) ConfigurableUDPSource.udpSocketToSources.keySet().toArray()[0]; assert socket.getLocalPort() == 3141; assert socket.getLocalAddress().equals(ConfigurableUDPSource.LISTENING_ADDRESS); addressName = "localhost"; address = InetAddress.getByName(addressName); general.put("listenPort", port); general.put("listenAddress", addressName); assumeTrue("Cannot perform test. Address at port:" + port + " and IP:" + address + "already bound" , socketCanBind(port, address)); udpConfig.handleMessage(msg); assert ConfigurableUDPSource.udpSocketToSources.keySet().size() == 2; }