将新的 rxjs 可观察量合并到现有的订阅可观察量中

Question

我有一个已经订阅的 rxjs observable：

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 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>

希望这有帮助

Answer 2

您可以将调用的结果直接传递给可观察量。

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>