Using React useEffect hook with rxjs mergeMap operator

Question

I'm trying to implement a data stream that has to use inner observables, where I use one from mergeMap, concatMap etc.

e.g.:

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

  output$$.subscribe(console.log);

This works fine when logging into console. But when I try to use it in React like below utilizing useEffect and useState hooks to update some text:

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>
  );
}

it starts acting weird/unpredictable (e.g.: sometimes the text is updated in the middle of typing, sometimes it doesn't update at all).

Things I have noticed:

• If the inner observable completes immediately/is a promise that resolves immediately, it works fine.
• If we print to console instead of useEffect, it works fine.

I believe this has to do something with the inner workings of useEffect and how it captures and notices outside changes, but cannot get it working.
Any help is much appreciated.

Minimal reproduction of the case:
https://codesandbox.io/s/hooks-and-observables-1-7ygd8

Answer 1

I'm not quite sure what you're trying to achieve, but I found a number of problems which hopefully the following code fixes:

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>
    );
}