Коннектор к Kafka
===================

Включает в себя `Kafka producer`_ и `Kafka consumer`_.


Публичные методы
-----------------

get_messages_from_kafka
^^^^^^^^^^^^^^^^^^^^^^^^

Метод, предназначенный для получения сообщений из канала потребителя **Kafka**, отправки их в ``adg_input_processor`` для записи в *storage* и последующего коммита в **Kafka**.

Запускается в отдельном файбере *insert_fiber*. По умолчанию вызывается каждые *0.2* секунды.

send_messages_to_kafka
^^^^^^^^^^^^^^^^^^^^^^^^

Метод, предназначенный для отправки сообщений в **Kafka**.

*Параметры:*

* ``topic_name`` -- имя топика, куда производится отправка;
* ``messages`` -- массив сообщений для отправки в Kafka;
* ``opts`` -- дополнительные опции отправки.

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

* *true* -- при успешной вставке;
* *false,error* -- при ошибке.

get_metric
^^^^^^^^^^^

Метод, предназначенный для передачи метрик данной роли в систему сбора метрик.

*Возвращаемое значение:* Набор метрик в формате *JSON*.


Kafka producer
---------------

Предназначен для отправки сообщений в **Kafka**.

Публичные методы
^^^^^^^^^^^^^^^^^

create
~~~~~~~

Метод, предназначенный для создания экземпляра продьюсера кафки *producer*.

*Параметры:*

* ``brokers`` -- cписок пар хост/порт для подключения к кластеру Kafka;

* ``options`` -- список пар ключ/значение из https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

* ``additional_opts`` -- список дополнительных настроек подключения.

*Конфигурация:* 

* `Настройка брокеров <../Configuration/Kafka_connection_settings/index>`_

* `Настройка продьюсера <../Configuration/Kafka_producer_settings/index>`_


produce
~~~~~~~~

Метод, предназначенный для отправки сообщений в **Kafka** с возможностью выбора синхронно/асинхронно.

*Параметры:*

* ``topic_name`` -- имя топика, куда необходимо отправить сообщения;

* ``messages`` -- список сообщений, которые необходимо загрузить в Kafka;

* ``opts`` -- дополнительные опции.


close
~~~~~~

Метод, предназначенный для удаления экземпляра продьюсера кафки *producer*.

get_producer
~~~~~~~~~~~~~

Метод, предназначенный для возврата текущего эксземпляра *producer*.

*Возвращаемое значение:* Текущий экземпляр *producer*.


Приватные методы
^^^^^^^^^^^^^^^^^

produce_messages
~~~~~~~~~~~~~~~~~

Метод, предназначенный для отправки сообщений в **Kafka**. 

*Параметры:*

* ``topic_name`` -- имя топика, куда необходимо отправить сообщения;

* ``messages`` -- список сообщений, которые необходимо загрузить в Kafka;

* ``is_async`` -- асинхронно или нет отправлять сообщения в Kafka.


Kafka consumer
---------------

Предназначен для отправки сообщений в **Kafka**.

Публичные методы
^^^^^^^^^^^^^^^^^

create
~~~~~~~

Метод, предназначенный для создания экземпляра продьюсера кафки *producer*.

*Параметры:*

* ``brokers`` -- cписок пар хост/порт для подключения к кластеру Kafka;

* ``options`` -- список пар ключ/значение из https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

* ``additional_opts`` -- список дополнительных настроек подключения.

*Конфигурация:* 

* `Настройка брокеров <../Configuration/Kafka_connection_settings/index>`_

* `Настройка продьюсера <../Configuration/Kafka_producer_settings/index>`_


produce
~~~~~~~~

Метод, предназначенный для отправки сообщений в **Kafka** с возможностью выбора синхронно/асинхронно.

*Параметры:*

* ``topic_name`` -- имя топика, куда необходимо отправить сообщения;

* ``messages`` -- список сообщений, которые необходимо загрузить в кафку;

* ``opts`` -- дополнительные опции.


close
~~~~~~

Метод, предназначенный для удаления экземпляра продьюсера кафки *producer*.

get_producer
~~~~~~~~~~~~~

Метод, предназначенный для возврата текущего эксземпляра *producer*.

*Возвращаемое значение:* Текущий экземпляр *producer*.


Приватные методы
^^^^^^^^^^^^^^^^^

produce_messages
~~~~~~~~~~~~~~~~~

Метод, предназначенный для отправки сообщений в **Kafka**. 

*Параметры:*

* ``topic_name`` -- имя топика, куда необходимо отправить сообщения;

* ``messages`` -- список сообщений, которые необходимо загрузить в Kafka;

* ``is_async`` -- асинхронно или нет отправлять сообщения в Kafka.