#define ZMQ_BUILD_DRAFT_API
#include "zmq.h"

#include <assert.h>
#include <stdlib.h>
#include <thread>

void *ctx = NULL;
void *sock = NULL;

// поток обработки сообщений
void worker(int num) {
	printf("worker %d start\n", num);
	void* queue = zmq_socket(ctx, ZMQ_PULL);	assert(queue);
	int rc = zmq_connect(queue, "inproc://echo-queue"); assert(rc == 0);
	while(1) {
		// получение указателя из очереди
		zmq_msg_t p;
		rc = zmq_msg_init(&p); assert(rc == 0);
		rc = zmq_msg_recv(&p, queue, 0); assert(rc == sizeof(zmq_msg_t*));
		zmq_msg_t* msg = NULL;
		memcpy(&msg, zmq_msg_data(&p), sizeof(zmq_msg_t*));
		rc = zmq_msg_close(&p); assert(rc == 0);
		// обработка сообщения
		//printf("worker %d recv %d bytes from %X\n", num, zmq_msg_size(msg), zmq_msg_routing_id(msg));
		rc = zmq_msg_send(msg, sock, 0); assert(rc != -1); // отправка msg и освобождение памяти
		free(msg);
	}
	zmq_close(queue);
}

int main (void) {
	ctx = zmq_ctx_new (); assert (ctx);
	// сокет для сообщений от клиентов
	sock = zmq_socket(ctx, ZMQ_SERVER);	assert(sock);
	int rc = zmq_bind(sock, "tcp://*:12345"); assert(rc == 0); 
	// сокет для очереди обработки
	void* queue = zmq_socket(ctx, ZMQ_PUSH);	assert(queue);
	rc = zmq_bind(queue, "inproc://echo-queue"); assert(rc == 0);
	// Запуск потоков обработки
	std::thread w1(worker, 1), w2(worker, 2), w3(worker, 3);
	// прием сообщений от клиентов
	while (1) {
		zmq_msg_t* msg = (zmq_msg_t*)malloc(sizeof(zmq_msg_t)); assert(msg); // инициализация пустого сообщения
		rc = zmq_msg_init(msg); assert(rc == 0);// инициализация пустого сообщения
		rc = zmq_msg_recv(msg, sock, 0); assert(rc > 0); // ожидание приема, прием с выделением памяти
		// Постановка в очередь на обработку
		zmq_msg_t p;
		rc = zmq_msg_init_size(&p, sizeof(zmq_msg_t*)); assert(rc == 0);
		memcpy(zmq_msg_data(&p), &msg, sizeof(zmq_msg_t*));
		rc = zmq_msg_send(&p, queue, 0); assert(rc == sizeof(zmq_msg_t*));
	}
	zmq_close(sock);
	zmq_ctx_destroy(ctx);
	return 0;
}

#define COUNT 1000 // предел 1023, для 3000 запускал три client.exe
int main (void)
{
	printf("Connecting to ECHO server\n");
	void *ctx = zmq_ctx_new(); assert(ctx);

	void *sock[COUNT];
	int rc;
	for(int i = 0; i < COUNT; i++) {
		sock[i] = zmq_socket(ctx, ZMQ_CLIENT); assert(sock[i]);
		rc = zmq_connect(sock[i], "tcp://127.0.0.1:12345"); assert(rc == 0);
	}


	int t = GetTickCount();

	for (int i = 0; i < 100; i++) {
		for(int j = 0; j < COUNT; j++) {
			zmq_msg_t msg;
			rc = zmq_msg_init_size(&msg, 5); assert(rc == 0);
			memcpy(zmq_msg_data(&msg), "Hello", 5);
			rc = zmq_msg_send(&msg, sock[j], 0); assert(rc == 5);
		}
		for (int j = 0; j < COUNT; j++) {
			zmq_msg_t msg;
			rc = zmq_msg_init(&msg); assert(rc == 0);
			rc = zmq_msg_recv(&msg, sock[j], 0); assert(rc == 5);
		}
	}
	printf("Time: %d\n", GetTickCount() - t);
	zmq_close(sock);
	zmq_ctx_destroy(ctx);
	return 0;
}

#define ZMQ_BUILD_DRAFT_API
#include "zmq.h"

#include <assert.h>
#include <stdlib.h>
#include <thread>

void *ctx = NULL;
void *sock = NULL;

// рабочий поток обработки сообщений
void worker(int num) {
	printf("worker %d start\n", num);
	void* queue = zmq_socket(ctx, ZMQ_PULL);	assert(queue);
	int rc = zmq_connect(queue, "inproc://echo-queue"); assert(rc == 0);
	while(1) {
		// получение указателя из очереди
		zmq_msg_t p;
		rc = zmq_msg_init(&p); assert(rc == 0);
		rc = zmq_msg_recv(&p, queue, 0); assert(rc == sizeof(zmq_msg_t));
		zmq_msg_t* msg = (zmq_msg_t*)zmq_msg_data(&p);
		// обработка сообщения
		//printf("worker %d recv %d bytes from %X\n", num, zmq_msg_size(msg), zmq_msg_routing_id(msg));
		rc = zmq_msg_send(msg, sock, 0); assert(rc != -1); // отправка msg и освобождение памяти
		rc = zmq_msg_close(&p); assert(rc == 0); // освобождение памяти сообщения с указателем
	}
	zmq_close(queue);
}

int main (void) {
	ctx = zmq_ctx_new (); assert (ctx);
	// сокет для сообщений от клиентов
	sock = zmq_socket(ctx, ZMQ_SERVER);	assert(sock);
	int rc = zmq_bind(sock, "tcp://*:12345"); assert(rc == 0); 
	// сокет для очереди обработки
	void* queue = zmq_socket(ctx, ZMQ_PUSH);	assert(queue);
	rc = zmq_bind(queue, "inproc://echo-queue"); assert(rc == 0);
	// Запуск потоков обработки
	std::thread w1(worker, 1), w2(worker, 2), w3(worker, 3);
	// прием сообщений от клиентов
	while (1) {
		zmq_msg_t msg;
		rc = zmq_msg_init(&msg); assert(rc == 0);// инициализация пустого сообщения
		rc = zmq_msg_recv(&msg, sock, 0); assert(rc > 0); // ожидание приема, прием с выделением памяти
		// Постановка в очередь на обработку
		zmq_msg_t p;
		rc = zmq_msg_init_size(&p, sizeof(zmq_msg_t)); assert(rc == 0);
		memcpy(zmq_msg_data(&p), &msg, sizeof(zmq_msg_t));
		rc = zmq_msg_send(&p, queue, 0); assert(rc == sizeof(zmq_msg_t));
	}
	zmq_close(sock);
	zmq_ctx_destroy(ctx);
	return 0;
}

#define ZMQ_BUILD_DRAFT_API
#include "zmq.h"

#include <assert.h>
#include <stdlib.h>
#include <thread>

void *ctx = NULL;
void *sock = NULL;

// поток обработки сообщений
void worker(int num) {
	printf("worker %d start\n", num);
	void* queue = zmq_socket(ctx, ZMQ_PULL); assert(queue);
	int rc = zmq_connect(queue, "inproc://echo-queue"); assert(rc == 0);

	while (1) {
		// получение сообщения из очереди
		zmq_msg_t msg;
		rc = zmq_msg_init(&msg); assert(rc == 0);
		rc = zmq_msg_recv(&msg, queue, 0); assert(rc > 0);
		// обработка сообщения
		printf("worker %d recv %d bytes at %X from %X\n", num, zmq_msg_size(&msg), zmq_msg_data(&msg), zmq_msg_routing_id(&msg));
		rc = zmq_msg_send(&msg, sock, 0); assert(rc != -1); // отправка msg и освобождение памяти
	}
	zmq_close(queue);
}

int main (void) {
	ctx = zmq_ctx_new (); assert (ctx);
	// сокет для сообщений от клиентов
	sock = zmq_socket(ctx, ZMQ_SERVER); assert(sock);
	int rc = zmq_bind(sock, "tcp://*:12345"); assert(rc == 0); 
	// сокет для очереди обработки
	void* queue = zmq_socket(ctx, ZMQ_PUSH);	assert(queue);
	rc = zmq_bind(queue, "inproc://echo-queue"); assert(rc == 0);
	// запуск потоков обработки
	std::thread w1(worker, 1), w2(worker, 2), w3(worker, 3);
	// прием сообщений от клиентов
	while (1) {
		// Ожидание сообщений от клиентов
		zmq_msg_t msg;
		rc = zmq_msg_init(&msg); assert(rc == 0);// инициализация пустого сообщения
		rc = zmq_msg_recv(&msg, sock, 0); assert(rc > 0); // ожидание приема, прием с выделением памяти
		printf("recv %d bytes at %X from %X\n", zmq_msg_size(&msg), zmq_msg_data(&msg), zmq_msg_routing_id(&msg));
		// Постановка в очередь на обработку
		rc = zmq_msg_send(&msg, queue, 0); assert(rc > 0);
	}
	zmq_close(sock);
	zmq_ctx_destroy(ctx);
	return 0;
}