variable_io.hpp

Go to the documentation of this file.

#ifndef CORAL_BUS_VARIABLE_IO_HPP_INCLUDED
#define CORAL_BUS_VARIABLE_IO_HPP_INCLUDED

#include <chrono>
#include <memory>
#include <queue>
#include <string>
#include <unordered_map>

#include "coral/model.hpp"
#include "coral/net.hpp"


// Forward declaration to avoid dependency on ZMQ headers
namespace zmq { class socket_t; }


namespace coral
{
namespace bus
{


class VariablePublisher
{
public:
    VariablePublisher();

    void Bind(const coral::net::Endpoint& endpoint);

    coral::net::Endpoint BoundEndpoint() const;

    void Publish(
        coral::model::StepID stepID,
        coral::model::SlaveID slaveID,
        coral::model::VariableID variableID,
        coral::model::ScalarValue value);

private:
    std::unique_ptr<zmq::socket_t> m_socket;
};


class VariableSubscriber
{
public:
    VariableSubscriber();

    void Connect(
        const coral::net::Endpoint* endpoints,
        std::size_t endpointsSize);

    void Subscribe(const coral::model::Variable& variable);

    void Unsubscribe(const coral::model::Variable& variable);

    bool Update(
        coral::model::StepID stepID,
        std::chrono::milliseconds timeout);

    const coral::model::ScalarValue& Value(const coral::model::Variable& variable)
        const;

private:
    typedef std::queue<std::pair<coral::model::StepID, coral::model::ScalarValue>>
        ValueQueue;

    // A hash function for Variable objects, so we can put them in a
    // std::unordered_map (below)
    struct VariableHash
    {
        std::size_t operator()(const coral::model::Variable& v) const
        {
            return static_cast<std::size_t>((v.Slave() << 16) + v.ID());
        }
    };

    coral::model::StepID m_currentStepID;
    std::unique_ptr<zmq::socket_t> m_socket;
    std::unordered_map<coral::model::Variable, ValueQueue, VariableHash> m_values;
};


}} // namespace
#endif // header guard