program LoadBalancingBroker;

{$APPTYPE CONSOLE}
// Брокер с балансировкой нагрузки.
//=================================
// Для упрощения запуска пример реализована как многопоточное приложение,
// с использованием протокола inproc
//=================================
// Имеется клиенты (c_NBR_CLIENTS шт) и рабочие (c_NBR_WORKERS шт).
// Каждый рабочий при запуске сообщает брокеру о своей готовности.
// Рабочий ставится в очередь.
//
// Клиент при запуске обращается к брокеру с заданием. Брокер отправляет
// задание первому свободному рабочему. Рабочий, выполнив задание, возвращает
// результат брокеру, брокер пересылает задание исходному клиенту

uses
//  Fastmm4,
  Windows,
  SysUtils,
  zmq_h,
  ZMQ_Utils;

const
  c_NBR_CLIENTS = 5; // Число клиентов
  c_NBR_WORKERS = 3; // // Число рабочих
  c_NMBR_REQ = 2; // // Число запросов от каждого клиента

  // Конечные точки подключения
  c_url_clients = 'inproc://clients';
  c_url_workers = 'inproc://workers';

procedure client_thread_proc(aContext: Pointer);
// Процедура потока клиента
var
  fSocketClient: Pointer;
  fReply: string;
  i: integer;
begin
  fSocketClient := zmq_socket(aContext, ZMQ_REQ);
  // Назначение идентификатора соединения (строка случайных символов)
  s_set_id(fSocketClient);
  zmq_connect(fSocketClient, c_url_clients); // Коннект к брокеру

  for i := 0 to Pred(c_NMBR_REQ) do begin
  // Отправка запроса, получение ответа
    s_send(fSocketClient, 'HELLO');
    fReply := s_recv(fSocketClient);
    z_Log(Format('Client : %s', [fReply]));
  end;
  zmq_close(fSocketClient);
end;


procedure worker_thread_proc(aContext: Pointer);
// Процедура рабочего потока
var
  fSocketWorker: Pointer;
  fIdentity: string;
  fEmpty: string;
  fRequest: string;

begin
  fSocketWorker := zmq_socket(aContext, ZMQ_REQ);
   // Назначение идентификатора соединения
  s_set_id(fSocketWorker);
  zmq_connect(fSocketWorker, c_url_workers); // Коннект к брокеру

  // Соощаем брокеру, что рабочий поток запущен и готов к работе
  s_send(fSocketWorker, 'READY');

  while true do // Рабочий цикл
  begin
    // Читаем и запоминаем все кадры вплоть пустого (fEmpty)
    // В данном примере кадров всего один, но реально их может быть больше
    fIdentity := s_recv(fSocketWorker); // Идентификатор клиента
    if zmq_errno() = ETERM then
      Break; // Уходим, если контекст в процессе завершения

    fEmpty := s_recv(fSocketWorker); // Кадр - разделитель
    Assert(fEmpty = '');
    // Получение запроса, отправка ответа
    fRequest := s_recv(fSocketWorker);
    z_Log(Format('Worker : %s', [fRequest]));

    Sleep(5); // Имитируем выполнение полезной работы

    // Формирование конверта составного сообщения:
    s_send(fSocketWorker, fIdentity, ZMQ_SNDMORE);
      // Идентификатор клиента
    s_send(fSocketWorker, '', ZMQ_SNDMORE); // Разделитель
    s_send(fSocketWorker, 'OK'); // Результат работы
  end;
  zmq_close(fSocketWorker);
end;

procedure thread_proc(aContext: Pointer); cdecl;
var
  fVal: Integer;
begin
  fVal := Integer(aContext^);
  z_Log(IntToStr(fVal));
end;


var
  fContext: Pointer;
  fSocketClients: Pointer;
  fSocketWorkers: Pointer;
  i: Integer;
  fThrId: Cardinal;
  fAvailableWorkers: Integer;
  fZMQPoll: array[0..1] of zmq_pollitem_t;
  fWrkrs_Que: array[0..Pred(c_NBR_WORKERS)] of string; // Очередь рабочих
  fRC: Integer;
  fWorker_id: string;
  fClient_id: string;
  fEmpty: string;
  fReplay: string;
  fRequest: string;
  fCliReqNmbr: Integer; // Номер клиентского запроса
  fThrWIds: array[0..Pred(c_NBR_WORKERS)] of Cardinal;
  fThrCIds: array[0..Pred(c_NBR_CLIENTS)] of Cardinal;
  fP: Pointer;
begin
 // Подготовка контекста и сокетов
  fContext := zmq_ctx_new();
  fSocketClients := zmq_socket(fContext, ZMQ_ROUTER);
  fSocketWorkers := zmq_socket(fContext, ZMQ_ROUTER);
  zmq_bind(fSocketClients, c_url_clients);
  zmq_bind(fSocketWorkers, c_url_workers);

  for i := 0 to Pred(c_NBR_WORKERS) do // Создаем рабочих
    fThrWIds[i] := BeginThread(nil, 0, @worker_thread_proc, fContext, 0, fThrId);

  for i := 0 to Pred(c_NBR_CLIENTS) do // Создаем клиентов
    fThrCIds[i] := BeginThread(nil, 0, @client_thread_proc, fContext, 0, fThrId);


// Главный цикл для LRU очереди. Используется два сокета: fSocketClients для
// клиентов и fSocketWorkers для рабочих. Опрос fSocketWorkers
// выполняется всегда, а fSocketClients - только тогда, когда есть один или
// больше готовых рабочих.
// Сообщения, которые еще не готовы к обработке, в ZMQ хранятся
// во встроенной очередей сообщений.
// Когда мы получаем запрос клиента, мы берем рабочего из начала
// очереди (fWrkrs_Que[0]) и посылаем ему запрос, которых включаеи исходный
// идентификатор клиента.
// Когда же приходит запрос от рабочего, этого рабочего ставим в конец очереди,
// а ответ переправляем исходному клиенту (используя идинтификатор в конверте).

  fAvailableWorkers := 0; // Число доступных рабочих

  fCliReqNmbr := 0;

  // Подготовка пула сокетов
  fZMQPoll[0].socket := fSocketWorkers;
  fZMQPoll[0].fd := 0;
  fZMQPoll[0].events := ZMQ_POLLIN;
  fZMQPoll[1].socket := fSocketClients;
  fZMQPoll[1].fd := 0;
  fZMQPoll[1].events := ZMQ_POLLIN;

  while fCliReqNmbr < c_NBR_CLIENTS * c_NMBR_REQ do begin

    fZMQPoll[0].revents := 0; // Сброс результатов опроса
    fZMQPoll[1].revents := 0;


    // Читать из fSocketClients только тогда, когда есть свободные рабочие
    // Если нет свободных - читать только из fSocketWorkers
    if fAvailableWorkers = 0 then
      fRC := zmq_poll(@fZMQPoll, 1, 11)
        // Только ждем готовности от рабочих
    else
      fRC := zmq_poll(@fZMQPoll, 2, 11); // Читаем оба сокета

    if fRC = -1 then
      Break; // Цикл прерван

    // Обработка действий рабочих на fSocketWorkers
    if (fZMQPoll[0].revents and ZMQ_POLLIN) <> 0 then begin
      fWorker_id := s_recv(fSocketWorkers);
      Assert(fAvailableWorkers < c_NBR_WORKERS);
      // Помещаем рабочего в конец очереди
      fWrkrs_Que[fAvailableWorkers] := fWorker_id;
      Inc(fAvailableWorkers);

      // Второй кадр - пустой
      fEmpty := s_recv(fSocketWorkers);
      assert(fEmpty = '');

      // Третий кадр - готовность ("READY"), иначе это Id клиента в ответе
      fClient_id := s_recv(fSocketWorkers);

      // Если это ответ клиенту, отправить ответ в fSocketClients
      if fClient_id <> 'READY' then begin
        fEmpty := s_recv(fSocketWorkers);
        Assert(fEmpty = '');
        fReplay := s_recv(fSocketWorkers);
        s_send(fSocketClients, fClient_id, ZMQ_SNDMORE);
        s_send(fSocketClients, '', ZMQ_SNDMORE);
        s_send(fSocketClients, fReplay);
        Inc(fCliReqNmbr); // Число обслуженных запросов

      end;
    end;

    // Обработка запросов клиентов:
    if (fZMQPoll[1].revents and ZMQ_POLLIN) <> 0 then
    begin
      // Получаем очередной клиентский запрос, отправляем его рабочему из
      // начала очереди
      // Конверт запроса клиента: [identity][empty][request]
      fClient_id := s_recv(fSocketClients);
      fEmpty := s_recv(fSocketClients);
      Assert(fEmpty = '');
      fRequest := s_recv(fSocketClients);

      s_send(fSocketWorkers, fWrkrs_Que[0], ZMQ_SNDMORE);
      s_send(fSocketWorkers, '', ZMQ_SNDMORE);
      s_send(fSocketWorkers, fClient_id, ZMQ_SNDMORE);
      s_send(fSocketWorkers, '', ZMQ_SNDMORE);
      s_send(fSocketWorkers, fRequest);

      // Извлечение из очереди
      Dec(fAvailableWorkers);
      for i := 0 to Pred(fAvailableWorkers) do
        fWrkrs_Que[i] := fWrkrs_Que[i + 1];
    end;
  end;

  for I := 0 to High(fThrCIds) do // Ждем завершения клиентов
    WaitForSingleObject(fThrCIds[i], INFINITE);
  zmq_threadstart(thread_proc, @i);
  Sleep(1000);
  zmq_close(fSocketClients);
  zmq_close(fSocketWorkers);
  zmq_ctx_destroy(fContext);

//  for I := 0 to High(fThrWIds) do
//    WaitForSingleObject(fThrWIds[i], INFINITE);

  Readln;

end.

  while fCliReqNmbr < c_NBR_CLIENTS * c_NMBR_REQ do begin

    fZMQPoll[0].revents := 0; // Сброс результатов опроса
    fZMQPoll[1].revents := 0;


    // Читать из fSocketClients только тогда, когда есть свободные рабочие
    // Если нет свободных - читать только из fSocketWorkers
    if fAvailableWorkers = 0 then
      fRC := zmq_poll(@fZMQPoll, 1, 11)
        // Только ждем готовности от рабочих
    else
      fRC := zmq_poll(@fZMQPoll, 2, 11); // Читаем оба сокета

    if fRC = -1 then
      Break; // Цикл прерван

    // Обрабока действий рабочих на fSocketWorkers
    if (fZMQPoll[0].revents and ZMQ_POLLIN) <> 0 then begin
      fWorker_id := s_recv(fSocketWorkers);
      Assert(fAvailableWorkers < c_NBR_WORKERS);
      // Помещаем рабочего в конец очереди
      fWrkrs_Que[fAvailableWorkers] := fWorker_id;
      Inc(fAvailableWorkers);

      // Второй кадр - пустой
      fEmpty := s_recv(fSocketWorkers);
      assert(fEmpty = '');

      // Третий кадр - готовность ("READY"), иначе это Id клиента в ответе
      fClient_id := s_recv(fSocketWorkers);

      // Если это ответ клиенту, отправить ответ в fSocketClients
      if fClient_id <> 'READY' then begin
        fEmpty := s_recv(fSocketWorkers);
        Assert(fEmpty = '');
        fReplay := s_recv(fSocketWorkers);
        s_send(fSocketClients, fClient_id, ZMQ_SNDMORE);
        s_send(fSocketClients, '', ZMQ_SNDMORE);
        s_send(fSocketClients, fReplay);
        Inc(fCliReqNmbr); // Число обслуженных запросов

      end;
    end;

    // Обработка запросов клиентов:
    if (fZMQPoll[1].revents and ZMQ_POLLIN) <> 0 then
    begin
      // Получаем очередной клиентский запрос, отправляем его рабочему из
      // начала очереди
      // Конверт запроса клиента: [identity][empty][request]
      fClient_id := s_recv(fSocketClients);
      fEmpty := s_recv(fSocketClients);
      Assert(fEmpty = '');
      fRequest := s_recv(fSocketClients);

      s_send(fSocketWorkers, fWrkrs_Que[0], ZMQ_SNDMORE);
      s_send(fSocketWorkers, '', ZMQ_SNDMORE);
      s_send(fSocketWorkers, fClient_id, ZMQ_SNDMORE);
      s_send(fSocketWorkers, '', ZMQ_SNDMORE);
      s_send(fSocketWorkers, fRequest);

      // Извлечение из очереди
      Dec(fAvailableWorkers);
      for i := 0 to Pred(fAvailableWorkers) do
        fWrkrs_Que[i] := fWrkrs_Que[i + 1];
    end;
  end;

program hl_LoadBalancingBroker;

{$APPTYPE CONSOLE}
// Брокер с балансировкой нагрузки.
// Демонстрируется использование CZMQ и реактора
//==============================================
// Имеется клиенты (c_NBR_CLIENTS шт) и рабочие (c_NBR_WORKERS шт).
// Каждый рабочий при запуске сообщает брокеру о своей готовности (c_WORKER_READY).
// Рабочий ставится в очередь.
//
// Клиент при запуске обращается к брокеру с заданием. Брокер отправляет
// задание первому свободному рабочему. Рабочий, выполнив задание, возвращает
// результат брокеру, брокер пересылает результат клиенту - заказчику.

uses
  SysUtils
  , zmq_h
  , czmq_h
  , ZMQ_Utils
  ;

const
  c_NBR_CLIENTS = 10; // Число клиентов
  c_NBR_WORKERS = 3; // // Число рабочих

  // Конечные точки подключения
  c_url_clients = 'tcp://%s:5555'; //'inproc://clients';
  c_url_workers = 'tcp://%s:5556'; //'inproc://workers';
  c_domain = 'localhost'; // Сетевой адрес брокера
  c_interf: string = '*'; // Адрес для биндинга

  c_WORKER_READY: byte = 1; // Сигнал готовности рабочего


function client_task(args: Pointer): Pointer; cdecl;
// Функция треда клиента
var
  client: Pointer;
  ctx: p_zctx_t;
  reply: PChar;
begin
  ctx := args; // zctx_new();
  client := zsocket_new(ctx, ZMQ_REQ);
  zsocket_connect(client, c_url_clients, c_domain);
// Запрос - ответ
  while true do begin
    zstr_send(client, 'HELLO');
    reply := zstr_recv(client);
    if (reply = nil) then
      break;
    z_Log('Client: ' + reply);
    zstr_free(reply);
    sleep(1);
  end;
//  zctx_destroy(ctx);
  Result := nil;
end;

function worker_task(args: Pointer): Pointer; cdecl;
// Функция треда рабочего
var
  ctx: p_zctx_t;
  frame: p_zframe_t;
  msg: p_zmsg_t;
  worker: Pointer;
begin
  ctx := args; // zctx_new();
  worker := zsocket_new(ctx, ZMQ_REQ);
  zsocket_connect(worker, c_url_workers, c_domain);

// Сообщаем брокеру о готовности работать
  frame := zframe_new(@c_WORKER_READY, 1);
  zframe_send(frame, worker, 0);

// ОБработка сообщений по мере их получения
  while True do begin
    msg := zmsg_recv(worker);
    if msg = nil then
      break; // Interrupted
    zframe_reset(zmsg_last(msg), PChar('OK'), 2);
    zmsg_send(msg, worker);
  end;
//  zctx_destroy(ctx);
  Result := nil;
end;


// Структура, передаваемая в реактор
type
  p_lbbroker_t = ^lbbroker_t;
  lbbroker_t = packed record
    frontend: Pointer; // Сокет - слушать клиентов
    backend: Pointer; // Сокет - слушать рабочих
    workers: p_zlist_t; // Список свободных рабочих
  end;

// Устройство реактора таково, что все сообщения, приходящие в сокет,
// передаются ректором в функцию обработки. У нас - два обработчика:
// для фронтэнда (клиенты) и для бэкэнда (рабочие)

// Обработка ввода от клиентjd (на фронтэнд)

function s_handle_frontend(loop: p_zloop_t; poller: p_zmq_pollitem_t; arg: Pointer):
  Integer; cdecl;
var
  msg: p_zmsg_t;
  self: p_lbbroker_t;
begin
  self := p_lbbroker_t(arg);
  msg := zmsg_recv(self.frontend); // Сообщение от клиента
  if msg <> nil then begin
    zmsg_wrap(msg, p_zframe_t(zlist_pop(self.workers)));
    zmsg_send(msg, self.backend);
// Завершение обработчика, если нет доступных рабочих
    if zlist_size(self.workers) = 0 then begin
      poller.socket := self.frontend;
      poller.fd := 0;
      poller.events := ZMQ_POLLIN;
      poller.revents := 0;
      zloop_poller_end(loop, poller);
    end;
  end;
  Result := 0;
end;

// Обработка ввода от рабочих (на бэкэнд)

function s_handle_backend(loop: p_zloop_t; poller: p_zmq_pollitem_t; arg: Pointer):
  Integer; cdecl;
var
  frame: p_zframe_t;
  identity: p_zframe_t;
  msg: p_zmsg_t;
  self: p_lbbroker_t;
begin
// Для балансировки нагрузки снова используем идентификацию
  self := p_lbbroker_t(arg);
  msg := zmsg_recv(self.backend);
  if msg <> nil then begin
    identity := zmsg_unwrap(msg);
    zlist_append(self.workers, identity);


    if zlist_size(self.workers) = 1 then begin
    // Разрешить ридер фронтэнда
      poller.socket := self.frontend;
      poller.fd := 0;
      poller.events := ZMQ_POLLIN;
      poller.revents := 0;
      zloop_poller(loop, poller, @s_handle_frontend, self);
    end;
// Переброска сообщения клиенту, если это не "ГОТОВ".
    frame := zmsg_first(msg);
    if CompareMem(zframe_data(frame), @c_WORKER_READY, 1) then
      zmsg_destroy(msg)
    else
      zmsg_send(msg, self.frontend);
  end;
  result := 0;
end;

// Основной тред, порождающий дочерние, запускающий затем реактор.
// Если нажать Ctrl-C, реактор завершится и главный тред тоже завершится.

procedure DoMain;
var
  ctx: p_zctx_t;
  frame: p_zframe_t;
  i: Integer;
  poller: zmq_pollitem_t;
  reactor: p_zloop_t;
  self: p_lbbroker_t;
begin
  ctx := zctx_new(); // Контекст
  New(self); // Данные реактора

  self.frontend := zsocket_new(ctx, ZMQ_ROUTER); // Сокеты реактора
  self.backend := zsocket_new(ctx, ZMQ_ROUTER);

  zsocket_bind(self.frontend, c_url_clients, c_interf);
    // Привязка к интерфейсу
  zsocket_bind(self.backend, c_url_workers, c_interf);

  for i := 0 to pred(c_NBR_CLIENTS) do // Запуск тредов клиентов
    zthread_new(@client_task, ctx);
  for i := 0 to Pred(c_NBR_WORKERS) do // Запуск тредов рабочих
    zthread_new(@worker_task, ctx);

// Очередь доступных рабочих
  self.workers := zlist_new();

// Подготовка и запуск реактора
  reactor := zloop_new();

  poller.socket := self.backend;
  poller.fd := 0;
  poller.events := ZMQ_POLLIN;
  poller.revents := 0;

  zloop_poller(reactor, @poller, @s_handle_backend, self);
  zloop_start(reactor);
  zloop_destroy(reactor);

// Аккуратно завершаем все при выходе
  while zlist_size(self.workers) > 0 do begin
    frame := zlist_pop(self.workers);
    zframe_destroy(frame);
  end;
  zlist_destroy(self.workers);
  zctx_destroy(ctx);
  Dispose(self);
end;

begin
  DoMain;
  Readln;
end.

program hl_LoadBalancingBrokerSimple;

{$APPTYPE CONSOLE}
// Брокер с балансировкой нагрузки.
// Демонстрируется использование CZMQ
//======================================================
// Имеется клиенты (c_NBR_CLIENTS шт) и рабочие (c_NBR_WORKERS шт).
// Каждый рабочий при запуске сообщает брокеру о своей готовности (c_WORKER_READY).
// Рабочий ставится в очередь.
//
// Клиент при запуске обращается к брокеру с заданием. Брокер отправляет
// задание первому свободному рабочему. Рабочий, выполнив задание, возвращает
// результат брокеру, брокер пересылает результат клиенту - заказчику.


uses
  SysUtils,
  zmq_h,
  czmq_h,
  ZMQ_Utils;

const
  c_NBR_CLIENTS = 1; // Число клиентов
  c_NBR_WORKERS = 1; // // Число рабочих

  // Конечные точки подключения
  c_url_clients = 'tcp://%s:5555'; //'inproc://clients';
  c_url_workers = 'tcp://%s:5556'; //'inproc://workers';
  c_domain = 'localhost'; // Сетевой адрес брокера
  c_interf: string = '*'; // Адрес для биндинга

  c_WORKER_READY: byte = 1;
    // Сигнал готовности рабочего         ё


function client_task(args: Pointer): Pointer; cdecl;
// Функция треда клиента
var
  client: Pointer;
  ctx: p_zctx_t;
  reply: PChar;
begin
  ctx := args; // zctx_new();
  client := zsocket_new(ctx, ZMQ_REQ);
  zsocket_connect(client, c_url_clients, c_domain);
// Запрос - ответ
  while true do begin
    zstr_send(client, 'HELLO');
    reply := zstr_recv(client);
    if (reply = nil) then
      break;
    z_Log('Client: ' + reply);
    zstr_free(reply);
    sleep(1);
  end;
//  zctx_destroy(ctx);
  Result := nil;
end;

function worker_task(args: Pointer): Pointer; cdecl;
// Функция треда рабочего
var
  ctx: p_zctx_t;
  frame: p_zframe_t;
  msg: p_zmsg_t;
  worker: Pointer;
begin
  ctx := args; // zctx_new();
  worker := zsocket_new(ctx, ZMQ_REQ);
  zsocket_connect(worker, c_url_workers, c_domain);

// Сообщаем брокеру о готовности работать
  frame := zframe_new(@c_WORKER_READY, 1);
  zframe_send(frame, worker, 0);

// ОБработка сообщений по мере их получения
  while True do begin
    msg := zmsg_recv(worker);
    if msg = nil then
      break; // Interrupted
    zframe_reset(zmsg_last(msg), PChar('OK'), 2);
    zmsg_send(msg, worker);
  end;
//  zctx_destroy(ctx);
  Result := nil;
end;




// Основной тред, порождающий дочерние, запускающий затем реактор.
// Если нажать Ctrl-C, реактор завершится и главный тред тоже завершится.

procedure DoMain;
var
  backend: Pointer;
  ctx: p_zctx_t;
  frame: p_zframe_t;
  frontend: Pointer;
  i: Integer;
  identity: p_zframe_t;
  poller: zmq_pollitem_t;
  reactor: p_zloop_t;
  items: array[0..1] of zmq_pollitem_t;
  msg: p_zmsg_t;
  rc: Integer;
  workers: p_zlist_t;
begin
  ctx := zctx_new(); // Контекст

  frontend := zsocket_new(ctx, ZMQ_ROUTER); // Сокеты реактора
  backend := zsocket_new(ctx, ZMQ_ROUTER);

  zsocket_bind(frontend, c_url_clients, c_interf);
    // Привязка к интерфейсу
  zsocket_bind(backend, c_url_workers, c_interf);

  for i := 0 to pred(c_NBR_CLIENTS) do // Запуск тредов клиентов
    zthread_new(@client_task, ctx);
  for i := 0 to Pred(c_NBR_WORKERS) do // Запуск тредов рабочих
    zthread_new(@worker_task, ctx);

// Очередь доступных рабочих
  workers := zlist_new();

 // Главный цикл балансировщика нагрузок.
 // Длинне, чем с реактором, но короче, чем на "чистом" ZMQ
  while True do begin
    items[0].socket := backend;
    items[0].fd := 0;
    items[0].events := ZMQ_POLLIN;
    items[0].revents := 0;
    items[1].socket := frontend;
    items[1].fd := 0;
    items[1].events := ZMQ_POLLIN;
    items[1].revents := 0;

// Опрашиваем клиентов только если есть незанятые рабочие
    if zlist_size(workers) > 0 then
      rc := zmq_poll(@items[0], 2, -1)
    else
      rc := zmq_poll(@items[0], 1, -1);
    if rc = -1 then
      break; // прерывание

// Обработка данных рабочих (от backend)
    if (items[0].revents and ZMQ_POLLIN) <> 0 then begin
  // Используем идентификацию
      msg := zmsg_recv(backend);
      if msg = nil then
        break; // Interrupted
      identity := zmsg_unwrap(msg);
      zlist_append(workers, identity);

// Если это не сообщение о готовности, переслать сообщение клиенту
      frame := zmsg_first(msg);
      if CompareMem(zframe_data(frame), @c_WORKER_READY, 1) then
        zmsg_destroy(msg)
      else
        zmsg_send(msg, frontend);
    end;
    if (items[1].revents and ZMQ_POLLIN) <> 0 then begin
// Получение запроса от клиента, передача первому незанятому рабочему
      msg := zmsg_recv(frontend);
      if msg <> nil then begin
        zmsg_wrap(msg, zlist_pop(workers));
        zmsg_send(msg, backend);
      end;
    end;
  end;
// Аккуратно завершаем все при выходе
  while zlist_size(workers) > 0 do begin
    frame := zlist_pop(workers);
    zframe_destroy(frame);
  end;
  zlist_destroy(workers);
  zctx_destroy(ctx);
end;

begin
  DoMain;
  Readln;
end.

program hl_AsynchClientServer;

{$APPTYPE CONSOLE}

uses
  SysUtils
  , zmq_h
  , czmq_h
  ;


// Асинхронные клиент - сервер ( от DEALER к ROUTER)
//
// Так как это пример, для облегчения запуска все работает в рамках одного процесса.
// В реальности каждая задача должна быть отдельным процессом.


// Задача клиента
// Клиент коннектится к серверу и шлет ему запросы по одному в секунду.
// Собирает ответы в порядке поступления а потом распечатывает.
// Несколько клиентов работаеют параллельно, каждый со своим Id.

function client_task(args: Pointer): Pointer; cdecl;
var
  centitick: Integer;
  client: Pointer;
  ctx: p_zctx_t;
  identity: string;
  items: zmq_pollitem_t;
  msg: p_zmsg_t;
  request_nbr: Integer;
begin
  ctx := zctx_new();
  client := zsocket_new(ctx, ZMQ_DEALER);

// Случайный идентификатор (текст: для облегчения трассировки)
  identity := Format('%4x - %4x', [Random($10000), Random($10000)]);
  zsocket_set_identity(client, PChar(identity));
  zsocket_connect(client, 'tcp://localhost:5570');

  items.socket := client;
  items.fd := 0;
  items.events := ZMQ_POLLIN;
  items.revents := 0;
  request_nbr := 0;
  while true do begin
// Тики по одному в секунду: получаем приходящие сообщения
    for centitick := 0 to 99 do begin
      zmq_poll(@items, 1, 10 * ZMQ_POLL_MSEC);
        // Опрос с таймаутом 0,01 сек
      if (items.revents and ZMQ_POLLIN) <> 0 then begin // Что-то есть
        msg := zmsg_recv(client);
        zframe_print(zmsg_last(msg), PChar(identity));
        zmsg_destroy(&msg);
      end;
    end;
    Inc(request_nbr);
    zstr_sendf(client, PChar('request # %d'), request_nbr);
  end;
  zctx_destroy(ctx);
  result := nil;
end;

// Задача сервера
// Используется многонитевая модель. Для вытягивания запросов в пул
// рабочих и распределения ответов обратно клиентам. Один рабочий может
// одновременно обработать один запрос, но один клиент может общаться с
// множеством рабочих одновременно.

procedure server_worker(args: Pointer; ctx: p_zctx_t; pipe: Pointer); cdecl; forward;

function server_task(args: Pointer): Pointer;
var
  backend: Pointer;
  ctx: p_zctx_t;
  frontend: Pointer;
  thread_nbr: Integer;
begin
// Фронтенд сокет общается с клиентами по tcp
  ctx := zctx_new();
  frontend := zsocket_new(ctx, ZMQ_ROUTER);
  zsocket_bind(frontend, 'tcp://*:5570');

// Бэкенд сокет общается с рабочими по inproc
  backend := zsocket_new(ctx, ZMQ_DEALER);
  zsocket_bind(backend, 'inproc://backend');

// Запуск пула рабочих нитей, точное количество не важно

  for thread_nbr := 0 to 4 do
    zthread_fork(ctx, @server_worker, nil);

// Коннект бэкэнда к фронтэнду через прокси
  zmq_proxy(frontend, backend, nil);

  zctx_destroy(ctx);
  result := nil;

end;

// Каждая задача рабочего работает одновременно над одним запросом и
// отправляет случайное число ответов со случайными паузами между ответами:

procedure server_worker(args: Pointer; ctx: p_zctx_t; pipe: Pointer); cdecl;
var
  content: p_zframe_t;
  identity: p_zframe_t;
  msg: p_zmsg_t;
  replies: Integer;
  reply: Integer;
  worker: Pointer;
begin
  worker := zsocket_new(ctx, ZMQ_DEALER);
  zsocket_connect(worker, 'inproc://backend');

  while true do begin
// Сокет DEALER дает нам конверт ответа и мообщение
    msg := zmsg_recv(worker);
    identity := zmsg_pop(msg);
    content := zmsg_pop(msg);
    assert(content <> nil);
    zmsg_destroy(msg);

// Отправка обратно 0..4 ответов
    replies := Random(5);
    for reply := 0 to pred(replies) do begin
// Sleep какое-то случайное время
      zclock_sleep(Random(1000) + 1);
      zframe_send(identity, worker, c_ZFRAME_REUSE + c_ZFRAME_MORE);
      zframe_send(&content, worker, c_ZFRAME_REUSE);
    end;
    zframe_destroy(&identity);
    zframe_destroy(&content);
  end;
end;

// Главный тред просто запускает 3 клиента и 1 сервер и ждет; затем сервер завершается.

procedure DoMain;
begin
  Randomize;
  zthread_new(@client_task, nil);
  zthread_new(@client_task, nil);
  zthread_new(@client_task, nil);
  zthread_new(@server_task, nil);
  zclock_sleep(5 * 1000);
    // Работаем 5 секунд, потом завершение
end;

begin
  DoMain;
  Readln; // Для отладки
end.