将 React useEffect 钩子与 rxjs mergeMap 运算符一起使用

Question

我正在尝试实现一个必须使用 inner observables 的数据流，其中我使用来自mergeMap、concatMap 等的一个。

例如：

const output$$ = input$$.pipe(
    mergeMap(str => of(str).pipe(delay(10))),
    share()
  );

  output$$.subscribe(console.log);

登录控制台时可以正常工作。 但是当我尝试在 React 中使用它时，如下所示，利用 useEffect 和 useState 挂钩来更新一些文本：

function App() {
  const input$ = new Subject<string>();
  const input$$ = input$.pipe(share());
  const output$$ = input$$.pipe(
    mergeMap(str => of(str).pipe(delay(10))),
    share()
  );

  output$$.subscribe(console.log);
  // This works

  const [input, setInput] = useState("");
  const [output, setOutput] = useState("");

  useEffect(() => {
    const subscription = input$$.subscribe(setInput);

    return () => {
      subscription.unsubscribe();
    };
  }, [input$$]);

  useEffect(() => {
    const subscription = output$$.subscribe(setOutput);
    // This doesn't

    return () => {
      subscription.unsubscribe();
    };
  }, [output$$]);

  return (
    <div className="App">
      <input
        onChange={event => input$.next(event.target.value)}
        value={input}
      />
      <p>{output}</p>
    </div>
  );
}

它开始表现得很奇怪/不可预测（例如：有时文本会在输入过程中更新，有时它根本不会更新）。

我注意到的事情：

• 如果内部 observable 立即完成/是一个承诺 立即解决，它工作正常。
• 如果我们打印到控制台而不是useEffect，它工作正常。

我认为这与useEffect 的内部工作原理以及它如何捕获和注意外部变化有关，但无法使其正常工作。
任何帮助深表感谢。

最小复制案例：
https://codesandbox.io/s/hooks-and-observables-1-7ygd8

Answer 1

我有一个类似的任务，但目标是管道和去抖动输入测试并执行 ajax 调用。 简单的答案是，您应该在反应钩子“useState”中使用箭头函数初始化 RxJS 主题，以便在每次初始化时初始化一次主题。

那么你应该使用带有空数组 [] 的效果，以便在组件初始化时创建一个管道。

import React, { useEffect, useState } from "react";
import { ajax } from "rxjs/ajax";
import { debounceTime, delay, takeUntil } from "rxjs/operators";
import { Subject } from "rxjs/internal/Subject";

const App = () => {
  const [items, setItems] = useState([]);
  const [loading, setLoading] = useState(true);
  const [filterChangedSubject] = useState(() => {
    // Arrow function is used to init Singleton Subject. (in a scope of a current component)
    return new Subject<string>();
  });

  useEffect(() => {
    // Effect that will be initialized once on a react component init. 
    // Define your pipe here.
    const subscription = filterChangedSubject
      .pipe(debounceTime(200))
      .subscribe((filter) => {
        if (!filter) {
          setLoading(false);
          setItems([]);
          return;
        }
        ajax(`https://swapi.dev/api/people?search=${filter}`)
          .pipe(
            // current running ajax is canceled on filter change.
            takeUntil(filterChangedSubject)
          )
          .subscribe(
            (results) => {
              // Set items will cause render:
              setItems(results.response.results);
            },
            () => {
              setLoading(false);
            },
            () => {
              setLoading(false);
            }
          );
      });

    return () => {
      // On Component destroy. notify takeUntil to unsubscribe from current running ajax request
      filterChangedSubject.next("");
      // unsubscribe filter change listener
      subscription.unsubscribe();
    };
  }, []);

  const onFilterChange = (e) => {
    // Notify subject about the filter change
    filterChangedSubject.next(e.target.value);
  };
  return (
    <div>
      Cards
      {loading && <div>Loading...</div>}
      <input onChange={onFilterChange}></input>
      {items && items.map((item, index) => <div key={index}>{item.name}</div>)}
    </div>
  );
};

export default App;

Answer 2

我不太确定您想要实现什么，但我发现了一些问题，希望以下代码可以解决：

function App() {
    // Create these observables only once.
    const [input$] = useState(() => new Subject<string>());
    const [input$$] = useState(() => input$.pipe(share()));
    const [output$$] = useState(() => input$$.pipe(
        mergeMap(str => of(str).pipe(delay(10))),
        share()
    ));

    const [input, setInput] = useState("");
    const [output, setOutput] = useState("");

    // Create the subscription to input$$ on component mount, not on every render.
    useEffect(() => {
        const subscription = input$$.subscribe(setInput);

        return () => {
            subscription.unsubscribe();
        };
    }, []);

    // Create the subscription to output$$ on component mount, not on every render.
    useEffect(() => {
        const subscription = output$$.subscribe(setOutput);

        return () => {
            subscription.unsubscribe();
        };
    }, []);

    return (
        <div className="App">
            <input
                onChange={event => input$.next(event.target.value)}
                value={input}
            />
            <p>{output}</p>
        </div>
    );
}