Операторы rxjs собирают строки из источника и частично генерируют их в соответствии с шаблоном

Question

Так что я использую typcript / RXJS / React-Native в проекте с поддержкой Bluetooth;У меня есть функция, которая получает строки от данного периферийного устройства, но есть некоторые предостережения, которые я не могу избежать от этого периферийного устройства.

Во-первых, он связывался со мной через определенный набор команд;а именно, каждая команда имеет вид /[a-zA-Z][^;]*;/ (он же один алфавит, сопровождаемый любым количеством символов, заканчивающийся точкой с запятой).

За этими командами могут или не могут следовать пробелы, которые должныбыть проигнорированным.Кроме того, эти команды могут отправляться или не отправляться каскадно: a1234;bFGe4; будет двумя командами a1234; и bFGe4;, отправленными в одном сообщении.Однако есть предостережение: команда, если она достаточно длинная, может быть неполной.Например, я мог бы получить два сообщения c444a;X132124122412431 и затем 1234124;, которые должны быть переведены в две отдельные команды c444a; и X1321241224124311234124;.Это связано с аппаратным ограничением.

Мне удалось справиться с этой ситуацией с помощью наблюдаемой, которая хранит в себе окончание последней полученной строки:

const ANY_MSG = /[a-zA-Z][^;]*;/g

const messages$ = new Observable<string>(sub => {
  let previousMsg = "";

  // monitor messages is the function w/ a callback that receives messages from the device
  monitorMessages((err, msg) => {
    if (err) {
      return sub.error(err);
    }

    const currMsg = previousMsg + msg
    const matches = currMsg.match(ANY_MSG)

    let lastIndex = 0
    if(matches) {
      for(const match of matches) {
        sub.next(match)
        lastIndex += match.length
      }
    }
    previousMsg = currMsg.slice(lastIndex)
  });
});

Эта наблюдаемая испускает мои сообщениякак я и ожидал: они разделяются точкой с запятой и объединяются, если они были частично получены функцией monitorMessages.

Дело в том, что мне кажется, что эту функцию трудно читать и понимать, и было бы лучшевыкл, если он был составлен с помощью pipe d функций из RXJS.Другими словами, я хочу сделать что-то вроде этого:

const messages$ = bindNodeCallback(monitorMessages).pipe(
  // ??????
);

Но я не могу понять, какие операторы мне нужно применять там (или даже если мне нужно написать свои собственные), вЧтобы произошло следующее:

• Всякий раз, когда monitorMessages отправляет сообщение, разбейте его на две части: часть до последней точки с запятой и часть после последней точки с запятой сообщения
• Извлеките все разделенные точкой с запятой команды из первой части и сохраните вторую часть
• Всякий раз, когда monitorMessages выдает снова, объедините сохраненную вторую часть с новой строкой и заставьте ее пройти тот же процесс, что и раньше

Возможно ли это даже через операторов RXJS?Может быть, мне нужно создать новую наблюдаемую, которая работает как «хранилище» для второй части?Я просто чувствую, что нынешний подход (создание какого-то внутреннего состояния для наблюдаемого) очень странный и относительно сложный для понимания.

Answer 1

Я думаю, что ваша реализация в порядке.Однако, если вы действительно хотите реализовать это с помощью операторов, вы можете использовать scan для поддержания некоторого состояния в составленной наблюдаемой цепочке, например:

const messages$ = bindNodeCallback(monitorMessages).pipe(
  scan((acc, received) => {
    const data = acc.remainder + received;
    const messages = data.match(ANY_MSG);
    if (messages) {
      const length = messsages.reduce((total, message) => total + message.length, 0);
      return { messages, remainder: data.slice(length) };
    }
    return { messages: [], remainder: data };
  }, { messages: [], remainder: "" }),
  mergeMap(({ messages }) => messages)
);

И для генерации вашегомассив сообщений можно использовать mergeMap, возвращая массив - в качестве массива можно указать ObservableInput.