Commit f23dcf7f authored by fred's avatar fred
Browse files

zapi class to communicate with zerogw, extract ftp server

parent a5a85d42
......@@ -392,6 +392,137 @@ void Ztracker::destructor()
}
Zapi::Zapi(zmq::context_t *a_context) : m_context(a_context)
{
std::cout << "Zapi::Zapi constructeur" << std::endl;
m_mutex_http = new QMutex();
m_socket_http = new zmq::socket_t (*m_context, ZMQ_REP);
int hwm = 50000;
m_socket_http->setsockopt(ZMQ_SNDHWM, &hwm, sizeof (hwm));
m_socket_http->setsockopt(ZMQ_RCVHWM, &hwm, sizeof (hwm));
m_socket_http->connect("tcp://*:2503");
int http_socket_fd;
size_t socket_size = sizeof(http_socket_fd);
m_socket_http->getsockopt(ZMQ_FD, &http_socket_fd, &socket_size);
qDebug() << "RES getsockopt : " << "res" << " FD : " << http_socket_fd << " errno : " << zmq_strerror (errno);
check_http_data = new QSocketNotifier(http_socket_fd, QSocketNotifier::Read, this);
connect(check_http_data, SIGNAL(activated(int)), this, SLOT(receive_http_payload()));
}
Zapi::~Zapi()
{
qDebug() << "Zapi::~Zapi";
delete(m_socket_http);
}
void Zapi::receive_http_payload()
{
m_mutex_http->lock();
check_http_data->setEnabled(false);
std::cout << "Zapi::receive_payload" << std::endl;
qint32 events = 0;
std::size_t eventsSize = sizeof(events);
m_socket_http->getsockopt(ZMQ_EVENTS, &events, &eventsSize);
std::cout << "Zapi::receive_payload ZMQ_EVENTS : " << events << std::endl;
if (events & ZMQ_POLLIN)
{
std::cout << "Zapi::receive_payload ZMQ_POLLIN" << std::endl;
while (true)
{
flush_socket:
// Initialize poll set
// zmq::pollitem_t items = { *m_socket, 0, ZMQ_POLLIN, 0 };
//while (true) {
// Wait for next request from client
zmq::message_t request;
// zmq::poll (&items, 1, 5000);
// if (items.revents & ZMQ_POLLIN)
// {
bool res = m_socket_http->recv(&request, ZMQ_NOBLOCK);
if (!res && zmq_errno () == EAGAIN) break;
std::cout << "Zapi::receive_payload received request: [" << (char*) request.data() << "]" << std::endl;
if (request.size() == 0) {
std::cout << "Zapi::worker_response received request 0" << std::endl;
break;
}
//m_socket->recv (&request, ZMQ_NOBLOCK);
BSONObj data;
try {
data = mongo::fromjson((char*)request.data());
if (data.isValid() && !data.isEmpty())
{
std::cout << "Zapi received : " << res << " data : " << data << std::endl;
std::cout << "!!!!!!! BEFORE FORWARD PAYLOAD !!!!" << std::endl;
//emit forward_payload(data.copy());
std::cout << "!!!!!!! AFTER FORWARD PAYLOAD !!!!" << std::endl;
}
else
{
std::cout << "DATA NO VALID !" << std::endl;
}
}
catch (mongo::MsgAssertionException &e)
{
std::cout << "error on data : " << data << std::endl;
std::cout << "error on data BSON : " << e.what() << std::endl;
goto flush_socket;
}
}
}
check_http_data->setEnabled(true);
m_mutex_http->unlock();
}
void Zapi::destructor()
{
qDebug() << "Zapi destructor";
m_mutex_http->lock();
check_http_data->setEnabled(false);
m_socket_http->close ();
}
Zpull::Zpull(QString base_directory, zmq::context_t *a_context) : m_context(a_context)
{
......@@ -1845,6 +1976,17 @@ void Zeromq::init()
thread_api = new QThread;
zapi = new Zapi(m_context);
zapi->moveToThread(thread_api);
thread_api->start();
connect(this, SIGNAL(shutdown()), zapi, SLOT(destructor()), Qt::BlockingQueuedConnection);
connect(zapi, SIGNAL(forward_payload(BSONObj)), dispatch, SLOT(push_payload(BSONObj)), Qt::QueuedConnection);
thread_stream = new QThread;
stream_push = new Zstream_push(m_context);
stream_push->moveToThread(thread_stream);
......
......@@ -133,6 +133,30 @@ private slots:
typedef QSharedPointer<Zworker_push> Zworker_pushPtr;
class Zapi : public QThread
{
Q_OBJECT
public:
Zapi(zmq::context_t *a_context);
~Zapi();
private:
QSocketNotifier *check_http_data;
zmq::context_t *m_context;
zmq::socket_t *m_socket_http;
QMutex *m_mutex_http;
signals:
void forward_payload(BSONObj data);
public slots:
void destructor();
private slots:
void receive_http_payload();
};
class Zpull : public QThread
{
Q_OBJECT
......@@ -140,7 +164,7 @@ public:
Zpull(QString base_directory, zmq::context_t *a_context);
~Zpull();
private:
private:
QSocketNotifier *check_http_data;
QSocketNotifier *check_zeromq_data;
QSocketNotifier *check_worker_response;
......@@ -216,6 +240,7 @@ public:
zmq::context_t *m_context;
Zapi *zapi;
Zpull *pull;
Zdispatch *dispatch;
Ztracker *ztracker;
......@@ -233,7 +258,7 @@ private:
QThread *thread_tracker;
QThread *thread_pull;
QThread *thread_stream;
QThread *thread_api;
zmq::socket_t *z_workers;
QTimer *pull_timer;
......
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