Обеспечение использования в функции только синхронных каналов (запрет асинхронных каналов)

Question

Я работаю над приложением Angular, в котором данные должны извлекаться из различных источников (BehaviorSubjects с функциями отображения каналов) синхронно.

До сих пор это делалось так:

let magicNumber;
this.magicService.magicNumber$.pipe(take(1)).subscribe(val => magicNumber = val));

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

export function select<T>(inStream: Observable<T>): T {
  let value: T;

  race(
    inStream,
    throwError(
      new Error(
        select.name +
          " expects a synchronous stream. Received an asynchronous stream."
      )
    )
  )
    .pipe(take(1))
    .subscribe(val => (value = val));

  return value;
}

, что позволяет мне в службахпросто написать:

const magicNumber = select(this.magicService.magicNumber$);

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

Проблема теперь в том, что разработчики не заметят, что они сделали что-то не так, пока не запустят код, а затем код завершится сбоем из-за передачи асинхронного потока.

ДляНапример, скажем, у меня есть три службы: userService, basketService, purchaseService.

В userService У меня есть следующее:

userData$: BehaviorSubject<UserDataStore>;
...
userName$ = userData$.pipe(
  map(({ userName }) => userName));

аналогично в basketService:

basketStore$: BehaviorSubject<BasketStore>;
...
basketContent$ = basketStore$.pipe(
  map(({ content }) => content));

В purchaseService у меня есть метод http-post, который в этой реализации выглядит следующим образом:

function purchase() {
  const userName = select(this.userService.userName$);
  const basketContent = select(this.basketService.basketContent$);
  this.http.post("http://niceUrl.com", {
    userName,
    basketContent
  }).subscribe(
    ReduceResponse
  )

(В этом примере каналы упрощены; на практике онисодержат код повтора, перехват ошибок и т. д.)

Другими словами, я говорю о процедуре сбора данных из сервисов, уже хранящихся в BehaviorSubject где я знаю, что сбор будет синхронным , и для этого сделаем короткую руку.

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

для сравнения;этот пример выше до реализации метода select:

function purchase() {
  let userName, basketContent;
  this.userService.userName$.pipe(take(1)).subscribe(val => userName = val);
  this.basketService.basketContent$.pipe(take(1)).subscribe(val => basketContent = val);
  this.http.post("http://niceUrl.com", {
    userName,
    basketContent
  }).subscribe(
    ReduceResponse
  )

PS! Мы уже практикуем строго асинхронное обновление данных с помощью OnPush в коде компонента, поэтому комментарии об использовании RxJ в асинхронном режиме в компонентах не нужны