Commit 924d8cfe authored by legoc's avatar legoc
Browse files

Renamed ContextImpl into ContextZmq

parent ae998829
......@@ -37,7 +37,7 @@ namespace application {
class EventListener;
class EventThread;
class ContextImpl;
class ContextZmq;
class RequestSocket;
class Server {
......@@ -166,7 +166,7 @@ private:
Endpoint m_serverEndpoint;
std::array<int, 3> m_serverVersion;
int m_statusPort;
std::unique_ptr<ContextImpl> m_contextImpl;
std::unique_ptr<ContextZmq> m_contextImpl;
std::unique_ptr<RequestSocket> m_requestSocket;
std::mutex m_eventListenersMutex;
......
......@@ -29,7 +29,7 @@
#include "impl/StreamSocketImpl.h"
#include "impl/WaitingImpl.h"
#include "impl/WaitingImplSet.h"
#include "impl/ContextImpl.h"
#include "impl/zmq/ContextZmq.h"
#include "Strings.h"
#include "Server.h"
#include "Messages.h"
......
......@@ -19,7 +19,7 @@
#include "JSON.h"
#include "impl/SocketWaitingImpl.h"
#include "impl/StreamSocketImpl.h"
#include "impl/ContextImpl.h"
#include "impl/zmq/ContextZmq.h"
#include "Messages.h"
#include <iostream>
......
......@@ -17,7 +17,6 @@
#include "RequestSocket.h"
#include "ConnectionTimeout.h"
#include "impl/ContextImpl.h"
#include "impl/zmq/RequestSocketZmq.h"
#include <iostream>
#include <chrono>
......
......@@ -23,7 +23,7 @@
#include "EventThread.h"
#include "impl/CancelIdGenerator.h"
#include "impl/StreamSocketImpl.h"
#include "impl/ContextImpl.h"
#include "impl/zmq/ContextZmq.h"
#include "JSON.h"
#include "Messages.h"
#include "RequestSocket.h"
......@@ -636,7 +636,7 @@ void Server::unregisterEventListener(EventListener * listener) {
void Server::initContext() {
// Set the impl.
m_contextImpl.reset(new ContextImpl());
m_contextImpl.reset(new ContextZmq());
}
void Server::initRequestSocket() {
......
......@@ -14,37 +14,36 @@
* limitations under the Licence.
*/
#include "ContextZmq.h"
#include "SocketException.h"
#include "ConnectionTimeout.h"
#include "JSON.h"
#include "../Messages.h"
#include "../../Messages.h"
#include "../../RequestSocket.h"
#include <iostream>
#include <sstream>
#include "ContextImpl.h"
#include "../RequestSocket.h"
using namespace std;
namespace cameo {
ContextImpl::ContextImpl() : Context(),
ContextZmq::ContextZmq() : Context(),
m_context(1), m_timeout(0) {
}
ContextImpl::~ContextImpl() {
ContextZmq::~ContextZmq() {
}
void ContextImpl::setTimeout(int timeout) {
void ContextZmq::setTimeout(int timeout) {
m_timeout = timeout;
}
int ContextImpl::getTimeout() const {
int ContextZmq::getTimeout() const {
return m_timeout;
}
zmq::socket_t * ContextImpl::createEventSubscriber(const std::string& endpoint, const std::string& cancelEndpoint) {
zmq::socket_t * ContextZmq::createEventSubscriber(const std::string& endpoint, const std::string& cancelEndpoint) {
zmq::socket_t * subscriber = new zmq::socket_t(m_context, ZMQ_SUB);
......@@ -66,7 +65,7 @@ zmq::socket_t * ContextImpl::createEventSubscriber(const std::string& endpoint,
return subscriber;
}
zmq::socket_t * ContextImpl::createOutputStreamSubscriber(const std::string& endpoint, const std::string& cancelEndpoint) {
zmq::socket_t * ContextZmq::createOutputStreamSubscriber(const std::string& endpoint, const std::string& cancelEndpoint) {
zmq::socket_t * subscriber = new zmq::socket_t(m_context, ZMQ_SUB);
......@@ -86,7 +85,7 @@ zmq::socket_t * ContextImpl::createOutputStreamSubscriber(const std::string& end
return subscriber;
}
zmq::socket_t * ContextImpl::createCancelPublisher(const std::string& endpoint) {
zmq::socket_t * ContextZmq::createCancelPublisher(const std::string& endpoint) {
zmq::socket_t * publisher = new zmq::socket_t(m_context, ZMQ_PUB);
publisher->bind(endpoint.c_str());
......@@ -94,7 +93,7 @@ zmq::socket_t * ContextImpl::createCancelPublisher(const std::string& endpoint)
return publisher;
}
zmq::socket_t * ContextImpl::createRequestSocket(const std::string& endpoint) {
zmq::socket_t * ContextZmq::createRequestSocket(const std::string& endpoint) {
zmq::socket_t* socket = new zmq::socket_t(m_context, ZMQ_REQ);
......@@ -113,7 +112,7 @@ zmq::socket_t * ContextImpl::createRequestSocket(const std::string& endpoint) {
return socket;
}
bool ContextImpl::isAvailable(RequestSocket * socket, int timeout) {
bool ContextZmq::isAvailable(RequestSocket * socket, int timeout) {
try {
socket->requestJSON(createSyncRequest(), timeout);
......@@ -129,7 +128,7 @@ bool ContextImpl::isAvailable(RequestSocket * socket, int timeout) {
return false;
}
void ContextImpl::sendSyncStream(RequestSocket * socket, const std::string& name) {
void ContextZmq::sendSyncStream(RequestSocket * socket, const std::string& name) {
try {
socket->requestJSON(createSyncStreamRequest(name));
......@@ -142,7 +141,7 @@ void ContextImpl::sendSyncStream(RequestSocket * socket, const std::string& name
}
}
void ContextImpl::waitForStreamSubscriber(zmq::socket_t * subscriber, RequestSocket * socket, const std::string& name) {
void ContextZmq::waitForStreamSubscriber(zmq::socket_t * subscriber, RequestSocket * socket, const std::string& name) {
// Poll subscriber.
zmq_pollitem_t items[1];
......@@ -162,7 +161,7 @@ void ContextImpl::waitForStreamSubscriber(zmq::socket_t * subscriber, RequestSoc
}
}
void ContextImpl::waitForSubscriber(zmq::socket_t * subscriber, RequestSocket * socket) {
void ContextZmq::waitForSubscriber(zmq::socket_t * subscriber, RequestSocket * socket) {
// Poll subscriber.
zmq_pollitem_t items[1];
......
......@@ -14,23 +14,23 @@
* limitations under the Licence.
*/
#ifndef CAMEO_CONTEXTIMPL_H_
#define CAMEO_CONTEXTIMPL_H_
#ifndef CAMEO_CONTEXTZMQ_H_
#define CAMEO_CONTEXTZMQ_H_
#include "Context.h"
#include "zmq.hpp"
#include <vector>
#include <memory>
#include <zmq.hpp>
namespace cameo {
class RequestSocket;
class ContextImpl : public Context {
class ContextZmq : public Context {
public:
ContextImpl();
virtual ~ContextImpl();
ContextZmq();
virtual ~ContextZmq();
void setTimeout(int timeout);
int getTimeout() const;
......
......@@ -17,7 +17,7 @@
#include "RequestSocketZmq.h"
#include "ConnectionTimeout.h"
#include "../ContextImpl.h"
#include "ContextZmq.h"
#include <iostream>
#include <chrono>
#include <thread>
......@@ -27,7 +27,7 @@ using namespace std;
namespace cameo {
RequestSocketZmq::RequestSocketZmq(Context * context, const std::string& endpoint, int timeout) :
m_services(dynamic_cast<ContextImpl *>(context)), m_endpoint(endpoint) {
m_services(dynamic_cast<ContextZmq *>(context)), m_endpoint(endpoint) {
init();
......
......@@ -24,7 +24,7 @@
namespace cameo {
class Context;
class ContextImpl;
class ContextZmq;
class RequestSocketZmq : public RequestSocketImpl {
......@@ -38,7 +38,7 @@ public:
virtual std::string request(const std::string& requestPart1, const std::string& requestPart2, int overrideTimeout);
virtual std::string request(const std::string& requestPart1, const std::string& requestPart2, const std::string& requestPart3, int overrideTimeout);
ContextImpl * m_services;
ContextZmq * m_services;
std::string m_endpoint;
std::unique_ptr<zmq::socket_t> m_socket;
int m_timeout;
......
......@@ -16,13 +16,13 @@
#include "PublisherSubscriber.h"
#include "../base/impl/ContextImpl.h"
#include "JSON.h"
#include "Server.h"
#include "../base/impl/zmq/ContextZmq.h"
#include "../base/Messages.h"
#include "../base/RequestSocket.h"
#include "impl/PublisherImpl.h"
#include "impl/SubscriberImpl.h"
#include "JSON.h"
#include "Server.h"
namespace cameo {
namespace coms {
......
......@@ -16,14 +16,14 @@
#include "RequesterResponder.h"
#include "../base/impl/ContextImpl.h"
#include "JSON.h"
#include "Server.h"
#include "../base/impl/zmq/ContextZmq.h"
#include "../base/Messages.h"
#include "../base/RequestSocket.h"
#include "impl/RequesterImpl.h"
#include "impl/RequestImpl.h"
#include "impl/ResponderImpl.h"
#include "JSON.h"
#include "Server.h"
namespace cameo {
namespace coms {
......
......@@ -19,7 +19,7 @@
#include "Application.h"
#include "Serializer.h"
#include "JSON.h"
#include "../../base/impl/ContextImpl.h"
#include "../../base/impl/zmq/ContextZmq.h"
#include "../../base/Messages.h"
#include "../../base/RequestSocket.h"
#include <sstream>
......@@ -34,7 +34,7 @@ PublisherImpl::PublisherImpl(int publisherPort, int synchronizerPort, const std:
m_ended(false) {
// create a socket for publishing
ContextImpl* contextImpl = dynamic_cast<ContextImpl *>(application::This::getCom().getContext());
ContextZmq* contextImpl = dynamic_cast<ContextZmq *>(application::This::getCom().getContext());
m_publisher.reset(new zmq::socket_t(contextImpl->m_context, ZMQ_PUB));
std::stringstream pubEndpoint;
pubEndpoint << "tcp://*:" << publisherPort;
......@@ -69,7 +69,7 @@ bool PublisherImpl::waitForSubscribers() {
}
// Create a socket to receive the messages from the subscribers.
ContextImpl* contextImpl = dynamic_cast<ContextImpl *>(application::This::getCom().getContext());
ContextZmq* contextImpl = dynamic_cast<ContextZmq *>(application::This::getCom().getContext());
zmq::socket_t synchronizer(contextImpl->m_context, ZMQ_REP);
std::stringstream syncEndpoint;
......
......@@ -19,7 +19,7 @@
#include "Application.h"
#include "Serializer.h"
#include "JSON.h"
#include "../../base/impl/ContextImpl.h"
#include "../../base/impl/zmq/ContextZmq.h"
#include "../../base/Messages.h"
#include "../../base/RequestSocket.h"
#include <sstream>
......
......@@ -19,7 +19,7 @@
#include "Application.h"
#include "Serializer.h"
#include "JSON.h"
#include "../../base/impl/ContextImpl.h"
#include "../../base/impl/zmq/ContextZmq.h"
#include "../../base/Messages.h"
#include "../../base/RequestSocket.h"
#include <sstream>
......@@ -42,7 +42,7 @@ RequesterImpl::RequesterImpl(const Endpoint& endpoint, int requesterPort, int re
m_requestSocket = application::This::getCom().createRequestSocket(endpoint.withPort(responderPort).toString());
// Create a socket REP.
ContextImpl* contextImpl = dynamic_cast<ContextImpl *>(application::This::getCom().getContext());
ContextZmq* contextImpl = dynamic_cast<ContextZmq *>(application::This::getCom().getContext());
m_repSocket.reset(new zmq::socket_t(contextImpl->m_context, ZMQ_REP));
std::stringstream reqEndpoint;
reqEndpoint << "tcp://*:" << m_requesterPort;
......
......@@ -20,7 +20,7 @@
#include "Application.h"
#include "Serializer.h"
#include "JSON.h"
#include "../../base/impl/ContextImpl.h"
#include "../../base/impl/zmq/ContextZmq.h"
#include "../../base/Messages.h"
#include "../../base/RequestSocket.h"
#include <sstream>
......@@ -36,7 +36,7 @@ ResponderImpl::ResponderImpl(int responderPort, const std::string& name) :
m_canceled(false) {
// create a socket REP
ContextImpl* contextImpl = dynamic_cast<ContextImpl *>(application::This::getCom().getContext());
ContextZmq* contextImpl = dynamic_cast<ContextZmq *>(application::This::getCom().getContext());
m_responder.reset(new zmq::socket_t(contextImpl->m_context, ZMQ_REP));
std::stringstream repEndpoint;
repEndpoint << "tcp://*:" << m_responderPort;
......
......@@ -20,7 +20,7 @@
#include "Server.h"
#include "JSON.h"
#include "../../base/impl/CancelIdGenerator.h"
#include "../../base/impl/ContextImpl.h"
#include "../../base/impl/zmq/ContextZmq.h"
#include "../../base/Messages.h"
#include "../../base/RequestSocket.h"
#include <sstream>
......@@ -47,7 +47,7 @@ SubscriberImpl::~SubscriberImpl() {
void SubscriberImpl::init() {
// Create a socket for publishing.
ContextImpl* contextImpl = dynamic_cast<ContextImpl *>(application::This::getCom().getContext());
ContextZmq* contextImpl = dynamic_cast<ContextZmq *>(application::This::getCom().getContext());
m_subscriber.reset(new zmq::socket_t(contextImpl->m_context, ZMQ_SUB));
m_subscriber->setsockopt(ZMQ_SUBSCRIBE, message::Event::SYNC, std::string(message::Event::SYNC).length());
m_subscriber->setsockopt(ZMQ_SUBSCRIBE, message::Event::STREAM, std::string(message::Event::STREAM).length());
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment