Разработка базовой библиотеки шины ION AIB, код MODAIB-7

[akumidv]

В рамках данной задачи нужно реализовать:

1. Компонент шины AIB, представляющий собой обьект с состоянием предоставляющий следующий интерфейс программного взаимодействия:

class Bus {
  handle(Signal signal)
  wait(code, sender, thread): Promise
  register(Agent agent)
  eject(id)
  status(id)
  setStatus(id, data)
  serve()
  stop()
}

Описание функционала методов:

handle(Signal signal) - обработка входящего сигнала. Объект сигнала помещается в очередь для дальнейшей обработки. Очередь сообщений представлена внутренним компонентом реализующим интерфейс взаимодействия Queue:

class Queue {
  enqueue(Signal signal)
  dequeue(): Signal
}

Данный компонент должен помещаться в компонент шины посредством подсистемы внедрения зависимостей, и реализовываться слой абстракции представляющий очередь сигналов. В рамках данной задачи требуется реализовать простой компонент очереди сообщений, хранящий очередь в БД посредством компонента DataSource.

Спецификация объекта класса Signal:
class Signal {
code: string
sender: string
thread: string
data: mixed
}

Где code - уникальный идентификатор типа сигнала, sender - уникальный идентификатор отправителя сигнала (может отсутствовать), thread - уникальный идентификатор потока обработки сигнала, который присваивается шиной в методе handle если не указан явно, data - данные произвольного формата, определяемого типом сигнала.
Таким образом метод handle просто помещает объект типа Signal в очередь сигналов, присваивая ему идентификатор потока обработки.

wait(code, sender, thread, timeout): Promise - ожидание сигнала. Создает и регистрирует во внутреннем хранилище ожидание события идентифицируемого переданными параметрами. Если метод вызван без параметров выполняется ожидание ЛЮБОГО следующего события. Также последним аргументом можно передать время ожидания сигнала (значение по умолчанию должно настраиваться на уровне всей шины), по истечении которого, если сигнал не был получен должно выбрасываться соответствующее исключение.

register(Agent agent, Signal trigger) - регистрирует на шине агента, который будет запускаться при поступлении сигнала описываемого аргументом trigger. Спецификация объекта Agent:

class Agent {
  id: string
  constructor: function
}

Где id - уникальный идентификатор агента, constructor - функция выполняющая инициализацию агента на шине. Эта функция будет вызвана при регистрации агента с передачей объекта шины в качестве единственного аргумента. Если результатом функции-конструктора является функция, то в случае если
передан параметр trigger, она становится обработчиком этого события на шине, если trigger не передан, то функция будет вызвана ОДИН раз при поступлении ЛЮБОГО следующего события (посредством wait).

eject(id) - отключает агента от шины, очищает его состояние, исключает его обработчик из цикла обработки сигнала.

setStatus(id, data) - устанавливает во внутреннем хранилище шины состояние агента идентифицируемого аргументом id. Состояние описывается аргументом data произвольного формата. Внутреннее хранилище шины - реализация интерфейса KeyValueRepository ядра ION, объект задаваемый через внедрение зависимостей.

status(id) - возвращает статус агента идентифицируемого аргументом id, если агент не зарегистрирован выбрасывает исключение.

serve() - запускает шину. метод выбирает сигнал из очереди сообщений и запускает его обработчики, после чего посредством process.setImmediate планирует свой следующий рекурсивный запуск.

stop() - останавливает шину. Устанавливает внутренний флаг прерывающий рекурсию метода serve()

Пример конструктора агента шины сообщений:

const AGENT_ID = 'cool-agent';
const extApi = {method: (data) => {return 'ololo'}}

(bus) => async ({code, sender, thread, data}) => {
  switch (signal.code) {
    case 'sig1':
      const sig2 = await bus.wait('sig2')
      bus.handle({code: 'sig3', sender: AGENT_ID, thread, data: data.a + sig2.data.b})
    break;
    case 'sig3':
      if (sender !== AGENT_ID) {
        await extApi.method(data)
      }
    break;
  }
}