⚡ В БИТ:ERP открыты вакансии разработчиков 1С, присоединяйся к нашей команде!

PinkRabbitMQ library

Введение

Внешняя Native API компонента для 1C 8.3, которая реализует методы для работы с сервером Rabbit MQ через TCP IP протокол. Компонента поддерживается командой разработки БИТ.MDT (Мобильный терминал данных), БИТ.Адаптер, БИТ.IIoT (Промышленный интернет вещей) и используется в наших продуктах. Ид компоненты - PinkRabbitMQ. Компонента поставляется для 32 и 64 битной платформы 1С.

Поддерживаемые системы

Windows 32 и 64 бит
Linux 64 бит

API

Компонента реализует протокол AMQP 0.9.1. Компонента имеет ряд методов для работы из встроенного языка 1С. Параметры с тегом [НЕ РЕАЛИЗОВАНО] имеют пустую реализацию, но передавать их все равно требуется. Параметры с тегом [НЕОБЯЗАТЕЛЬНЫЙ] можно не передавать.

Список свойств комоненты, связанных с сообщением:

correlationId - строка, чтение и запись
messageId - строка, чтение и запись
Type - строка, чтение и запись
AppId - строка, чтение и запись
ContentEncoding - строка, чтение и запись
ContentType - строка, чтение и запись
UserId - строка, чтение и запись. В данное свойство разрешается передавать только логин пользователя, использованный для установки соединения
ClusterId - строка, чтение и запись
Expiration - число строкой, чтение и запись. Передача строки, отличной от числовой, приведет к вызову исключения.
ReplyTo - строка, чтение и запись

Эти свойства передаются в качестве метаданных вместе с сообщением в RabbitMQ. Их можно прочитать также на стороне получателя - они реинициализируется в момент вызова basicConsumeMessage().

Список прочих свойств:

Version - версия компоненты, только чтение

Список методов:

Connect - устанавливает соединение с сервером RabbitMQ. Таймаут - 5 секунд. Если в течение этого времени сервер недоступен, а также указаны неверные логин, пароль или vhost, то будет возвращена ошибка, которая может быть получена через метод GetLastError.

Параметры:

host - Строка - Адрес сервера RabbitMQ
port - Число - Порт работы через tcp (обычно 5672 для amqp и 5671 для amqps)
login - Строка - Имя пользователя
pwd - Строка - Пароль пользователя
vhost - Строка - Vhost пользователя
pingRate - Число - [НЕ РЕАЛИЗОВАНО][НЕОБЯЗАТЕЛЬНЫЙ]. Частота пульса
secure - Булево - [НЕОБЯЗАТЕЛЬНЫЙ] использовать защищенное соединение по протоколу TLS 1.2. По-умолчанию Ложь. Если Истина, то нужно также передавать соответствующий порт для amqps соединения
timeout - Число - [НЕОБЯЗАТЕЛЬНЫЙ] время ожидания ответа на команды от сервера в секундах. По-умолчанию 5

DeclareExchange - Объявить точку обмена

Параметры:

name - Строка - Имя exchange
type - Строка - Тип точки обмена. Поддерживаются "direct", "fanout", "topic"
onlyCheckIfExists - Булево - Не создавать новую, выбросить исключение, если такой точки нет.
durable - Булево - Сохранять сообщения на диске на случай рестарта RMQ (не рекомендуется)
autodelete - Булево - Удалить после того, как от точки будут отвязаны все очереди.
arguments - Строка - [НЕОБЯЗАТЕЛЬНЫЙ] произвольные свойства в формате Json-объект. Пример: {"x-message-ttl": 60000}.

DeleteExchange - Удалить точку обмена

Параметры:

name - Строка- Имя точки обмена.
ifunused - Булево - Удаление будет выполнено, только если точка обмена не используется.

DeclareQueue - Объявить очередь

Параметры:

name - Строка - Имя объявляемой очереди.
onlyCheckIfExists - Булево - Не создавать очередь с таким именем, использовать существующую
save - Булево - Кешировать сообщения на диске, на случай падения RMQ.
exclusive - Булево - Только текущее соединение может иметь доступ к этой очереди.
autodelete - Булево - Удалить очередь если к ней был подключен, а затем отключен читающий клиент.
maxPriority - Число - [НЕОБЯЗАТЕЛЬНЫЙ] устанавливает максимальное значение приоритета, которое может быть затем установлено для отправляемых сообщений через метод setPriority. Если 0 - то приоритет не используется. По-умолчанию 0.
arguments - Строка - [НЕОБЯЗАТЕЛЬНЫЙ] произвольные свойства в формате Json-объект. Пример: {"x-message-ttl": 60000}. Документация: https://www.rabbitmq.com/queues.html#optional-arguments

Возвращаемое значение:

Имя очереди, заданное явно в 1-м параметре.

DeleteQueue - Удаляет очередь с сервера

Параметры:

name - Строка - Имя очереди
onlyIfIdle - Булево - Удаление будет выполнено, только если очередь не используется
onlyIfEmpty - Булево - Удаление будет выполнено, только если очередь пуста

BindQueue - Установить связь очереди. Создает маршрут от точки обмена до очереди.

Параметры:

queue - Строка - Имя очереди
exchange - Строка - Имя точки обмена
routingKey - Строка - Rлюч маршрутизации.
arguments - Строка - [НЕОБЯЗАТЕЛЬНЫЙ] произвольные свойства в формате Json-объект. Пример: {"x-match": "all"}.

UnbindQueue - Отсоединяет очередь от точки обмена.

Параметры:

queue - Строка - Имя очереди
exchange - Строка - Имя точки обмена
routingKey - Строка - Rлюч маршрутизации.

BasicPublish - Отправить сообщение

Параметры:

exchange - Строка - Имя точки в которую отправляется сообщение
routingKey - Строка - Ключ маршрутизации (см. руководство RMQ)
message - Строка - Тело сообщения
livingTime - Число - [НЕ РЕАЛИЗОВАНО] Время жизни сообщения в миллисекундах
persist - Булево - Сбрасывать сообщение на диск
arguments - Строка - [НЕОБЯЗАТЕЛЬНЫЙ] произвольные свойства в формате Json-объект. Пример: {"x-message-ttl": 60000}.

BasicAck Отсылает серверу подтверждение (ack), что сообщение обработано и его можно удалить. Подтвердить можно любое прочитанное сообщение.

Параметры:

messageTag - число - тег сообщения, который возвращается в параметре метода BasicConsumeMessage

BasicReject - Отказывается от последнего полученного сообщения. Работает по принципу Ack, но в обратную сторону.

Параметры:

messageTag - число - тег сообщения, который возвращается в параметре метода BasicConsumeMessage

BasicConsume - Начать чтение. Регистрирует потребителя сообщений для очереди.

Параметры:

queue - Строка - Очередь из которой будем читать сообщения.
consumerId - Строка - [НЕ РЕАЛИЗОВАНО] имя потребителя. Если не задан, то имя потребителя сгенерирует сервер и вернет из метода
noConfirm - Булево - не ждать подтверждения обработки. Сообщения будут удалены из очереди сразу после отправки на клиента.
exclusive - Булево - монопольно захватить очередь
selectSize - Число - количество единовременно считываемых сообщений из очереди в кеш компоненты. Оптимизационный параметр, который влияет на скорость забора сообщений. Рекомендуемый диапазон 100-1000. Нежелательно устанавливать слишком высокие значения, т.к. чтение большого числа накопленных сообщений в очереди может спровоцировать нехватку памяти на клиенте 1С и падение компоненты без вызова исключения.
arguments - Строка - [НЕОБЯЗАТЕЛЬНЫЙ] произвольные свойства в формате Json-объект. Пример: {"x-stream-offset": "first"}.

Возвращаемое значение:

Строка. Имя потребителя, сгенерированное сервером или переданное в параметре ИмяПотребителя.

BasicConsumeMessage - Получить сообщение

Параметры:

consumerId - Строка - [НЕ РЕАЛИЗОВАНО] Имя зарегистрированного потребителя
outdata - Строка - Выходной параметр. Тело сообщения.
messageTag - Число - Выходной параметр. Тег сообщения для подтверждения через метод BasicAck
timeout - Число - Таймаут ожидания сообщения в миллисекундах. Не рекомендуется ставить параметр слшком низким, т.к. сообщение просто может не успеть прийти из очереди. Рекомендуемый диапазон 3000-60000.

Возвращаемое значение:

Булево. Истина, если из очереди прочитано очередное сообщение. Ложь, если очередь пустая.

BasicCancel - Закрывает канал для чтения сообщений

Параметры:

channelId - Строка - Имя созданного ранее потребителя.

SetPriority -устанавливает приоритет для всех отправляемых сообщений. Если значение равно 0, то приоритет не будет установлен.

Параметры:

priority - Число - значение в диапазоне разрешенного приоритета для очереди, но не более 255.

GetPriority - получает приоритет для последнего считанного сообщения. Имеет смысл вызывать внутри цикла чтения сообщений методом BasicConsumeMessage

Параметры: Отсутствуют

Возвращаемое значение:

числовое значение приоритета от 0 до 255 для сообщения

GetRoutingKey - получает ключ маршрутизации для последнего считанного сообщения. Имеет смысл вызывать внутри цикла чтения сообщений методом BasicConsumeMessage

Параметры: Отсутствуют

Возвращаемое значение:

Строка - ключ маршрутизации очереди считанного сообщения.

GetHeaders - получает заголовки для последнего считанного сообщения. Имеет смысл вызывать внутри цикла чтения сообщений методом BasicConsumeMessage

Параметры: Отсутствуют

Возвращаемое значение:

Строка - заголовки считанного сообщения в формате объекта json. Пример: {"customIntHeader": 42}.

GetLastError - получает информацию о последней ошибке

Параметры отсутствуют

Возвращаемое значение:

Строка - Последнее сообщение ошибки

SleepNative - останавилвает выполнение программы на указанное время

Параметры:

timeout - число - время остановки в милисекундах

Правила работы с компонентой из 1С

Завершать работу с компонентой посредством обнуления объекта компоненты. Например:

Компонента = Новый("AddIn.BITERP.PRMQMOBILE");
// Работа с компонентой
Компонента = Неопределено;

Все выходные параметры в методах компоненты передавать как Неопределено или пустая строка, а также обнулять возвращенные значения. Несоблюдение этого простого правила приводит к огромным утечкам памяти!!! Например:

Клиент = Новый("AddIn.BITERP.PinkRabbitMQ");
ТекстСообщения = "";
ТегСообщения = 0;
Пока Клиент.BasicConsumeMessage(Потребитель, ТекстСообщения, ТегСообщения, Таймаут) Цикл
    // Работа с компонентой
    // ...
    //
    ТекстСообщения = ""; 
    ТегСообщения = 0; // при каждой итерации очищаем память по указателю, который неявной хранится в этой переменной
КонецЦикла;

Не вызывать исключения внутри цикла чтения сообщений. Это приводит к потерянной памяти в rphost-е, т.к. внутри компоненты могут остаться закешированные сообщения, которые не попали в алгоритм очистки ссылок.

Клиент = Новый("AddIn.BITERP.PinkRabbitMQ");
ВыходнойПараметр = "";
Пока Клиент.BasicConsumeMessage(Потребитель, ВыходнойПараметр, Таймаут) Цикл
    // Работа с компонентой
    Если Клиент.CorrelationId <> "МОЙ_ИД" Тогда
        ВызватьИсключение "Ошибка чтения свойств!"; // ТАК ДЕЛАТЬ СТРОГО НЕ РЕКОМЕНДУЕТСЯ!!!
    КонецЕсли;
    ВыходнойПараметр = "";
КонецЦикла;

На сервере все исключения компоненты обрабратывать через Попытку ... Иключение с вызовом в секциии исключения метода GetLastError() для получения истинной детальной ошибки. Пример

Клиент = Новый("AddIn.BITERP.PinkRabbitMQ");
Попытка
    Клиент.DeclareQueue(ИмяОчереди, Ложь, Ложь, Ложь, Ложь);
Исключение
    ВызватьИсключение Клиент.GetLastError();
КонецПопытки

Примеры

Подключение внешней компоненты на клиенте

УстановитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ");
ПодключитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ", "BITERP", ТипВнешнейКомпоненты.Native);
Клиент  = Новый("AddIn.BITERP.PinkRabbitMQ");

Создание точки обмена, очереди и их связывание на клиенте

ИмяТочкиОбмена = "data";
ИмяОчереди = "testroute";
ОтправляемоеСообщение = "Test";
ОтветноеСообщение = "";

УстановитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ");
ПодключитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ", "BITERP", ТипВнешнейКомпоненты.Native);
Клиент  = Новый("AddIn.BITERP.PinkRabbitMQ");

Клиент.Connect("127.0.0.1", 5672, "user", "password", "user");
Клиент.DeclareExchange(ИмяТочкиОбмена, "topic", Ложь, Истина, Ложь);
Клиент.DeclareQueue(ИмяОчереди, Ложь, Ложь, Ложь, Ложь);
Клиент.BindQueue(ИмяОчереди, ИмяТочкиОбмена, "#" + ИмяОчереди + "#");

Создание очереди, отправка и получение сообщения на клиенте

    ИмяОчереди = "testroute";
    ОтправляемоеСообщение = "Test";
    ОтветноеСообщение = "";
    ТегСообщения = 0;
    
    УстановитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ");
    ПодключитьВнешнююКомпоненту("ОбщийМакет.PinkRabbitMQ", "BITERP", ТипВнешнейКомпоненты.Native);
    Клиент  = Новый("AddIn.BITERP.PinkRabbitMQ");
    
    Клиент.Connect("127.0.0.1", 5672, "user", "password", "vhost");
    Клиент.DeclareQueue(ИмяОчереди, Ложь, Ложь, Ложь, Ложь);
    Клиент.BasicPublish("", ИмяОчереди, ОтправляемоеСообщение, 0, Ложь);
    Попытка
        Потребитель = Клиент.BasicConsume(ИмяОчереди, "", Истина, Ложь, 0);
        Пока Клиент.BasicConsumeMessage("", ОтветноеСообщение, ТегСообщения, 5000) Цикл
            Клиент.BasicAck(ТегСообщения);
            Сообщить("Успешно! Из очереди прочитано сообщение " + ОтветноеСообщение);
            ОтветноеСообщение = ""; // Обнуляем, чтобы избежать утечку памяти
            ТегСообщения = 0; // Обнуляем, чтобы избежать утечку памяти
        КонецЦикла;
        Клиент.BasicCancel("");
    Исключение
        Сообщить(Клиент.GetLastError());
    КонецПопытки;

Установка в конфигурацию

Скачать архив с релизом компоненты.
Загрузить в конфигурацию в качестве общего макета с двоичными данными

Поддержка TLS

Компонента поддерживает соединение по протоколу TLS версии 1.2. Сертификаты сервера верифицируются при подключении, клиентские сертификаты не поддерживаются. На сервере RabbitMQ необходимо отключить проверку клиентских сертификатов опциями ssl_options.verify и ssl_options.fail_if_no_peer_cert. Подробнее про опции можно узнать из официальной документации: https://www.rabbitmq.com/ssl.html#peer-verification-configuration

Разворачивание окружения разработки на Windows 10 (Рекомендуется)

Вместе с проектом поставляются заранее скомпилированные зависимости в папке libsPrecompiled для 32 и 64 битных платформ в вариантах debug и release. Поэтому для создания окружения разработки нужно сделать только следующее:

Установить Visual Studio 2019 (можно Community). При установке выбрать среди обязательных компонентов также WIndows SDK 10
Скачать исходники репозитория данного проекта.
Запустить файл проекта AddIn.sln в корневом каталоге

Разворачивание окружения разработки на Windows 10 и самостоятельная компиляция всех зависимостей (Не рекомендуется)

Порядок изложенный ниже требуется, если нужно с нуля собрать весь проект и скомпилировать все зависимые билиотеки. Если это не требуется, то рекомендуется следовать простому порядку разворачивания окружения, приведенному выше.

Установить Visual Studio 2019 (можно Community). При установке выбрать среди обязательных компонентов также WIndows SDK 10
Скачать исходники репозиторий данного проекта.
Скачать исходники либы для работы с протоколом раббита AMQP-CPP 4.1.4. https://github.com/CopernicaMarketingSoftware/AMQP-CPP в папку libs
Скомпилировать дебажную и релизную dll в папкe libs\AMQP-CPP-4.1.4\build с помощью cmake.

mkdir build
cd build
cmake .. 
cd ..
cmake --build

Для компиляции 64-битной release версии либы

Открыть файл проекта amqpcpp.sln
Выбрать свойства проекта amqcpp => C/C++ => Code generation => Runtime library => Multi-threaded (/MT)
Скомпилировать проект Release

Для компиляции 32-битной release версии либы

Нужно создать свой проект с нуля на базе проекта 64-битной компоненты, сгенерированного через cmake

Скачать исходники либы для работы с сокетом POCO 1.9.0. https://github.com/pocoproject/poco/tree/master в папку libs
Скомпилировать дебажную .lib и .dll (debug_shared) и релизную (release_static_mt) для следующих подпроектов Foundation и Net. Компиляцию следует выполнять через sln проекты
Для дебажного запуска компоненты поместить скомпилированные dll-ки PocoFoundationd64.dll и PocoNet64d.dll в папку 1cv8\8.3.10.2667\bin
Запустить файл проекта компоненты NativeRabbitMQClient в каталоге AddIn.sln
Проверить линковку зависимостей для заголовков Properties узла проекта -> С/С++ -> General -> Additional include directories (переключить вкладку debug)

..\libs\poco-poco-1.9.0-release\Foundation\include;
..\libs\poco-poco-1.9.0-release\Net\include;
..\libs\AMQP-CPP-4.1.4\include;

Проверить линковку зависимостей для папок lib и dll файлов Properties узла проекта -> Linker -> General -> Additional include directories

..\libs\poco-poco-1.9.0-release\lib64;
..\libs\AMQP-CPP-4.1.4\build\bin\Debug;

Проверить линкову для lib файлов зависимостей Properties узла проекта -> Linker -> Input -> Additional dependencies

amqpcpp.lib;
PocoFoundationd.lib;

Скомпилировать компоненту через проект

Разворачивание окружения разработки для Linux из-под WIndows 10

Установить Visual Studio Community 2019 с последними обновлениями
Развернуть Linux system for Windows 10 и подключить VS Studio к своей машине Туториал
Запустить проект PinrRabbitMQ для Linux

Сборка проекта через cmake для Windows 10

mkdir build
cd build
cmake .. 
cmake --build . --config Release

Сборка проекта через Visual Studio для Linux

Запустить проект PinrRabbitMQ для Linux
Открыть свойства проекта и в меню Confugartion properties -> General -> Configuration type выбрать Dynamic library (so)
Скомпилировать проект через f5
Либа будет скомпилирована в каталоге ~/projects/PinkRabbitMQLinux/bin/x64/debug/ удаленной машины

Сборка проекта через cmake для Linux

Сборку необходимо производить на Ubuntu 16 или Debian 8.

# Установить необходимые пакеты
sudo apt install build-essential cmake

# Собрать библиотеку
cd PinkRabbitMQLinux
mkdir build
cd build
cmake .. 
cmake --build .

Запуск unit тестов в Visual Studio 2019

В паке выполнения тестов \windows\tests переименовать файл test.conf.example в test.conf и указать в нем параметры подключения к тестовому кролику
Скопировать в папку выполнения тестов \windows\x64\Debug собранные dll библиотеки poco из \lib\win_x64\cfg_Debug
Запустить тесты из инетрфейса Visual Studio

Лицензия

Лицензировано на условиях MIT. Смотрите файл LICENSE

Используемые сторонние продукты

AMQP-CPP - для взаимодействия с RabbitMQ на основе лицензии Apache
POCO - для работы с сокетом по TCP протоколу для windows на основе лицензии Boost
LIBEVENT - для работы с сокетом по TCP протоколу для Linux на основе 3-clause BSD лицензии

BITERP/PinkRabbitMQ

PinkRabbitMQ library

Введение

Поддерживаемые системы

API

Список свойств комоненты, связанных с сообщением:

Список прочих свойств:

Список методов:

Правила работы с компонентой из 1С

Примеры

Подключение внешней компоненты на клиенте

Создание точки обмена, очереди и их связывание на клиенте

Создание очереди, отправка и получение сообщения на клиенте

Установка в конфигурацию

Поддержка TLS

Разворачивание окружения разработки на Windows 10 (Рекомендуется)

Разворачивание окружения разработки на Windows 10 и самостоятельная компиляция всех зависимостей (Не рекомендуется)

Разворачивание окружения разработки для Linux из-под WIndows 10

Сборка проекта через cmake для Windows 10

Сборка проекта через Visual Studio для Linux

Сборка проекта через cmake для Linux

Запуск unit тестов в Visual Studio 2019

Лицензия

Используемые сторонние продукты