core/agent_configuration/agent_configuration_listener.h

/* configuration_listener.h                                        -*- C++ -*-
   Jeremy Barnes, 26 November 2012
   Copyright (c) 2012 Datacratic Inc.  All rights reserved.

   Agent that listens for configuration.
*/

#pragma once

#include "soa/service/zmq_endpoint.h"
#include "rtbkit/core/router/router_types.h"
#include "soa/gc/rcu_protected.h"


namespace RTBKIT {


/*****************************************************************************/
/* AGENT CONFIG ENTRY                                                        */
/*****************************************************************************/

struct AgentConfigEntry {
    std::string name;
    std::shared_ptr<const AgentConfig> config;

    bool valid() const { return !!config; }

    Json::Value toJson() const;
};


struct AllAgentConfig : public std::vector<AgentConfigEntry> {
    std::unordered_map<std::string, int> agentIndex;
    std::unordered_map<AccountKey, std::vector<int> > accountIndex;
};


/*****************************************************************************/
/* AGENT CONFIGURATION LISTENER                                              */
/*****************************************************************************/

struct AgentConfigurationListener: public MessageLoop {

    AgentConfigurationListener(std::shared_ptr<zmq::context_t> context)
        : allAgents(new AllAgentConfig()), configEndpoint(context)
    {
    }

    ~AgentConfigurationListener()
    {
        shutdown();
    }

    typedef std::function<void (std::string, std::shared_ptr<const AgentConfig>)>
    OnConfigChange;

    OnConfigChange onConfigChange;

    void init(std::shared_ptr<ConfigurationService> config)
    {
        configEndpoint.init(config);
        configEndpoint.connectToServiceClass("rtbAgentConfiguration",
                                             "listen");
        configEndpoint.messageHandler
            = [=] (const std::vector<std::string> & message)
            {
                if (message.size() > 1) {
                    using namespace std;
                    cerr << "got configuration message " << message[1] << endl;
                }
                this->onMessage(message);
            };
        addSource("AgentConfigurationListener::configEndpoint", configEndpoint);
    }

    void start()
    {
        MessageLoop::start();
    }

    void shutdown()
    {
        MessageLoop::shutdown();
        configEndpoint.shutdown();
    }

    /*************************************************************************/
    /* AGENT INTERACTIONS                                                    */
    /*************************************************************************/

    typedef std::function<void (const AgentConfigEntry & info)> OnAgentFn;
    void forEachAgent(const OnAgentFn & onAgent) const;

    void forEachAccountAgent(const AccountKey & account,
                             const OnAgentFn & onAgent) const;

    AgentConfigEntry getAgentEntry(const std::string & agent) const;

private:
    void onMessage(const std::vector<std::string> & message);

    AllAgentConfig * allAgents;
    mutable GcLock allAgentsGc;

    ZmqNamedClientBusProxy configEndpoint;
};


} // namespace RTBKIT