Репликация через Kafka

Сообщения об изменённых объектах выкладываются в топик SemanticMDM.<ИдентификаторСистемы>+ ключ маршрутизации по приоритету при наличии. Атрибуты исходящего сообщения:

Наименование	Значение	Пояснение
id	Целое число	Идентификатор исходящего сообщения
Header: reply_to	Строка	Наименование топика, в который следует отправлять сообщения об обработке
Тело сообщения	Строка в кодировке UTF-8	Строка является сериализованным в Json ExportRootObject

Функции RPC

Вызов функций RPC осуществляется через топик SemanticMDM.RPC. Ответ, при потребности, будет опубликован в топик указанный в заголовке replyTo. Если размер ответа превышает лимит max.request.size, оно будет разбито на несколько частей. Каждая из частей будет содержать заголовки

• size - полный размер сообщения в байтах
• indxex - номер части, zero-based
• finished - признак последней части true/false

Трансферы параметров вызова функции должны быть унаследованы от базового класса RpcFunction, чтобы при сериализации их в Json возникло указание на фактический тип данных в атрибуте @type.

@XmlSeeAlso({RpcGetFile.class, RpcPing.class})
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonSubTypes({
        @JsonSubTypes.Type(value = RpcGetFile.class),
        @JsonSubTypes.Type(value = RpcPing.class),
})
public abstract class RpcFunction {
}

RpcPing

public class RpcPing extends RpcFunction {
    public String systemId;
}

Сообщить SemanticMDM, что система с идентификатором systemId жива и готова к приёму сообщений. Рекомендуется вызывать функцию один раз в минуту. Возвращаемый результат отсутствует. Пример Json:

{
  "@type": "RpcPing",
  "systemId": "1C"
}

RpcGetFile

public class RpcGetFile extends RpcFunction {
    public String fileHash;
}

Получить тело файла с известным хэшем. Возвращаемый результат – массив байт. Пример Json:

{
  "@type": "RpcGetFile",
  "fileHash": "123456789"
}

Обработка сообщения

После обработки входящего сообщения, успешной или неуспешной, клиент обязан уведомить SemanticMDM о данном факте. Уведомление должно быть выслано в очередь, указанную в атрибуте reply_to принятого сообщения со следующими атрибутами:

Наименование	Значение	Пояснение
id	Целое число	Идентификатор обработанного сообщения
header: systemId	Строка	Идентификатор внешней системы
Тело сообщения	Строка в кодировке UTF-8	Строка должна быть сериализованным в Json ExportedObjectReply

class ExportedObjectReply {
    String sourceId; // Идентификатор созданного объекта, если был создан, в терминах клиента 
    Set<String> deletedSourceIds; // Идентификаторы удалённых объектов в терминах клиента 
    String info; // Информация о процессе обработки для журналирования 
    boolean encounteredError; // Возникла ли ошибка во время обработки 
    String mdmId; // Идентификатор обработанного объекта SemanticMDM как он был указан во входящем сообщении 
}

Пример Json:

{
  "sourceId": "some-key-value",
  "deletedSourceIds": [
    "deleted-key-1",
    "deleted-key-2"
  ],
  "info": "successfully parsed",
  "encounteredError": false,
  "mdmId": "some-item-uuid"
}

Нотификация через Kafka

Уведомления рассылаются в kafka по требованию в system properties:

# параметры подключения kafka producer с префиксом "kafka.producer." . Полный список см в org.apache.kafka.clients.producer.ProducerConfig
# если нет ни одного параметра kafka.producer.*, считаю что kafka не используется
#./system-property=kafka.producer.bootstrap.servers:add(value=хост:порт)
# топики нотификации по потребности
#./system-property=kafka.producer.notification.topics:add(value="transactional-element,app-processing-journal,source-record")
# параметры подключения kafka consumer с префиксом "kafka.consumer." . Полный список см в org.apache.kafka.clients.consumer.ConsumerConfig
# если нет ни одного параметра kafka.consumer.*, считаю что kafka не используется
#./system-property=kafka.consumer.bootstrap.servers:add(value=хост:порт)

Уведомление высылается только при указании топика в kafka.producer.notification.topic

Topic: transactional-element

Уведомление об изменении транзакционного элемента после публикации пользовательской транзакции. Уведомления сгруппированы по псевдо-типам Item, Node, Transaction, ConstraintTable. В топике гарантируется, что сообщения по всем типам одной транзакции высылаются

• в одну партицию
• до сообщения типа Transaction

Сообщение типа Transaction высылается последним.

Тип Item

Уведомление об изменении айтема или его зависимостей

Id - дескриптор ручки айтема

Headers

• type == Item
• action - тип действия, Created, Edited, Deleted
• transaction - ид изменившей транзакции
• transactionOwner - владелец транзакции
• parentNodeUid - гуид ноды
• nestedleft - левая граница интервала айтема
• nestedRight - правая граница интервала айтема
• itemStatus - статус айтема

Body

public static class ItemBody {
    // Наименование
    public final String displayName;
    // Гуид
    public final String uid;
    // Дескриптор ручки или ид
    public final Long descriptor;
    // Признак наличия изменений в ручке
    public final boolean changedHandle;
    // Признак наличия изменений в списке дубликатов
    public final boolean changedDuplicates;
    // Признак операции перемещения
    public final boolean moved;
    // Идентификаторы внешних систем с изменёнными переходными ключами
    public final Set<String> changedKeysSystem;
    // Гуиды свойств с изменёнными значениями
    public final Set<String> changedValuesProperties;
    // Комментарий
    public final String comment;
}

Тип Node

Уведомление об изменении ноды, её деклараций или ограничений

Id - дескриптор ручки ноды

Headers

• type == Node
• action - тип действия, Node, Created, Edited, Deleted,
• transaction - ид изменившей транзакции
• transactionOwner - владелец транзакции
• parentNodeUid - гуид родительской ноды
• nestedleft - левая граница интервала ноды
• nestedRight - правая граница интервала ноды

Body

public static class NodeBody extends Body {
    // Наименование
    public final String displayName;
    // Гуид
    public final String uid;
    // Дескриптор родительской ноды
    public final Long parentDescriptor;
    // Комментарий
    public final String comment;
    // Идентификатор еОТД
    public final String eOTDid;
    // Признак наличия скрипта обогащения после создания айтема
    public final boolean hasCreateEnrichmentFormula;
    // Признак наличия скрипта обогащения после изменения статуса айтема
    public final boolean hasStatusUpdateEnrichmentFormula;
    // Нода содержит айтемы структуры древовидного классификатора        
    public final Boolean isTreeClassifierStructure;
    // Декларации и ограничения
    public final List<Supposition> suppositions;
}

public static abstract class Supposition {
    // Тип действия над объектом, Created, Edited, Deleted
    public Action action;
    // Тип объекта
    public String type;
}

public static class Declaration extends Supposition {
    // PropertyDeclaration
    // Ид свойства
    public final Long propertyId;
    // Гуид свойства
    public final String propertyUid;
    // Идентификатор еОТД свойства
    public final String eOTDid;
}

public static class Constraint extends Supposition {
    // дескриптор ограничения
    public final Long descriptor;
}

Тип ConstraintTable

Уведомление об изменении строк или колонок ограничительной таблицы

Id - идентификатор ограничительной таблицы

Headers

• type == ConstraintTable
• action == Edited
• transaction - ид изменившей транзакции
• transactionOwner - владелец транзакции

Body

public static class ConstraintTableBody {
    public final Long tableId;
    // Дескрипторы нод
    public final List<Long> nodeDescriptors;
    // Дескрипторы ограничений
    public final List<Long> constraintDescriptors;
    // Дескриптор селектора
    public final Long selectorDescriptor;
}

Тип Transaction

Уведомление о публикации пользовательской транзакции

Id - идентификатор транзакции

Headers

• type == Transaction
• action == Created
• transactionOwner - владелец транзакции

Body

public static class TransactionBody {
    // Наименование
    public final String displayName;
    // Дата публикации
    public Date finishDate;
    // Ид связанной заявки
    public long relatedApplication;
    // Строковый ид связанной заявки
    public String relatedAppUid;
    // Приоритет экспорта через репликацию
    public ReplicaExportPriority replicaPriority;
}

Topic: app-processing-journal

Уведомление об изменении статуса заявки

Id - идентификатор заявки

Headers

• uid - строковый идентификатор заявки
• priority - порядок сортировки по приоритету. Чем больше prioritySortOrder тем хуже приоритет
• template - идентификатор шаблона или null для заявок без шаблона
• applicant - заявитель
• executor - исполнитель

Body

public static class Body {
    // Пользователь, сгенерировавший событие
    public String account;
    // Предыдущий статус заявки
    public ApplicationStatus status;
    // Следующий статус заявки
    public ApplicationStatus nextStatus;
    // Тип процесса: Init, Correction, Execution, Negotiation
    public ApplicationProcessType processType;
    // Идентификатор процесса, начинает отсчитываться с 0 в каждой заявке
    public long processId;
    // Список ид прототипов этапов согласования
    public JsonLongsList negotiationStageProtoIds;
    // Список групп согласующих
    @JsonProperty("negotiantAppUserGroups")
    public JsonStringsList negotiantDepartments;
    // Текущая группа этапов согласования
    public String negoStageGroup;
    // Дата начала процесса
    public Date dateCreated;
    // Дата завершения процесса
    public Date dateFinished;
    // Время выполнения, секунд
    public long spentSecDuration;
    // Рабочее время выполнения, секунд
    public long spentLabourSecDuration;
    // Завершено в рабочее-нерабочее время: Pause, Labour
    public AppProcessingJournal.EventMomentType eventMomentType;
    // Время выполнения превысило предельное: Ok, Expired
    public AppProcessingJournal.ExpirationType expirationType;
    // Тип события по отношению к процессу: ProcessImmutable, ProcessMutatedBack, ProcessMutatedForward, AppWorkflowTerminated
    public AppProcessingJournal.EventType eventType;
    // Предельное время выполнения, секунд
    public Long targetSecDuration;
    // Время выполнения процесса, секунд
    public Long processSecDuration;
    // Статус согласования: Awaiting, PartialNego, NegoDone, Canceled, NegoGroupDone
    public AppProcessingJournal.NegoStatus negoStatus;
    // Время в секундах, которое было затрачено на согласование/отклонение этапа(ов)
    public Long negoSecDuration;
    // Рабочее время в секундах, которое было затрачено на согласование/отклонение этапа(ов)
    public Long negoLabourSecDuration;
    // Предельное время в секундах, допустимое для согласования/отклонения этапа(ов)
    public Long negoTargetSecDuration;
    // Коды причин изменения статуса
    public JsonStringsList reasonCodes;
    // Комментарий
    public String comment;
}

Topic: app-expired-events

Уведомление о просрочке исполнения или согласования заявки

Id - идентификатор заявки

Headers

• uid - строковый идентификатор заявки
• priority - порядок сортировки по приоритету. Чем больше prioritySortOrder тем хуже приоритет
• template - идентификатор шаблона или null для заявок без шаблона
• applicant - заявитель
• executor - исполнитель
• eventtype - тип события:
  • ProcessingExpirationPending - скоро истечёт срок обработки
  • ProcessingExpired - срок обработки истёк
  • TransactionAnnulmentPending - скоро будет отменена транзакция
  • TransactionAnnulled - транзакция отменена
  • NegoExpirationPending - скоро истечёт срок согласования
  • NegoExpired - срок согласования истёк

Body

public static class Body implements IKafkaMessageBody {
    // статус заявки
    public ApplicationStatus status;
    // Список ид прототипов этапов согласования
    public List<Long> negotiationStageProtoIds;
    // Текущая группа этапов согласования
    public String negoStageGroup;
    // пользователи, которым могла быть отправлена нотификация, если они подключены по websocket
    public Set<String> notifiedLogins;
    // пользователи обработавшие заявку ранее
    public Set<String> processedLogins;
}

Topic: source-record

Уведомление об обработке сообщения импорта

Id - идентификатор сообщения имопрта

Headers

• system - идентификатор внешней системы
• class - идентификатор обменного класса

Body

public static class Body {
    // Статус: SESSION_ACTIVE, PROCESSING_PENDING, PROCESSING, PROCESSED, DECLINED
    public IngestionStatus status;
    // Идентификатор связанной заявки
    public Long applicationId;
    // Идентификатор во внешней системе
    public String sourceSystemKey;
    // Сессия импорта
    public Long importSession;
    // Обработано в транзакции
    public Long stagedInTransaction;
    // Гуид айтема с sourceSystemKey
    public String itemUid;
    // Ошибка обработки
    public String error;
}