Телеграф на RxJS / Хабр

На днях я смотрел кино, где оператор использовал телеграф. Он знал наизусть азбуку Морзе и очень быстро нажимал свою единственную кнопку. Я задумался: с RxJS мы способны на большее! Давайте запилим телеграф, используя единственный fromEvent и массу интересных трюков. Потренируемся с Dependency Injection, директивами и операторами RxJS, чтобы собрать демо, которое выглядит круто и звучит аутентично.

Азбука Морзе

Телеграф работает с азбукой Морзе. Для нас будет несложно в ней разобраться, ведь это просто бинарная последовательность. Сигнал дискретный: он разбит на куски определенной длительности. Каждый кусок — либо писк, либо тишина. Можно использовать Boolean или просто 1 и 0 для краткости. Азбука следует правилам:

есть два символа: точка и тире;
тире — три единицы подряд;
точка — одна единица;
символы разбиты однократным нулем;
буквы разбиты тремя нулями подряд;
слова разбиты семью нулями подряд. Примем, что пробел — четыре нуля и три нуля на разделение букв.

Каждая буква — последовательность из таблицы ниже. Мы видим шаг сетки сигнала. Черный квадрат равен единице, белый — нулю:

Возьмем букву «H», это четыре точки. Точка — одна единица, и точки разбиты однократным нулем: 1010101. Фраза HELLO WORLD будет выглядеть так:

10101010001000101110101000101110101000111011101110000000 = HELLO[конец слова]

10111011100011101110111000101110100010111010100011101010000000 = WORLD[конец слова]

Это стандартная таблица для США, но в мире есть и другие версии: в Angular это подсказывает нам, что тут стоит использовать InjectionToken. Благодаря второму параметру, в котором можно описать фабрику, мы добавим стандартную имплементацию. При необходимости проекты смогут ее подменить на свою таблицу:

export const MAP = new InjectionToken<Map<string, readonly (0 | 1)[]>>(
  'Morse code dictionary',
  {
    factory: () => new Map([
      [' ', [0, 0, 0, 0]],
      ['a', [1, 0, 1, 1, 1]],
      // ...
    ])
  },
)

Мы не хотим заставлять людей нажимать одну кнопку — пусть используют обычную клавиатуру, а мы закодируем буквы самостоятельно. После того как сигнал пройдет по воображаемой телеграфной линии, мы его расшифруем и выведем на экран. Начнем с одного fromEvent(document, 'keydown') и разветвим этот Observable по полной!

Кодирование

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

Пусть это тоже будет InjectionToken, чтобы его можно было менять.

Это значит, что буквы превратятся в последовательности, имеющие продолжительность. Чтобы сохранить порядок и не терять значения, мы воспользуемся оператором concatMap. Он входит в семейство Higher Order Observable операторов, с его помощью мы превратим каждое значение в Observable. Когда поступает новое значение (пользователь нажал клавишу), concatMap дождется, пока прошлый Observable закомплитится. Последовательность полностью отыграет прежде, чем создать Observable новой последовательности. Так мы можем печатать очень быстро и не бояться, что какие-то буквы пропадут при передаче.

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

@Injectable({
  providedIn: 'root',
})
export class MorseService extends Subject<readonly (0 | 1)[]> {
  constructor(
    @Inject(MAP) private readonly chars: Map<string, readonly (0 | 1)[]>,
  ) {
    super();
  }

  send(char: string) {
    this.next(this.chars.get(char));
  }
}

Теперь можно сделать токен, превращающий буквы в азбуку Морзе. Для начала запросим таблицу и длительность шага сигнала. Затем преобразуем все keydown-события на document в последовательности из таблицы. Объединим их с сервисом из примера выше и с помощью concatMap переведем все в поток нулей и единиц. Каждую последовательность завершим тремя нулями:

export const MORSE = new InjectionToken<Observable<0 | 1>(
  'A sequence of Morse code', 
  {
    factory: () => {
      const chars = inject(MAP);
      const duration = inject(UNIT);
      const service$ = inject(MorseService);
      const keydown$ = fromEvent(inject(DOCUMENT), 'keydown').pipe(
        map(({ key }: KeyboardEvent) => chars.get(key)),
        filter(Boolean),
      );
	
      return merge(service$, keydown$).pipe(
        concatMap(sequence =>
          from(sequence).pipe(
            endWith(...SPACE),
            delayEach(duration),
          )
        ),
        share(),
      );
    }
  }
);

Обратите внимание, мы используем from, а не of. При работе с массивами from превращает их в последовательность эмитов, в то время как of сделает единственный эмит всего массива.

В коде выше написан кастомный оператор delayEach. Он позволяет задержать каждый эмит, а не весь стрим:

export function delayEach<T>(duration: number): MonoTypeOperatorFunction<T> {
  return concatMap(x => of(x).pipe(delay(duration)));
}

Теперь мы можем получить закодированный поток, взяв токен MORSE и подписавшись на него.

Расшифровка

Этот шаг интереснее. Нам нужен оператор для сравнения данных с последовательностью. Поток надо сбросить, если значение не равно соответствующему по индексу значению в последовательности. А если вся последовательность совпала и затем пролетело три нуля — надо испустить расшифрованную букву. Затем все это надо повторить.

Прерываем и перезапускаем стрим, чтобы индексы эмитов совпадали с индексами в последовательности.

Мы хотим визуализировать все, так что давайте создадим специальный модуль для букв. Он будет состоять из компонента, директивы и сервиса. Зачем нам это? Для хорошего разделения логики. Нужно знать, с какой буквой мы работаем. Но это понадобится и в компоненте, и в сервисе. Чтобы избежать циклической зависимости, мы создадим директиву с единственной целью — знать букву, за которую все отвечают:

@Directive({
  selector: '[char]',
})
export class CharDirective {
  @Input() char = '';
}

В компоненте используем тот же селектор. Компонент будет передавать сигналы с помощью мыши и показывать прогресс расшифровки:

@Component({
  selector: '[char]',
  templateUrl: 'char.template.html',
  styleUrls: ['char.style.less'],
  providers: [CharService],
})
export class CharComponent {
  constructor(
    @Inject(CharService) readonly service: Observable<number | string>,
    @Inject(CharDirective) readonly directive: CharDirective,
    @Inject(MorseService) private readonly emitter: MorseService,
  ) {}

  @HostListener('click')
  onClick() {
    this.emitter.send(this.directive.char);
  }
}

Теперь нам нужно написать сам алгоритм расшифровки. Мы неслучайно указали у сервиса тип Observable<number | string>. Он будет репортить прогресс от 0 до 1 и букву, если расшифровка прошла успешно. Числовое значение позволит визуализировать работу в демо-приложении.

Сервис

Создание сервиса — самая сложная часть. Мне потребовалось немало времени, чтобы решить эту RxJS-головоломку. Труднее всего было обыграть пробел. Поначалу он неотличим от завершения буквы: и то и другое — это куча нулей подряд.

Начнем с создания пары функций-помощников. Наш сервис наследуется от Observable. Внутри мы сделаем приватный стрим, обрабатывающий поток азбуки Морзе, созданный ранее:

@Injectable()
export class CharService extends Observable<number | string> {
  private readonly inner$ = this.morse$.pipe(
    // ...
  )

  constructor(
    @Inject(MORSE) private readonly morse$: Observable<0 | 1>,
    @Inject(MAP) private readonly chars: Map<string, readonly (0 | 1)[]>,
    @Inject(CharDirective) private readonly directive: CharDirective,
  ) {
    super(subscriber => this.inner$.subscribe(subscriber));
  }
}

Нужно знать, с какой последовательностью мы работаем. Придется использовать геттер, а не read-only проперти, потому что на этапе создания инпуты директивы еще не обработаны:

private get sequence(): readonly (0 | 1)[] {
  return [...this.chars.get(this.directive.char), ...SPACE];
}

Если бы эта строка была статичной, можно было бы использовать @Attribute вместо @Input — и значение стало бы доступным в конструкторе. Но тогда не получится создавать директивы через *ngFor, динамически выставляя букву.

Нам совсем не хочется пересобирать массив на каждый вызов геттера. Для этих целей воспользуемся декоратором @tuiPure из Taiga UI. Мы часто пользуемся им в своей библиотеке для реализации паттерна ленивых геттеров. Это техника мемоизации для отложенных вычислений. После того, как значение будет посчитано первый раз, декоратор подменит геттер на простое поле класса с результатом.

Этот декоратор работает и с методами. Он будет проверять входные аргументы и возвращать прошлое посчитанное значение при полном их совпадении. Что-то вроде чистых пайпов в шаблонах.

Дальше нужно проверять, что значение с определенным индексом равно соответствующему значению в последовательности:

private isValid(value: number, index: number): boolean {
  return this.sequence[index] === value;
}

Посчитаем значение, которое сервис будет испускать. Напомню, это либо прогресс, либо полученная буква:

private getValue(index: number): number | string {
  return this.sequence.length === index + 1 
    ? this.char 
    : (index + 2) / this.sequence.length;
}

Соберем внутренний поток с помощью всех этих методов:

private readonly inner$ = this.morse$.pipe(
  takeWhile(this.isValid.bind(this)),
  map((_, index) => this.getValue(index)),
  startWith(0),
  endWith(0),
  takeWhile(isNumber, true),
)

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

Второй аргумент takeWhile позволяет пропустить значение, нарушившее условие.

Перезапуск стрима

С помощью завершения и перезапуска можно легко проверять значения на совпадение с последовательностью, потому что индексы остаются синхронизированными. Обрыв по второму takeWhile (буква полностью расшифрована) можно просто перезапустить. А вот при несовпадении значений перед перезапуском нужно дождаться завершения буквы (три последовательных нуля).

Первой мыслью было использовать repeatWhen. Но проблема в том, что функция-фабрика будет вызвана только один раз после первого завершения. Дальше поток продолжит слушать возвращенный Observable:

repeatWhen(() => threeZeroes$),

Это значит, что repeatWhen будет правильно работать только во втором и последующих перезапусках. При первом завершении три нуля, означающие конец буквы, могут уже пройти к моменту, когда мы начинаем слушать threeZeroes$.

Так что для начала давайте просто слушать три последовательных нуля. Это отличный кейс для оператора scan:

export function consecutive<T>(
  value: T,
  amount: number
): OperatorFunction<T, unknown> {
  return pipe(
    startWith(...new Array(amount).fill(value)),
    scan((result, item) => (item === value ? ++result : 0), 0),
    filter(v => v >= amount),
  );
}

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

И нужно разделить два takeWhile-оператора. Мы хотим, чтобы первый обрывал внутренний стрим. Это значит, нам снова понадобится concatMap:

private readonly inner$ = this.morse$.pipe(
  consecutive(0, SPACE.length),
  concatMapTo(this.morse$.pipe(
    takeWhile(this.isValid.bind(this)),
    map((_, index) => this.getValue(index)),
    startWith(0),
    endWith(0),
  )),
  takeWhile(isNumber, true),
  repeat(),
)

Теперь, когда внутренний стрим оборвется, три последовательных нуля его перезапустят. Символ пробела прострелит несколько раз, но concatMap все равно будет дожидаться завершения внутреннего стрима, и второй takeWhile сбросит этот процесс.

Демо

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

Телеграф и азбука Морзе будут неполноценными без звука. Простую пищалку очень легко сделать с помощью Web Audio API и OscillatorNode. Воспользуемся моей библиотекой @ng-web-apis/audio.

Она является частью open-source-инициативы Web APIs for Angular. Наша цель — создание качественных легких оберток для нативных Web API, чтобы их было комфортно использовать в Angular. Посмотрите, что у нас уже есть.

С ее помощью мы можем превратить декларативные директивы в граф аудионодов:

<ng-container waOscillatorNode autoplay frequency="523">
  <ng-container
    waGainNode
    gain="0"
    [gain]="morse$ | async | waAudioParam : 0.02"
  >
    <ng-container waAudioDestinationNode></ng-container>
  </ng-container>
</ng-container>

Этот код будет проигрывать писк для единиц и замолкать на нулях благодаря регулированию громкости. Теперь давайте выведем клавиатуру и расшифрованный результат. Кнопка очистки тоже не повредит:

<section>
  <button *ngFor="let char of chars" type="button" [char]="char">
    {{char}}
  </button>
</section>
<output>{{ output$ | async }}</output>
<footer>
  <button type="reset" (click)="reset$.next()">Clear</button>
</footer>

Но где же взять output$? Мы можем собрать сервисы прямо из шаблона. Вы знали, что запросы Angular позволяют получать сервисы и другие DI сущности из инжекторов?

@ViewChildren(CharService)
readonly services?: QueryList<Observable<string | number[]>>;

Главное — помнить, что они недоступны до ngAfterViewInit-хука. Соберем output$:

readonly reset$ = new Subject<void>();

readonly output$ = this.reset$.pipe(
  switchMap(() => merge(...this.services.toArray()).pipe(
    filter(x => !isNumber(x)),
    scan((result, char) => result + char, ''),
    startWith(''),
  )),
  startWith(''),
);

ngAfterViewInit() {
  this.reset$.next();
}

Чтобы показывать прогресс расшифровки, добавим простой span в CharComponent и будем управлять его высотой и цветом с помощью двух стримов:

readonly progress$ = this.service.pipe(
  filter(isNumber),
  map(value => value * 100),
);

readonly pending$ = this.service.pipe(
  filter(Boolean),
  map(isNumber),
);

И последний трюк, который я хотел показать, — управление длительностью CSS-переходов с помощью токена на длительность шага сигнала. После перехода на Ivy в Angular можно использовать байндинг для задания значений CSS-переменных. Давайте заведем --duration и зададим ему значение в нашем основном компоненте:

@Component({
  // ...
  host: {
    '[style.--tui-duration.ms]': 'unit',
  },
})
export class AppComponent implements AfterViewInit {
  constructor(
    // ...
    @Inject(UNIT) readonly unit: number,
  ) {}

  // ...
}

Все готово, и я приглашаю вас попробовать итоговый продукт!

Телеграф на RxJS