Слияние новой наблюдаемой rxjs с существующей наблюдаемой наблюдаемой

Question

У меня есть наблюдаемая rxjs, на которую уже подписаны:

const o = new rxjs.Subject();
o.subscribe( (v) => console.log("value: " + v);

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

const aPromise = axios.get("someurl...")

Затем я делаю следующее:

const secondObservable = rxjs.of(aPromise)
                             .subscribe( (response) => o.next(response.data) )

У меня вопрос, есть ли лучший способ поместить данные ответов в существующие, уже подписанные на наблюдаемые, возможно, с использованием операторов?

Answer 1

Вы можете просто передать результат вызова непосредственно вашему наблюдаемому.

const axios = {
  get: url => Promise.resolve('some value')
};

const o = new rxjs.Subject();
o.subscribe(v => { console.log("value: " + v); });

axios.get('url').then(data => { o.next(data); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>

Answer 2

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

const btnClick$ = new rxjs.Subject();

onClick(){
  this.btnClick$.next(void 0);
}

Затем мы превратим этот поток btnClick$ в поток запросов http-get, используя switchMap:

const request$ = btnClick$.pipe(
  // each click will be turned into a http-get
  switchMap(() => rxjs.of(axios.get("someurl...")))
)

И мы готовы использовать результаты как потоков, так и исходных данных и потока возможных обновлений данных с сервера при нажатии кнопки:

// we'll take results both from original stream
// and our new possible source of data
merge(original$, request$).subscribe(data => {
  console.log(data);
})

Вот небольшое приложение, использующее насмешки:

const { fromEvent, of, merge } = rxjs;
const { map, delay, switchMap } = rxjs.operators;

// create a stream of btn clicks
const theButton = document.getElementById('the-btn');
const btnClicks$ = fromEvent(theButton, 'click');

console.log('Processing...');

// clicks turned into http-gets
const request$ = btnClicks$.pipe(
  switchMap(() => makeRequest())
);

// original source of data
const original$ = of('Please, click the button!').pipe(delay(500));

merge(original$, request$)
  .subscribe(data => {
    console.log(data);
  });

// mock for requests
function makeRequest(){
  return of(Date.now()).pipe(delay(500))
}

<script src="https://unpkg.com/rxjs@6.4.0/bundles/rxjs.umd.min.js"></script>

<button
  id="the-btn"
  >Click me!</button>

Надеюсь, это поможет