soa/service/zmq_named_pub_sub.h

/* zmq_named_pub_sub.h                                             -*- C++ -*-
   Jeremy Barnes, 8 January 2013
   Copyright (c) 2013 Datacratic Inc.  All rights reserved.
   
   Named publish/subscribe endpoint.
*/

#pragma once

#include "zmq_endpoint.h"
#include "typed_message_channel.h"
#include <sys/utsname.h>
#include "jml/arch/backtrace.h"

namespace Datacratic {



/*****************************************************************************/
/* ZMQ NAMED PUBLISHER                                                       */
/*****************************************************************************/

struct ZmqNamedPublisher: public MessageLoop {

    ZmqNamedPublisher(std::shared_ptr<zmq::context_t> context,
                      int messageBufferSize = 10000)
        : publishEndpoint(context),
          publishQueue(messageBufferSize)
    {
    }

    virtual ~ZmqNamedPublisher()
    {
        shutdown();
    }

    void init(std::shared_ptr<ConfigurationService> config,
              const std::string & endpointName,
              const std::string & identity = "")
    {
        using namespace std;

        // Initialize the publisher endpoint
        publishEndpoint.init(config, ZMQ_XPUB, endpointName);

        // Called when we queue up a message to be published
        publishQueue.onEvent = [=] (std::vector<zmq::message_t> && message)
            {
                using namespace std;
                //cerr << "popped message to publish" << endl;
                publishEndpoint.sendMessage(std::move(message));
            };

        // Called when there is a new subscription.
        // The first byte is 1 (subscription) or 0 (unsubscription).
        // The rest of the message is the filter for the subscription
        auto doPublishMessage = [=] (const std::vector<std::string> & msg)
            {
#if 0
                cerr << "got publish subscription" << msg << endl;
                cerr << "msg.size() = " << msg.size() << endl;
                cerr << "msg[0].size() = " << msg[0].size() << endl;
                cerr << "msg[0][0] = " << (int)msg[0][0] << endl;
#endif
            };

        publishEndpoint.messageHandler = doPublishMessage;

        // Hook everything into the message loop
        addSource("ZmqNamedPublisher::publishEndpoint", publishEndpoint);
        addSource("ZmqNamedPublisher::publishQueue", publishQueue);

    }

    std::string bindTcp(PortRange const & portRange = PortRange(), std::string host = "")
    {
        return publishEndpoint.bindTcp(portRange, host);
    }

    void shutdown()
    {
        MessageLoop::shutdown();
        publishEndpoint.shutdown();
        //publishEndpointMonitor.shutdown();
    }

    //std::vector<std::string> getSubscribers()
    //{
    //}

    template<typename Head, typename... Tail>
    void encodeAll(std::vector<zmq::message_t> & messages,
                   Head head,
                   Tail&&... tail)
    {
        messages.emplace_back(std::move(encodeMessage(head)));
        encodeAll(messages, std::forward<Tail>(tail)...);
    }

    // Vectors treated specially... they are copied
    template<typename... Tail>
    void encodeAll(std::vector<zmq::message_t> & messages,
                   const std::vector<std::string> & head,
                   Tail&&... tail)
    {
        for (auto & m: head)
            messages.emplace_back(std::move(encodeMessage(m)));
        encodeAll(messages, std::forward<Tail>(tail)...);
    }

    void encodeAll(std::vector<zmq::message_t> & messages)
    {
    }

    template<typename... Args>
    void publish(const std::string & channel, Args&&... args)
    {
        std::vector<zmq::message_t> messages;
        messages.reserve(sizeof...(Args) + 1);
        
        encodeAll(messages, channel,
                  std::forward<Args>(args)...);
        publishQueue.push(messages);
    }

private:
    ZmqNamedEndpoint publishEndpoint;

    TypedMessageSink<std::vector<zmq::message_t> > publishQueue;
};


/*****************************************************************************/
/* ENDPOINT CONNECTOR                                                        */
/*****************************************************************************/

struct EndpointConnector : public MessageLoop {

    EndpointConnector()
        : changes(32)
    {
    }

    void init(std::shared_ptr<ConfigurationService> config)
    {
        this->config = config;

        changes.onEvent
            = std::bind(&EndpointConnector::handleEndpointChange,
                        this,
                        std::placeholders::_1);

        addSource("EndpointConnector::changes", changes);
    }

    void watchEndpoint(const std::string & endpointPath)
    {
        std::unique_lock<Lock> guard(lock);

        using namespace std;
        //cerr << "watching endpoint " << endpointPath << endl;

        auto & entry = endpoints[endpointPath];

        if (!entry.watch) {
            // First time that we watch this service; set it up

            //cerr << "=============== initializing watch for " << endpointPath << endl;

            entry.watch.init([=] (const std::string & path,
                                  ConfigurationService::ChangeType change)
                             {
                                 using namespace std;
                                 //cerr << "endpoint changed path " << path << " " << this << endl;

                                 {
                                     std::unique_lock<Lock> guard(lock);
                                     if (endpoints.count(endpointPath)) {
                                         auto & entry = endpoints[endpointPath];
                                         entry.watchIsSet = false;
                                     }
                                 }
                                 
                                 changes.push(endpointPath);
                             });

            changes.push(endpointPath);
        }
    }
    
    void unwatchEndpoint(const std::string & endpoint,
                         bool forceDisconnect)
    {
        throw ML::Exception("unwatchEndpoint not done");
    }

    std::function<bool (const std::string & endpointPath,
                        const std::string & entry,
                        const Json::Value & params)> connectHandler;
    
    virtual bool connect(const std::string & endpointPath,
                         const std::string & entryName,
                         const Json::Value & params)
    {
        if (connectHandler)
            return connectHandler(endpointPath, entryName, params);
        throw ML::Exception("no connect override");
    }

    void handleDisconnection(const std::string & endpointPath,
                             const std::string & entryName)
    {
        std::unique_lock<Lock> guard(lock);

        if (!endpoints.count(endpointPath))
            return;

        auto & entry = endpoints[endpointPath];

        ExcAssertEqual(entry.connectedTo, entryName);
        entry.connectedTo = "";

        // And we attempt to reconnect
        // Note that we can't push endpointPath on changes
        changes.push(endpointPath);
    }

private:
    typedef std::mutex Lock;
    mutable Lock lock;

    TypedMessageSink<std::string> changes;

    struct Entry {
        Entry()
            : watchIsSet(false)
        {
        }

        ConfigurationService::Watch watch;
        std::string connectedTo;
        bool watchIsSet;
    };

    std::map<std::string, Entry> endpoints;

    std::shared_ptr<ConfigurationService> config;

    void handleEndpointChange(const std::string & endpointPath)
    {
        using namespace std;

        //cerr << "handleEndpointChange " << endpointPath << endl;

        std::unique_lock<Lock> guard(lock);

        if (!endpoints.count(endpointPath))
            return;

        auto & entry = endpoints[endpointPath];

        //cerr << "handling service class change for " << endpointPath << endl;

        vector<string> children;
        if (entry.watchIsSet)
            children = config->getChildren(endpointPath);
        else children = config->getChildren(endpointPath, entry.watch);
        entry.watchIsSet = true;

        //cerr << "children = " << children << endl;

        // If we're connected, look for a change in the endpoints
        if (!entry.connectedTo.empty()) {

            // Does our connected node still exist?
            if (std::find(children.begin(), children.end(), entry.connectedTo)
                == children.end()) {
                // Node disappeared; we need to disconnect
                guard.unlock();
                handleDisconnection(endpointPath, entry.connectedTo);
                return;  // handleDisconnection will call into us recursively
            }
            else {
                // Node is still there
                // TODO: check for change in value
                return;
            }
        }

        guard.unlock();

        // If we got here, we're not connected
        for (auto & c: children) {
            Json::Value cfg = config->getJson(endpointPath + "/" + c);
            if (connect(endpointPath, c, cfg)) {
                notifyConnectionStatus(endpointPath, c, true);
                return;
            }
        }

        cerr << "warning: could not connect to " << endpointPath << " immediately"
             << endl;
    }

    void notifyConnectionStatus(const std::string & endpointPath,
                                const std::string & entryName,
                                bool status)
    {
        std::unique_lock<Lock> guard(lock);

        if (!endpoints.count(endpointPath))
            return;

        auto & entry = endpoints[endpointPath];

        if (status) {
            if (!entry.connectedTo.empty()) {
                // TODO
                throw ML::Exception("TODO: handle connection with one already "
                                    "active");
            }
            entry.connectedTo = entryName;
            // TODO: watch config
        }
        else {
            if (entry.connectedTo.empty())
                return;
            if (entry.connectedTo != entryName) {
                throw ML::Exception("TODO: handle disconnection from non-active");
            }
            entry.connectedTo = "";
        }
    }
};


/*****************************************************************************/
/* ZMQ NAMED SOCKET                                                      */
/*****************************************************************************/

struct ZmqNamedSocket: public MessageLoop {

    enum ConnectionState {
        NO_CONNECTION,   
        CONNECTING,      
        CONNECTED,       
        DISCONNECTED     
    };

    ZmqNamedSocket(zmq::context_t & context, int type)
        : context(&context),
          socketType(type),
          connectionState(NO_CONNECTION),
          monitor(context)
    {
        //using namespace std;
        //cerr << "created zmqNamedSocket at " << this << endl;
    }

    virtual ~ZmqNamedSocket()
    {
        shutdown();
    }

    void init(std::shared_ptr<ConfigurationService> config)
    {
        if (socket)
            throw ML::Exception("socket already initialized");
        socket.reset(new zmq::socket_t(*context, socketType));
        
        using namespace std;

        monitor.defaultHandler
            = std::bind(&ZmqNamedSocket::handleMonitorEvent,
                        this,
                        std::placeholders::_1,
                        std::placeholders::_2,
                        std::placeholders::_3);

        connector.connectHandler
            = std::bind(&ZmqNamedSocket::doConnect,
                        this,
                        std::placeholders::_1,
                        std::placeholders::_2,
                        std::placeholders::_3);

        connector.init(config);
        monitor.init(*socket);

        addSource("ZmqNamedSocket::connector", connector);
        addSource("ZmqNamedSocket::monitor", monitor);
        addSource("ZmqNamedSocket::socket",
                  std::make_shared<ZmqBinaryEventSource>
                  (*socket, [=] (std::vector<zmq::message_t> && message)
                   {
                       this->handleMessage(std::move(message));
                   }));
    }

    void shutdown()
    {
        MessageLoop::shutdown();

        if (!socket)
            return;

        disconnect();
        monitor.shutdown();
        connector.shutdown();

        socket->tryDisconnect(this->connectedAddress);
        socket.reset();
    }

    void connectToEndpoint(const std::string & endpointPath)
    {
        if (connectedEndpoint != "")
            throw ML::Exception("attempt to connect a ZmqNamedSocket "
                                "to an enpoint that is already connected");
     
        this->connectedEndpoint = connectedEndpoint;
        this->connectionState = CONNECTING;

        // No lock needed as the connector has its own lock

        // Tell the connector to watch the endpoint.  When an entry pops
        // up, it will connect to it.
        connector.watchEndpoint(endpointPath);
    }

    void disconnect()
    {
        std::unique_lock<Lock> guard(lock);
        ExcAssert(socket);
        socket->tryDisconnect(connectedAddress);
        this->connectedEndpoint = "";
        this->connectedAddress = "";
        this->connectionState = DISCONNECTED;
        //connector.disconnect();
    }

    std::string getConnectedEndpoint() const
    {
        std::unique_lock<Lock> guard(lock);
        return connectedEndpoint;
    }

    std::string getConnectedAddress() const
    {
        std::unique_lock<Lock> guard(lock);
        return connectedAddress;
    }

    ConnectionState getConnectionState() const
    {
        // No lock needed as performed atomically
        return connectionState;
    }

    typedef std::function<void (std::vector<zmq::message_t> &&)> MessageHandler;
    MessageHandler messageHandler;

    virtual void handleMessage(std::vector<zmq::message_t> && message)
    {
        using namespace std;
        //cerr << "named socket got message " << message.at(0).toString()
        //     << endl;
        if (messageHandler)
            messageHandler(std::move(message));
        else throw ML::Exception("need to override either messageHandler "
                                 "or handleMessage");
    }

    template<typename Fn>
    void performSocketOpSync(Fn fn)
    {
        std::unique_lock<Lock> guard(lock);
        fn(*socket);
    }

    void sendSync(std::vector<zmq::message_t> && message)
    {
        std::unique_lock<Lock> guard(lock);
        for (unsigned i = 0;  i < message.size();  ++i) {
            socket->send(message[i], i == message.size() - 1
                         ? 0 : ZMQ_SNDMORE);
        }
    }

private:
    // This lock is used to allow the synchronous methods to work without
    // needing ping-pong with the message loop.  Normally it should be
    // uncontended.
    typedef std::mutex Lock;
    mutable Lock lock;

    virtual bool doConnect(const std::string & endpointPath,
                           const std::string & entryName,
                           const Json::Value & epConfig)
    {
        using namespace std;

        //cerr << "   ((((((( doConnect for " << endpointPath << " " << entryName
        //     << endl;

        std::unique_lock<Lock> guard(lock);

        for (auto & entry: epConfig) {

            //cerr << "entry is " << entry << endl;

            if (!entry.isMember("zmqConnectUri"))
                continue;

            auto hs = entry["transports"][0]["hostScope"];
            if (!hs)
                continue;

            string hostScope = hs.asString();
            if (hs != "*") {
                utsname name;
                if (uname(&name))
                    throw ML::Exception(errno, "uname");
                if (hostScope != name.nodename)
                    continue;  // wrong host scope
            }

            string uri = entry["zmqConnectUri"].asString();

            if (connectedAddress != "") {
                // Already connected...
                if (connectedAddress == uri)
                    return true;
                else {
                    // Need to disconnect from the current address and connect to the new one
                    //cerr << "connectedAddress = " << connectedAddress << " uri = " << uri << endl;
                    socket->tryDisconnect(connectedAddress);
                    //throw ML::Exception("need to handle disconnect and reconnect to different "
                    //                    "address");
                }
            }
            
            connectedAddress = uri;
            connectedEndpointPath = endpointPath;
            connectedEntryName = entryName;
            socket->connect(uri);

            //cerr << "connection in progress to " << uri << endl;
            connectionState = CONNECTING;
            return true;
        }
        
        return false;
    }

    void handleMonitorEvent(std::string addr, int param,
                            const zmq_event_t & event)
    {
        using namespace std;

        std::unique_lock<Lock> guard(lock);

        switch (event.event) {

        case ZMQ_EVENT_CONNECTED:
            //cerr << "connecting to " << connectedAddress
            //     << " connected to " << addr << endl;
            connectionState = CONNECTED;
            connectedFd = param;
            return;

        case ZMQ_EVENT_CONNECT_DELAYED:
            //cerr << "connecting to " << connectedAddress
            //     << " connection is delayed for " << addr << endl;
            return;  // this is normal that connect not return immediately
            
        case ZMQ_EVENT_DISCONNECTED:
            //cerr << "*********** connecting to " << connectedAddress
            //     << " disconnected from " << addr << " " << this << endl;

            socket->disconnect(connectedAddress);

            // Notify the connector that the endpoint has disconnected.  It will either:
            // 1.  Call back connect again on the same address, or
            // 2.  Attempt to connect somewhere else

            connector.handleDisconnection(connectedEndpointPath, connectedEntryName);

            connectionState = DISCONNECTED;
            connectedEndpointPath = "";
            connectedAddress = "";
            connectedEntryName = "";
            return;
            
        default:
            break;
        }
        
        //cerr << "got socket event "
        //     << printZmqEvent(event.event)
        //     << " on " << addr << " with " << param;
        //if (zmqEventIsError(event.event))
        //    cerr << " " << strerror(param);
        //cerr << endl;
    }

    zmq::context_t * context;

    int socketType;
    
    std::string connectedEndpoint;

    std::string connectedAddress;

    std::string connectedEndpointPath;

    std::string connectedEntryName;

    int connectedFd;

    ConnectionState connectionState;

    EndpointConnector connector;

    std::unique_ptr<zmq::socket_t> socket;
    
    ZmqSocketMonitor monitor;
};


/*****************************************************************************/
/* ZMQ NAMED SUBSCRIBER                                                      */
/*****************************************************************************/

struct ZmqNamedSubscriber: public ZmqNamedSocket {

    ZmqNamedSubscriber(zmq::context_t & context)
        : ZmqNamedSocket(context, ZMQ_SUB)
    {
    }

    void subscribe(const std::string & prefix)
    {
        auto doSubscribe = [&] (zmq::socket_t & socket)
            {
                subscribeChannel(socket, prefix);
            };
        
        performSocketOpSync(doSubscribe);
    }

    void subscribe(const std::vector<std::string> & prefixes)
    {
        auto doSubscribe = [&] (zmq::socket_t & socket)
            {
                for (const auto& prefix : prefixes)
                    subscribeChannel(socket, prefix);
            };

        performSocketOpSync(doSubscribe);
    }

};




/*****************************************************************************/
/* SERVICE PROVIDER WATCHER                                                  */
/*****************************************************************************/

struct ServiceProviderWatcher: public MessageLoop {

    ServiceProviderWatcher()
        : currentToken(1), changes(128)
    {
    }

    ~ServiceProviderWatcher()
    {
        using namespace std;
        //cerr << "shutting down service provider watcher" << endl;
        shutdown();
        //cerr << "done" << endl;
    }

    void init(std::shared_ptr<ConfigurationService> config)
    {
        this->config = config;

        changes.onEvent
            = std::bind(&ServiceProviderWatcher::handleServiceClassChange,
                        this,
                        std::placeholders::_1);

        addSource("ServiceProviderWatcher::changes", changes);
    }

    typedef std::function<void (std::string path, bool)> ChangeHandler;
    ChangeHandler changeHandler;

    uint64_t watchServiceClass(const std::string & serviceClass,
                               ChangeHandler onChange)
    {
        using namespace std;
        //cerr << "watching service class " << serviceClass << endl;

        // Allocate a token, then push the message on to the thread to be
        // processed.
        uint64_t token = __sync_fetch_and_add(&currentToken, 1);

        std::unique_lock<Lock> guard(lock);

        auto & entry = serviceClasses[serviceClass];

        //cerr << "entry.watch = " << entry.watch << endl;

        if (!entry.watch) {
            // First time that we watch this service; set it up

            entry.watch.init([=] (const std::string & path,
                                  ConfigurationService::ChangeType change)
                             {
                                 using namespace std;
                                 //cerr << "changed path " << path << endl;

                                 changes.push(serviceClass);
                             });

            changes.push(serviceClass);
        }

        entry.entries[token].onChange = onChange;
        
        return token;
    }

    void unwatchServiceClass(const std::string & serviceClass,
                             uint64_t token)
    {
        // TODO: synchronous?  Yes, as otherwise we can't guarantee anything
        // about keeping the entries valid

        throw ML::Exception("unwatchServiceClass: not done");

#if 0        
        int done = 0;

        messages.push([&] () { this->doUnwatchServiceClass(serviceClass, token); done = 1; futex_wake(done); });

        while (!done)
            futex_wait(done, 0);
#endif
    }

private:
    typedef std::mutex Lock;
    mutable Lock lock;
    
    // Shared variable for the tokens to give out to allow unwatching
    uint64_t currentToken;

    struct WatchEntry {
        ChangeHandler onChange;
    };

    struct ServiceClassEntry {
        ConfigurationService::Watch watch;
        std::map<uint64_t, WatchEntry> entries; 

        std::vector<std::string> knownChildren;
    };
    
    std::map<std::string, ServiceClassEntry> serviceClasses;

    std::shared_ptr<ConfigurationService> config;

    void handleServiceClassChange(const std::string & serviceClass)
    {
        using namespace std;

        //cerr << "handleServiceClassChange " << serviceClass << endl;
        
        std::unique_lock<Lock> guard(lock);

        auto & entry = serviceClasses[serviceClass];

        //cerr << "handling service class change for " << serviceClass << endl;

        vector<string> children
            = config->getChildren("serviceClass/" + serviceClass,
                                  entry.watch);

        //cerr << "children = " << children << endl;

        std::sort(children.begin(), children.end());

        auto & knownChildren = entry.knownChildren;

        // Perform a diff between previously and currently known children
        vector<string> addedChildren, deletedChildren;
        std::set_difference(children.begin(), children.end(),
                            knownChildren.begin(), knownChildren.end(),
                            std::back_inserter(addedChildren));
        std::set_difference(knownChildren.begin(), knownChildren.end(),
                            children.begin(), children.end(),
                            std::back_inserter(deletedChildren));

        knownChildren.swap(children);

        guard.unlock();

        for (auto & c: addedChildren) {
            for (auto & e: entry.entries) {
                e.second.onChange("serviceClass/" + serviceClass + "/" + c,
                                  true);
            }
        }

        for (auto & c: deletedChildren) {
            for (auto & e: entry.entries) {
                e.second.onChange("serviceClass/" + serviceClass + "/" + c,
                                  false);
            }
        }
    }

    TypedMessageSink<std::string> changes;
};


/*****************************************************************************/
/* ZMQ NAMED MULTIPLE SUBSCRIBER                                             */
/*****************************************************************************/

struct ZmqNamedMultipleSubscriber: public MessageLoop {

    ZmqNamedMultipleSubscriber(std::shared_ptr<zmq::context_t> context)
        : context(context)
    {
    }

    ~ZmqNamedMultipleSubscriber()
    {
        shutdown();
    }

    void init(std::shared_ptr<ConfigurationService> config)
    {
        this->config = config;

        serviceWatcher.init(config);

        addSource("ZmqNamedMultipleSubscriber", serviceWatcher);

        //debug(true);
    }

    void shutdown()
    {
        MessageLoop::shutdown();

        for (auto & sub: subscribers)
            sub.second->shutdown();
        subscribers.clear();
    }

    void connectAllServiceProviders(const std::string & serviceClass,
                                    const std::string & endpointName,
                                    const std::vector<std::string> & prefixes
                                        = std::vector<std::string>())
    {
        auto onServiceChange = [=] (const std::string & service,
                                    bool created)
            {
                using namespace std;
                //cerr << "onServiceChange " << serviceClass << " " << endpointName
                //<< " " << service << " created " << created << endl;

                if (created)
                    connectService(serviceClass, service, endpointName);
                else
                    disconnectService(serviceClass, service, endpointName);
            };
        
        this->prefixes = prefixes;
        serviceWatcher.watchServiceClass(serviceClass, onServiceChange);
    }
    
    typedef std::function<void (std::vector<zmq::message_t> &&)> MessageHandler;
    MessageHandler messageHandler;

    virtual void handleMessage(std::vector<zmq::message_t> && message)
    {
        if (messageHandler)
            messageHandler(std::move(message));
        else throw ML::Exception("need to override either messageHandler "
                                 "or handleMessage");
    }

    void connectService(std::string serviceClass, std::string service,
                        std::string endpointName)
    {
        using namespace std;

        std::unique_lock<Lock> guard(lock);

        SubscriberMap::const_iterator found = subscribers.find(service);
        if(found != subscribers.end())
        {
           if(found->second->getConnectionState() == ZmqNamedSocket::CONNECTED)
           {
               std::cerr << "Already connected to service " << service << std::endl;
               return;
           }
           else
           {
             std::cerr << "we already had a connection entry to service " << service <<" - reuse " << std::endl;
             Json::Value value = config->getJson(service);
             std::string path = value["servicePath"].asString();
             found->second->connectToEndpoint(path); 
             return ;
           }
        } 
         
        std::unique_ptr<ZmqNamedSubscriber> sub
            (new ZmqNamedSubscriber(*context));

        // Set up to handle the message
        sub->messageHandler = [=] (std::vector<zmq::message_t> && msg)
            {
                //cerr << "SUB MESSAGE HANDLER " << this << endl;
                this->handleMessage(std::move(msg));
            };
                                        
        sub->init(config);

        // TODO: put a watch in to reconnect if this changes
        Json::Value value = config->getJson(service);
        std::string path = value["servicePath"].asString();

        //cerr << "(((((((((((( connecting to service " << service
        //     << " at endpoint " << path + "/" + endpointName << endl;
        
        //cerr << "+-+-+-+-+- connecting to endpoint " << path + "/" + endpointName << endl;
        //cerr << config->getChildren(path + "/" + endpointName) << endl;

        if (!prefixes.empty())
            sub->subscribe(prefixes);
        else sub->subscribe(""); // Subscribe to everything.

        sub->connectToEndpoint(path + "/" + endpointName);
        addSourceDeferred(service, *sub);
        subscribers[service] = std::move(sub);
    }

    void disconnectService(std::string serviceClass, std::string service,
                           std::string endpointName)
    {
        using namespace std;
        cerr << "need to disconenct from " << serviceClass << " "
             << service << " " << endpointName << endl;
//        cerr << "aborting as disconnect not done yet" << endl;
        SubscriberMap::const_iterator found = subscribers.find(service);
        if(found != subscribers.end())
        {
           found->second->disconnect();
        } 
        //abort();
    }

    ServiceProviderWatcher serviceWatcher;
             
    std::shared_ptr<zmq::context_t> context;

    typedef std::mutex Lock;
    mutable Lock lock;

    typedef std::map<std::string, std::unique_ptr<ZmqNamedSubscriber> > SubscriberMap; 
    SubscriberMap subscribers;

    std::shared_ptr<ConfigurationService> config;

    std::vector<std::string> prefixes;
};


} // namespace Datacratic