ZMQ testing

zmq_test.cpp

// ZMQ_test.cpp : Defines the entry point for the console application.
//

#include "zmq.hpp"
#include <iostream>
#include <boost\thread.hpp>
#include "Poco\Process.h"
#include <sstream>

#ifndef _WIN32
#include <unistd.h>
#else
#include <windows.h>
#endif

using namespace std;

void thread(void* params)
{
    zmq::context_t* context((zmq::context_t*)params);
    zmq::socket_t socket (*context, ZMQ_SUB);
    socket.connect ("inproc://blabla2");
	socket.setsockopt(ZMQ_SUBSCRIBE, nullptr, 0);
	stringstream ss;
	for (int i = 0; i < 100; ++i)
	{
        zmq::message_t request;

        //  Wait for next request from client
        socket.recv (&request);
		Sleep(100);
        std::cout << "THREAD: " << (char*)request.data() << std::endl;
	}
}

int main(int argc, char* argv[])
{
	if (argc > 1)
	{
		if (!memcmp(argv[1], "ipc", 3))
		{
			std::cout << "SUB PROCESS IPC: started" << endl;
			zmq::context_t context(1);//((zmq::context_t*)params);
			zmq::socket_t socket (context, ZMQ_SUB);
			socket.connect ("ipc:///test/blabla3");
			socket.setsockopt(ZMQ_SUBSCRIBE, nullptr, 0);
			stringstream ss;
			for (int i = 0; i < 100; ++i)
			{
				zmq::message_t request;

				//  Wait for next request from client
				socket.recv (&request);
				Sleep(200);
				std::cout << "SUB PROCESS IPC: " << (char*)request.data() << std::endl;
			}
		}
		else
		{
			std::cout << "SUB PROCESS TCP: started" << endl;
			zmq::context_t context(1);//((zmq::context_t*)params);
			zmq::socket_t socket (context, ZMQ_SUB);
			socket.connect ("tcp://localhost:5556");
			socket.setsockopt(ZMQ_SUBSCRIBE, nullptr, 0);
			stringstream ss;
			for (int i = 0; i < 100; ++i)
			{
				zmq::message_t request;

				//  Wait for next request from client
				socket.recv (&request);
				std::cout << "SUB PROCESS TCP: " << (char*)request.data() << std::endl;
			}
		}
	}
	else
	{
		int major, minor, patch;
		zmq::version(&major, &minor, &patch);
		cout <<  major << minor << patch << endl;
	
		//  Prepare our context and socket
		zmq::context_t context (1);
		zmq::socket_t socket (context, ZMQ_PUB);
		socket.bind ("tcp://*:5556");
		socket.bind ("inproc://blabla2");
		socket.bind ("ipc:///test/blabla3");

		boost::thread sec_thr(thread, &context);

		std::string command(argv[0]);
		std::vector<std::string> args;
		Poco::ProcessHandle Phandle = Poco::Process::launch(command + string(" param"), args); 
		Poco::ProcessHandle Phandle2 = Poco::Process::launch(command + string(" ipc"), args); 
		stringstream ss;
		for (int i = 0; i < 100; ++i)
		{
			Sleep(1000);
			zmq::message_t request (80);
			ss << "This is meesage " << i << endl;
			memset(request.data(), 0, 80);
			memcpy(request.data(), ss.str().c_str(), ss.str().size());
			socket.send (request);
			ss.str(std::string());
			ss.clear();
		}
	}
    return 0;
}