Конфигурация REST egress#

Данный раздел применим для роли Администратор.

Предусловия#

Не требуются.

Последовательность выполнения#

EVTA поддерживает работу с помощью REST взаимодействия. Если в конфигурационном файле adapter.conf в поле egressEnabled установлено rest, то EVTA позволяет обращаться извне к брокерам сообщений при помощи REST взаимодействия соответственно.

В конфигурации в блоке egressSubscribe указываются настройки egress, который позволяет вычитывать сообщения извне, а egressPublish, наоборот, отправлять сообщения. Для egressSubscribe в конфигурации должны быть указаны настройки для консьюмера, для egressPublish — настройки для продюсера.

Настройки для брокера берутся из файла, указанного в urlConfig.

Пример заполненного файла urlConfig.json (путь до него указывается в файле adapter.conf)

{
  "fpss": {
    "Domain": {
      "Federation": {
        "SegmentA": {
          "SystemName": {
            "EventName": {
              "1": {
                "type": "kafka",
                "topic": "EventNameTopic",
                "properties": "/path/to/producer.properties"
              }
            },
            "OtherEventName": {
              "1": {
                "type": "kafka",
                "topic": "OtherEventNameTopic",
                "properties": "/path/to/other_producer.properties"
              }
            }
          }
        }
      }
    }
  }
}

Внешний API сервиса#

Принципы работы REST-взаимодействий EVTA#

EVTA обслуживает запросы только типа POST, GET и PATCH. На остальные типы запросов будет дан ответ с кодом 405 «Method Not Allowed».

POST#

Используется для публикации событий в транспорт (например, в топик Platform V Corax / Apache Kafka, очередь/адрес SMBX, IBM MQ, RabbitMQ). Запись происходит по одному сообщению или событию.

Вид URL: https://host:port/publish/segment/domain/federation/publisher/name/version

, где:

• host — сервер, где развернут EVTA;

• port — порт, на котором развернут EVTA;

• publish — наименование метода, константа;

• segment — сетевая зона, где развернут домен, в который публикуется событие;

• domain — уникальное (в рамках одной федерации) имя событийного домена, в который публикуется событие;

• federation — уникальное имя федерации доменов, в которой находится домен, куда публикуется событие;

• publisher — имя источника события (АС);

• name — название типа события;

• version — номер версии типа события.

Если в файле конфигурации EVTA adapter.conf значение параметра isHttp2: true – ключ, тело и заголовки берутся из соответствующих полей сообщения для отправки в транспорт.

Пример запроса:

curl -X 'POST' --http2  'http://localhost:9002/publish/segment/Domain/Federation/System/TESTA/1' -d "{\"body\":\"body value\",\"key\":\"key value\",\"headers\":{\"header1\":\"header value\"}}"

Пример сообщения в JSON-формате:

{
    "body": "тело сообщения, обязательный параметр",
    "key": "ключ сообщения, опциональный параметр",
    "headers": {
      "header1": "заголовок сообщения, опциональный параметр",
      "header2": "заголовок сообщения, опциональный параметр"
  }
}

Если в файле конфигурации EVTA adapter.conf значение параметра isHttp2: false – тело запроса является телом сообщения в транспорте (сообщение может быть в любом формате, не только JSON). Дополнительно в запросе могут быть указаны флаги заголовков запроса:

• fpss.key — ключ сообщения;

• заголовки, начинающиеся с префикса fpss, например fpss.Header_Name — заголовки сообщения, их может быть несколько.

Пример запроса:

curl -X 'POST' 'http://localhost:9002/publish/segment/Domain/Federation/System/TESTA/1' -H 'fpss.header_name: string' -H 'fpss.key: 123'  -H 'fpss.transport-id: 1'  -d 'message body'

В случае успешного выполнения запроса придет ответ с кодом 201 «Created».

Если при выполнении запроса произойдет какая-либо ошибка, то вернется ответ с текстом ошибки и кодом 500 «Internal Server Error».

GET#

Используется для получения событий из транспорта (например, из топик Platform V Corax / Apache Kafka, очереди/адреса SMBX, IBM MQ, RabbitMQ). Вычитка происходит по одному сообщению.

Вид URL: https://host:port/subscribe/segment/domain/federation/publisher/name/version

, где:

• host — сервер, где развернут EVTA;

• port — порт, на котором развернут EVTA;

• subscribe — наименование метода, константа;

• segment — сетевая зона, где развернут домен, из которого будет производиться вычитка сообщений;

• domain — уникальное (в рамках одной федерации) имя событийного домена, из которого будет производиться вычитка сообщений;

• federation — уникальное имя федерации доменов, в которой расположен домен;

• publisher — имя источника события АС;

• name — название типа события;

• version — номер версии типа события.

Если в файле конфигурации EVTA adapter.conf значение параметра isHttp2: false – будет получено одно сообщение любого типа. Само сообщение транспорта будет телом сообщения в запросе, ключ сообщения будет указан в поле «fpss.key», заголовки сообщения транспорта будут указаны в заголовках запроса с префиксом «fpss.».

В случае успешного выполнения запроса придет ответ с кодом 200 «OK» и телом сообщения.

Пример запроса:

curl -X 'GET' http://localhost:9005/subscribe/segment/domain/federation/System/TESTA/1

Пример полученного ответа:

HTTP/1.1 200 OK
content-type: text/plain; charset=UTF-8 # заголовок ответа
content-length: 10 #заголовок ответа
fpss.header2: test # заголовок сообщения транспорта
fpss.header1: test # заголовок сообщения транспорта
fpss.key: key value # ключ сообщения транспорта
fpss.transport-id: 2 # id коммита
connection: close

* Closing connection 0
body value #тело сообщения

Если в файле конфигурации EVTA adapter.conf значение параметра isHttp2: true – в результате запроса будет получен поток сообщений. Ключ, тело и заголовки для каждого сообщения берутся из соответствующих полей сообщения транспорта.

В случае успешного выполнения запроса придет ответ с кодом 200 «OK» и поток сообщений.

Пример запроса:

curl -X 'GET' --http2 http://localhost:9005/subscribe/segment/domain/federation/System/TESTA/1

Получаемые сообщения в JSON-формате:

{
    "body": "тело сообщения, обязательный параметр",
    "key": "ключ сообщения, опциональный параметр",
    "transportId": "id для коммита",
    "headers": {
      "header1": "заголовок сообщения, опциональный параметр",
      "header2": "заголовок сообщения, опциональный параметр"
  }
}

Пример полученного ответа:

HTTP/2 200

{"headers":{"isTest":"true"},"body":"test value","key":"test key","transportId":"0"}
{"headers":{"isTest":"true"},"body":"test value 1","key":"test key 1","transportId":"1"}
{"headers":{"isTest":"true"},"body":"test value 2","key":"test key 2","transportId":"2"}
{"headers":{"header3":"test","isTest":"true"},"body":"test value 3","key":"test key 3","transportId":"3"}

Если при выполнении запроса произойдет какая-либо ошибка, то вернется ответ с текстом ошибки и кодом 500 «Internal Server Error».

Вычитка сообщений из Platform V Corax / Apache Kafka topics без смещения offset и функциональность по смещению offset в группе#

Для вычитки сообщений из Platform V Corax / Apache Kafka topics без смещения offset необходимосформировать URL, содержащий партицию и offset, по которым необходимо вычитать сообщение:

Вид URL: http://host:port/subscribe/segment/domain/federation/publisher/name/version/partition/offset , где:

• host — сервер, где развернут EVTA;

• port — порт, на котором развернут EVTA;

• subscribe — наименование сервиса публикации;

• segment — сетевая зона, где развернут домен;

• domain — уникальное (в рамках одной федерации) имя событийного домена;

• federation — уникальное имя федерации доменов;

• publisher — имя источника события;

• name — название типа события;

• version — номер версии типа события;

• partition — номер партиции топика, из которого необходимо вычитать сообщение;

• offset — номер offset, по которому необходимо вычитать сообщение.

Особенности работы

• После вычитки commit offset вычитанного сообщения не происходит.

• При обработке запроса в лог будет выведено сообщение:

On partition: partitionNumber will seek offset: offsetNumber

• При удачной вычитке сообщения по указанной партиции и offset в лог будет выведено сообщение:

Message from topic=topicName with partition=partitionNumber, offset=offsetNumber read successfully

Для того, чтобы осуществить смещение offset в группе, в запросе необходимо сформировать заголовки с ключами shift.partition, shift.offset.

Пример запроса для смещения offset на позицию 10 для партиции 0:

curl -X 'GET' -v -k http://host:port/subscribe/segment/domain/federation/publisher/name/version -H "shift.partition: 0" -H "shift.offset: 10"

Особенности работы

• Для корректной обработки запроса необходимо выполнить следующие условия:

  • для отправки запросов на смещение offset в группе необходимо поднять отдельный экземпляр EVTA, с которого будут направляться такие запросы;

  • при наличии нескольких партиций в топике должна быть, как минимум, одна свободная партиция, из которой не происходит вычитка сообщений;

  • при обработке запроса на смещение offset в группе необходимо, чтобы происходила вычитка сообщений из топика всеми клиентами, состоящими в группе.

• При удачном смещении offset в лог будет выведено сообщение:

Offset for group was shifted successfully

• При удачном выполнении смещения offset результатом выполнения запроса будет являться сообщение, вычитанное из произвольной партиции топика.

PATCH#

Используется для получения подтверждения АС сообщения из транспорта (например, из топика Platform V Corax / Apache Kafka, очереди/адреса SMBX, IBM MQ, RabbitMQ). Используется после отправки GET-запроса.

При вычитке сообщений по запросу GET вместе с сообщением приходит ID (параметр transport-id в сообщении), по которому АС может подтвердить получение данного сообщения.

Если в файле конфигурации adapter.conf:

• isHttp2: false – ID будет находиться в заголовке «fpss.transport-id»;

• isHttp2: true — в теле сообщения в поле transportId;

• autoCommit: true – подтверждение получения сообщений АС производится сразу после отправки автоматически;

• autoCommit: false – необходимо отправлять коммиты в формате, т.е. использовать метод PATCH.

Отправляемые данные в JSON-формате:

{
    "idsToCommit": <массив id сообщений>
}

Вид URL: https://host:port/commit/segment/domain/federation/publisher/name/version

, где:

• host — сервер, где развернут EVTA;

• port — порт, на котором развернут EVTA;

• commit — наименование метода, константа;

• segment — сетевая зона, где развернут домен, из которого производится вычитка сообщений;

• domain — уникальное (в рамках одной федерации) имя домена, из которого производится вычитка сообщений;

• federation — уникальное имя федерации доменов;

• publisher — имя источника события АС;

• name — название типа события;

• version — номер версии типа события.

Пример запроса:

curl -X 'PATCH'  http://localhost:9002/commit/segment/Domain/Federation/System/TESTA/1 -d "{\"idsToCommit\": [0]}"

Пример полученного ответа:

HTTP/1.1 200 OK
content-type: application/json
content-length: 43
connection: close

* Closing connection 0
{"status":"Successfully commit 1 messages"}

Если при выполнении запроса произойдет какая-либо ошибка, то вернется ответ с текстом ошибки и кодом 500 «Internal Server Error».

Возвращаемые статусы#

Правила эксплуатации#

Не требуется.

Результат#

Конфигурация REST egress настроена.