Commit 50d589e0 authored by fred's avatar fred
Browse files

added a zm proxy to improve api payload

parent bc5506de
......@@ -20,6 +20,58 @@
#include "service.h"
ZerogwProxy::ZerogwProxy(params a_ncs_params, QObject *parent) : m_ncs_params(a_ncs_params), QObject(parent)
{
thread = new QThread;
this->moveToThread(thread);
thread->start();
thread->connect(thread, SIGNAL(started()), this, SLOT(init()));
qDebug() << "ZerogwProxy::ZerogwProxy";
zeromq_ = Zeromq::getInstance ();
qDebug() << "ZerogwProxy::getInstance";
}
ZerogwProxy::~ZerogwProxy()
{
qDebug() << "!!!!!!! ZerogwProxy::CLOSE !!!!!!!!!!!!!!!!!!!!!!!!!!!!!";
delete(api_payload);
delete(api_payload2);
zerogw->close();
worker_payload->close();
}
void ZerogwProxy::init()
{
qDebug() << "ZerogwProxy::init";
zerogw = new zmq::socket_t(*zeromq_->m_context, ZMQ_ROUTER);
zerogw->bind ("tcp://*:2503");
worker_payload = new zmq::socket_t(*zeromq_->m_context, ZMQ_DEALER);
QString uri = "ipc://" + m_ncs_params.base_directory + "/payloads";
worker_payload->bind (uri.toLatin1());
QThread *thread_api_payload = new QThread;
api_payload = new Api_payload(m_ncs_params.base_directory, 0);
api_payload->moveToThread(thread_api_payload);
thread_api_payload->start();
QThread *thread_api_payload2 = new QThread;
api_payload2 = new Api_payload(m_ncs_params.base_directory, 0);
api_payload2->moveToThread(thread_api_payload2);
thread_api_payload2->start();
qDebug() << "ZerogwProxy::init BEFORE PROXY";
zmq::proxy (*zerogw, *worker_payload, NULL);
qDebug() << "ZerogwProxy::init AFTER PROXY";
}
Service::Service(params a_ncs_params, QObject *parent) : m_ncs_params(a_ncs_params), QObject(parent)
{
m_nodetrack = NULL;
......@@ -27,6 +79,8 @@ Service::Service(params a_ncs_params, QObject *parent) : m_ncs_params(a_ncs_para
m_http_api = NULL;
m_xmpp_server = NULL;
m_xmpp_client = NULL;
zerogwToPayload = NULL;
}
Service::~Service()
......@@ -53,7 +107,8 @@ Service::~Service()
qDebug() << "delete worker api";
emit shutdown();
delete(worker_api);
delete(api_payload);
//delete(api_payload);
if (zerogwToPayload) delete(zerogwToPayload);
delete(api_node);
delete(api_workflow);
delete(api_user);
......@@ -79,13 +134,37 @@ void Service::Http_api_init()
void Service::Http_api_init()
{
qDebug() << "Service::Http_api_init";
int port;
m_ncs_params.api_port == 0 ? port = 2502 : port = m_ncs_params.api_port;
zerogwToPayload = new ZerogwProxy(m_ncs_params);
/*
zmq::socket_t zerogw (*zeromq_->m_context, ZMQ_ROUTER);
zerogw.bind ("tcp://*:2503");
zmq::socket_t worker_payload (*zeromq_->m_context, ZMQ_DEALER);
worker_payload.bind ("inproc://workers");
qDebug() << "Service::Http_api_init BEFORE PROXY";
zmq::proxy (zerogw, worker_payload, NULL);
qDebug() << "Service::Http_api_init AFTER PROXY";
QThread *thread_api_payload = new QThread;
api_payload = new Api_payload(m_ncs_params.base_directory, port);
api_payload->moveToThread(thread_api_payload);
thread_api_payload->start();
QThread *thread_api_payload2 = new QThread;
api_payload2 = new Api_payload(m_ncs_params.base_directory, port + 101);
api_payload2->moveToThread(thread_api_payload2);
thread_api_payload2->start();
*/
//connect(this, SIGNAL(shutdown()), api_payload, SLOT(destructor()), Qt::BlockingQueuedConnection);
//connect(api_payload, SIGNAL(forward_payload(BSONObj)), dispatch, SLOT(push_payload(BSONObj)), Qt::QueuedConnection);
......
......@@ -37,6 +37,26 @@
#include "ncs_global.h"
#include "zerogw.h"
class ZerogwProxy : public QObject
{
Q_OBJECT
public:
ZerogwProxy(params a_ncs_params, QObject *parent = 0);
~ZerogwProxy();
private slots:
void init();
private:
QThread *thread;
Zeromq *zeromq_;
zmq::socket_t *zerogw;
zmq::socket_t *worker_payload;
Api_payload *api_payload;
Api_payload *api_payload2;
params m_ncs_params;
};
class Service : public QObject
{
Q_OBJECT
......@@ -53,12 +73,13 @@ public:
void Worker_init();
void link();
Worker_api *worker_api;
Api_payload *api_payload;
ZerogwProxy *zerogwToPayload;
Api_node *api_node;
Api_workflow *api_workflow;
Api_user *api_user;
private:
Http_admin *m_http_admin;
Nodeftp *m_nodeftp;
Http_api *m_http_api;
......
......@@ -24,7 +24,7 @@
Zerogw::Zerogw(QString basedirectory, int port, QObject *parent) : m_basedirectory(basedirectory), QObject(parent)
{
std::cout << "Zerogw::Zerogw constructeur" << std::endl;
std::cout << "Zerogw::Zerogw construct" << std::endl;
mongodb_ = Mongodb::getInstance();
zeromq_ = Zeromq::getInstance ();
......@@ -52,8 +52,18 @@ Zerogw::Zerogw(QString basedirectory, int port, QObject *parent) : m_basedirecto
int hwm = 50000;
m_socket_zerogw->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
m_socket_zerogw->setsockopt(ZMQ_RCVHWM, &hwm, sizeof (hwm));
QString uri = "tcp://*:" + QString::number(port);
m_socket_zerogw->bind(uri.toAscii());
if (port == 0 )
{
// QString uri = "tcp://127.0.0.1:2504" + QString::number(port);
QString uri = "ipc://" + m_basedirectory + "/payloads";
m_socket_zerogw->connect(uri.toAscii());
}
else
{
QString uri = "tcp://*:" + QString::number(port);
m_socket_zerogw->bind(uri.toAscii());
}
int http_socket_fd;
size_t socket_size = sizeof(http_socket_fd);
......@@ -205,8 +215,6 @@ Api_payload::Api_payload(QString basedirectory, int port) : Zerogw(basedirectory
void Api_payload::receive_http_payload()
{
m_mutex->lock();
check_http_data->setEnabled(false);
std::cout << "Api_payload::receive_payload" << std::endl;
......@@ -578,10 +586,7 @@ void Api_payload::receive_http_payload()
}
counter++;
}
qDebug() << "ZEROGW : " << zerogw;
check_http_data->setEnabled(true);
m_mutex->unlock();
}
......@@ -601,8 +606,6 @@ Api_node::Api_node(QString basedirectory, int port) : Zerogw(basedirectory, port
void Api_node::receive_http_payload()
{
m_mutex->lock();
check_http_data->setEnabled(false);
std::cout << "Api_node::receive_payload" << std::endl;
......@@ -720,7 +723,6 @@ void Api_node::receive_http_payload()
}
qDebug() << "ZEROGW : " << zerogw;
check_http_data->setEnabled(true);
m_mutex->unlock();
}
......@@ -739,8 +741,6 @@ Api_workflow::Api_workflow(QString basedirectory, int port) : Zerogw(basedirecto
void Api_workflow::receive_http_payload()
{
m_mutex->lock();
check_http_data->setEnabled(false);
std::cout << "Api_workflow::receive_payload" << std::endl;
......@@ -853,7 +853,6 @@ void Api_workflow::receive_http_payload()
}
qDebug() << "ZEROGW : " << zerogw;
check_http_data->setEnabled(true);
m_mutex->unlock();
}
......@@ -870,8 +869,6 @@ Api_user::Api_user(QString basedirectory, int port) : Zerogw(basedirectory, port
void Api_user::receive_http_payload()
{
m_mutex->lock();
check_http_data->setEnabled(false);
std::cout << "Api_user::receive_payload" << std::endl;
......@@ -1029,7 +1026,6 @@ void Api_user::receive_http_payload()
}
qDebug() << "ZEROGW : " << zerogw;
check_http_data->setEnabled(true);
m_mutex->unlock();
}
......
Supports Markdown
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