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.