class 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");

        // Acquire the general settings Map and the listening port and address
        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;
        }

        // Synchronization necessary because createUDPSocket creates a list which listenOnUDPSocket will add to.
        // If threads change between the creation of the UDP socket and the assignation of the list or while
        // a source is being added to the list, an error is likely to occur
        DatagramSocket socket;
        synchronized (ConfigurableUDPSource.socketLock) {
            // Create the socket that the source will use for UDP messages
            socket = ConfigurableUDPSource.createUDPSocket(port, address, sourceName);
            if (socket == null) { // No socket could be created for the given port and address
                socket = ConfigurableUDPSource.listenOnUDPSocket(port, address, sourceName);
            }
            if (socket == null) {
                log.error("Failed to obtain UDP socket at address '{}' and port '{}'", address, port);
                return;
            }
        }

        // Setup Publish handler as the configuration document requests
        if (isConfiguredToSend(outgoing)) {
            UDPPublishHandler handler = new UDPPublishHandler(outgoing, socket, sourceName);
            ConfigurableUDPSource.clients.get(sourceName).setPublishHandler(handler);
            log.debug("Publish handler created");
        }

        // Setup Notification handler as the configuration document requests
        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.");
    }

}