program MS_Reader;

{$APPTYPE CONSOLE}

uses
  SysUtils, ZMQ;

var

  fConext: Pointer;
  fSocketReceiver: Pointer;
  fSocketSubscriber: Pointer;
  fMsgBuff: array[0..255] of Char;
  fSize: Integer;
begin

// Подключение к задаче "ventilator"
  fConext := zmq_ctx_new();
  fSocketReceiver := zmq_socket(fConext, ZMQ_PULL);
  zmq_connect(fSocketReceiver, 'tcp://localhost:5557');

// Подклчение к серверу метеостанции
  fSocketSubscriber := zmq_socket(fConext, ZMQ_SUB);
  zmq_connect(fSocketSubscriber, 'tcp://localhost:5556');
  zmq_setsockopt(fSocketSubscriber, ZMQ_SUBSCRIBE, PChar('10001 '), 6);

// Процесс обработки для обоих сокетов
// Считаем, что трафик от  задачи "ventilator" для нас важнее
  while true do begin
    while true do begin
      fSize := zmq_recv(fSocketReceiver, fMsgBuff, SizeOf(fMsgBuff), ZMQ_DONTWAIT);
      if fSize >= 0 then
        // Выполнение задания
      else
        break;
    end;
    while true do begin
      fSize := zmq_recv(fSocketSubscriber, fMsgBuff, SizeOf(fMsgBuff), ZMQ_DONTWAIT);
      if fSize >= 0 then
    // Обработка данных о погоде
      else
        break;
    end;
    // Активности нет, спим 1 ms
    Sleep(1);
  end;
  zmq_close(fSocketReceiver);
  zmq_close(fSocketSubscriber);
  zmq_ctx_destroy(fConext);

  Readln;

end.

program MS_Reader_Poll;

{$APPTYPE CONSOLE}

uses
  SysUtils,
  ZMQ;

var

  fConext: Pointer;
  fSocketReceiver: Pointer;
  fSocketSubscriber: Pointer;
  fMsgBuff: array[0..255] of Char;
  fSize: Integer;
  fItems: array[0..1] of pollitem_t;
begin

// Подключение к задаче "ventilator"
  fConext := zmq_ctx_new();
  fSocketReceiver := zmq_socket(fConext, ZMQ_PULL);
  zmq_connect(fSocketReceiver, 'tcp://localhost:5557');

// Подклчение к серверу метеостанции
  fSocketSubscriber := zmq_socket(fConext, ZMQ_SUB);
  zmq_connect(fSocketSubscriber, 'tcp://localhost:5556');
  zmq_setsockopt(fSocketSubscriber, ZMQ_SUBSCRIBE, PChar('Temperature '), 12);

  fItems[0].socket := fSocketReceiver;
  fItems[0].fd := 0;
  fItems[0].events := ZMQ_POLLIN;
  fItems[0].revents := 0;

  fItems[1].socket := fSocketSubscriber;
  fItems[1].fd := 0;
  fItems[1].events := ZMQ_POLLIN;
  fItems[1].revents := 0;
// Процесс обработки для обоих сокетов
  while true do begin

    zmq_poll(fItems[0], Length(fItems), -1);
    if ((fItems[0].revents and ZMQ_POLLIN) <> 0) then begin
      fSize := zmq_recv(fSocketReceiver, fMsgBuff, SizeOf(fMsgBuff), 0);
      if fSize >= 0 then
        // Выполнение задания
      else
        break;
      if ((fItems[1].revents and ZMQ_POLLIN) <> 0) then begin
        fSize := zmq_recv(fSocketSubscriber, fMsgBuff, SizeOf(fMsgBuff), 0);
        if fSize >= 0 then
    // Обработка данных о погоде
        else
          break;
      end;
    end;
  end;

  zmq_close(fSocketReceiver);
  zmq_close(fSocketSubscriber);
  zmq_ctx_destroy(fConext);

  Readln;

end.

program BrRR_Client;

{$APPTYPE CONSOLE}
// Клиент
// Коннектится сокетом REQ к tcp://localhost:5559
// Шлет целое число сервису (серверу), обратно получает квадрат числа

uses
  SysUtils, ZMQ;
const
  c_ReqCnt = 100;
var
  fContext: Pointer;
  fSocketRequester: Pointer;
  fSrcVal: Cardinal;
  fResultVal: UInt64;
  i: integer;
begin

  fContext := zmq_ctx_new(); // Инициализация
  fSocketRequester := zmq_socket(fContext, ZMQ_REQ);
  zmq_connect(fSocketRequester, 'tcp://localhost:5559'); // Коннект к сервису
  Randomize();
  for i := 0 to Pred(c_ReqCnt) do begin
    fSrcVal := Cardinal(Random(-1)); // Генерация случайного целого
    zmq_send(fSocketRequester, fSrcVal, SizeOf(fSrcVal), 0); // Запрос
    zmq_recv(fSocketRequester, fResultVal, SizeOf(fResultVal), 0); // Ответ
    Writeln('Iter: ', i, ' src=', fSrcVal, ' result=', fResultVal);
  end;
  zmq_close(fSocketRequester);
  zmq_ctx_destroy(fContext);
  Readln;

end.

program BrRR_Service;

{$APPTYPE CONSOLE}
// Сервис (сервер)
// Находится в известной клиентам конечной точке, связывает сокет REP с tcp:*:5560,
// Получает целое число, возводит в квадрат и отправляет обратно результат

uses
  SysUtils, ZMQ;
var
  fContext: Pointer;
  fSocketResponder: Pointer;
  fSrcVal: Cardinal;
  fResultVal: UInt64;
  i: integer;
begin

  fContext := zmq_ctx_new(); // Инициализация
  fSocketResponder := zmq_socket(fContext, ZMQ_REP);
  zmq_bind(fSocketResponder, 'tcp://*:5559'); // Привязка к конечной точке
  Writeln('Starting service...');
  i := 0;
  while True do begin
    zmq_recv(fSocketResponder, fSrcVal, SizeOf(fSrcVal), 0); // Запрос
    fResultVal := UInt64(fSrcVal) * UInt64(fSrcVal); // "Полезная работа"
    zmq_send(fSocketResponder, fResultVal, SizeOf(fResultVal), 0); // Ответ
    Writeln('Iter: ', i, ' src=', fSrcVal, ' result=', fResultVal);
    Inc(i);
  end;
  zmq_close(fSocketResponder);
  zmq_ctx_destroy(fContext);
  Readln;

end.

program BrRR_Broker;

{$APPTYPE CONSOLE}
// Брокер
// Находится в известной клиентам и сервисам конечной точке,
// Находится в известной клиентам конечной точке,
// связывает сокет ZMQ_ROUTER с tcp:*:5559,
// связывает сокет ZMQ_DEALER с tcp:*:5560,
// Перекидывает сообщения между сокетами от FrontEnd к BackEnd
// и обратно

uses
  SysUtils, ZMQ, Math;
var
  fContext: Pointer;
  fSocketFrontEnd: Pointer;
  fSocketBackEnd: Pointer;
  fZMQPoll: array[0..1] of pollitem_t;
  fMsg: zmq_msg_t;
  fDoMore: Boolean;
begin

  fContext := zmq_ctx_new(); // Инициализация
  fSocketFrontEnd := zmq_socket(fContext, ZMQ_ROUTER);
  fSocketBackEnd := zmq_socket(fContext, ZMQ_DEALER);
  zmq_bind(fSocketFrontEnd, 'tcp://*:5559');
    // Конечная точка для клиентов
  zmq_bind(fSocketBackEnd, 'tcp://*:5560');
    // Кончная точка для сервисов
  fZMQPoll[0].socket := fSocketFrontEnd;
    // Инициализация пула сокетов
  fZMQPoll[0].fd := 0;
  fZMQPoll[0].events := ZMQ_POLLIN;
  fZMQPoll[0].revents := 0;
  fZMQPoll[1].socket := fSocketBackEnd;
  fZMQPoll[1].fd := 0;
  fZMQPoll[1].events := ZMQ_POLLIN;
  fZMQPoll[1].revents := 0;
  while true do begin
    zmq_poll(fZMQPoll[0], Length(fZMQPoll), -1);
      // Проверка состояния сокетов из пула
    if (fZMQPoll[0].revents and ZMQ_POLLIN) <> 0 then
      while True do
      begin // Трансляция сообщний от клиента к сервису
      // Обработка всх частей сообщения
        zmq_msg_init(fMsg);
        zmq_msg_recv(fMsg, fSocketFrontEnd, 0);
        fDoMore := zmq_msg_more(fMSG) <> 0;
        zmq_msg_send(fMsg, fSocketBackEnd, IfThen(fDoMore, ZMQ_SNDMORE, 0));
        zmq_msg_close(fMsg);
        if not fDoMore then
          Break; // Это была последняя часть сообщения
      end;
    if (fZMQPoll[1].revents and ZMQ_POLLIN) <> 0 then
      while True do
        // Трансляция сообщний от сервиса к клиенту
      begin // Обработка всх частей сообщения
        zmq_msg_init(fMsg);
        zmq_msg_recv(fMsg, fSocketBackEnd, 0);
        fDoMore := zmq_msg_more(fMSG) <> 0;
        zmq_msg_send(fMsg, fSocketFrontEnd, IfThen(fDoMore, ZMQ_SNDMORE, 0));
        zmq_msg_close(fMsg);
        if not fDoMore then
          Break; // Это была последняя часть сообщения
      end;
  end;
  zmq_close(fSocketFrontEnd);
  zmq_close(fSocketBackEnd);
  zmq_ctx_destroy(fContext);
  Readln;

end.

program BrRR_Broker;

{$APPTYPE CONSOLE}
// Брокер
// Находится в известной клиентам и сервисам конечной точке,
// Находится в известной клиентам конечной точке,
// связывает сокет ZMQ_ROUTER с tcp:*:5559,
// связывает сокет ZMQ_DEALER с tcp:*:5560,
// Перекидывает сообщения между сокетами от FrontEnd к BackEnd
// и обратно

uses
  SysUtils, ZMQ, Math;
var
  fContext: Pointer;
  fSocketFrontEnd: Pointer;
  fSocketBackEnd: Pointer;
begin

  fContext := zmq_ctx_new(); // Инициализация
  fSocketFrontEnd := zmq_socket(fContext, ZMQ_ROUTER);
  fSocketBackEnd := zmq_socket(fContext, ZMQ_DEALER);
  zmq_bind(fSocketFrontEnd, 'tcp://*:5559'); // Конечная точка для клиентов
  zmq_bind(fSocketBackEnd, 'tcp://*:5560'); // Кончная точка для сервисов

  zmq_proxy (fSocketFrontEnd, fSocketBackEnd, nil); // Старт прокси!

  zmq_close(fSocketFrontEnd);
  zmq_close(fSocketBackEnd);
  zmq_ctx_destroy(fContext);
  Readln;

end.

program BrRR_BrokerMeteoInterNet;

{$APPTYPE CONSOLE}
// Брокер для проекта "Метео"
// Реализует мост между интрасаетью метеосервера
// и внешней сетью. Размещается во внутренней сети,
// для брокера открыт "наружу" tcp порт 8100.
// "Внешние" клиенты коннектятся к известной конечной точке tcp://10.1.1.0:8100
// Коннектит сокет ZMQ_XSUB с известной конечной точкой метеосервера tcp://192.168.55.210:5556
// Связывает сокет ZMQ_XPUB с tcp://10.1.1.0:8100,
// Перекидывает сообщения подписки между сокетами от FrontEnd к BackEnd
// и обратно

uses
  SysUtils, ZMQ;
var
  fContext: Pointer;
  fSocketFrontEnd: Pointer;
  fSocketBackEnd: Pointer;
begin

  fContext := zmq_ctx_new(); // Инициализация
  fSocketFrontEnd := zmq_socket(fContext, ZMQ_XSUB);
  fSocketBackEnd := zmq_socket(fContext, ZMQ_XPUB);
  zmq_connect(fSocketFrontEnd, 'tcp://192.168.55.210:5556'); // Конечная точка метеосервиса
  zmq_bind(fSocketBackEnd, 'tcp://10.1.1.0:8100'); // Конечная точка для "внешних" подписчиков
//  zmq_bind(fSocketBackEnd, 'tcp://*:8100'); // Можно и так, главное - чтобы внешние клиенты знали адрес

  zmq_proxy (fSocketFrontEnd, fSocketBackEnd, nil); // Старт прокси

  zmq_close(fSocketFrontEnd);
  zmq_close(fSocketBackEnd);
  zmq_ctx_destroy(fContext);
  Readln;

end.

program PL_Sink;

{$APPTYPE CONSOLE}

uses
  SysUtils, Windows, ZMQ;

const
  c_task_count = 100;

var
  fContext: Pointer;
  fSocketReceiver: Pointer;
  fSocketController: Pointer;
  fStart: Cardinal;
  fTaskCount: Integer;
  fDummy: string;
begin
 // Инициализация
  fContext := zmq_ctx_new();

  // Сокет для приема сигналов
  fSocketReceiver := zmq_socket(fContext, ZMQ_PULL);
  zmq_bind(fSocketReceiver, 'tcp://*:5558');

  // Сокет для отправки сигнала "умри!"
  fSocketController := zmq_socket(fContext, ZMQ_PUB);
  zmq_bind(fSocketController, 'tcp://*:5559');

  // Ожидание первого сигнала о старте пакета задач
  zmq_recv(fSocketReceiver, PChar(nil)^, 0, 0);

  // Фиксация времени начала выполнения пакета
  fStart := GetTickCount();
  // Ждем подтверждения выполнения от 100 рабочих процессов
  for fTaskCount := 0 to Pred(c_task_count) do begin
    zmq_recv(fSocketReceiver, PChar(nil)^, 0, 0);

    if (fTaskCount mod 10 = 0) then // "Статус - бар" :)
      Write(':')
    else
      Write('.')
  end;
  // Вычисление и показ времени выполнения пакета задач
  Writeln('Total elapsed time: ', GetTickCount() - fStart, 'ms');

  // Отправка сигнала "умри!" рабочим процессам
  zmq_send(fSocketController, PChar('KILL')^, 4, 0);

  zmq_close(fSocketReceiver);
  zmq_ctx_destroy(fContext);
  Readln(fDummy);

end.