can someone help me how to bind React State to RxJS Observable? I did sth like

ponentDidMount() {
  let source = Rx.Observable.of(this.state.val)
}

The ideal result is, whenever this.state.val updated (via this.setState(...)) source get updated too, so I can bine source with other RxJS observable stream.

However, in this case, source only get updated once, even after this.state.val is updated and ponent is re-rendered.

// Ideal result:
this.state.val = 1
source.subscribe(val => console.log(x)) //=> 1
this.state.val = 2
source.subscribe(val => console.log(val)) //=> 2

// Real result:
this.state.val = 1
source.subscribe(val => console.log(x)) //=> 1
this.state.val = 2
source.subscribe(val => console.log(val)) //=> 1 ???WTH

It might be because ponentDidMount() only invoked once in React lifetime. so I move source to ponentDidUpdate() which is invoked everytime after ponent is rendered. However, the result still remain the same.

So the question is how to make source updated whenever this.state.val updated?

Updated: Here is the solution I used to solve the prob, using Rx.Subject

// Component file
constructor() {
  super(props)
  this.source = new Rx.Subject()
_onChangeHandler(e) {
 this.source.onNext(e.target.value)
}
ponentDidMount() {
  this.source.subscribe(x => console.log(x)) // x is updated
}
render() {
  <input type='text' onChange={this._onChangeHandler} />
}
// 

can someone help me how to bind React State to RxJS Observable? I did sth like

ponentDidMount() {
  let source = Rx.Observable.of(this.state.val)
}

The ideal result is, whenever this.state.val updated (via this.setState(...)) source get updated too, so I can bine source with other RxJS observable stream.

However, in this case, source only get updated once, even after this.state.val is updated and ponent is re-rendered.

// Ideal result:
this.state.val = 1
source.subscribe(val => console.log(x)) //=> 1
this.state.val = 2
source.subscribe(val => console.log(val)) //=> 2

// Real result:
this.state.val = 1
source.subscribe(val => console.log(x)) //=> 1
this.state.val = 2
source.subscribe(val => console.log(val)) //=> 1 ???WTH

It might be because ponentDidMount() only invoked once in React lifetime. so I move source to ponentDidUpdate() which is invoked everytime after ponent is rendered. However, the result still remain the same.

So the question is how to make source updated whenever this.state.val updated?

Updated: Here is the solution I used to solve the prob, using Rx.Subject

// Component file
constructor() {
  super(props)
  this.source = new Rx.Subject()
_onChangeHandler(e) {
 this.source.onNext(e.target.value)
}
ponentDidMount() {
  this.source.subscribe(x => console.log(x)) // x is updated
}
render() {
  <input type='text' onChange={this._onChangeHandler} />
}
// 

• javascript
• reactjs
• rxjs

5 Answers 5

Update

To abstract out some of the below plexity, use repose's mapPropsStream or ponentFromStream. E.g.

const WithMouseMove = mapPropsStream((props$) => {
  const { handler: mouseMove, stream: mouseMove$ } = createEventHandler();

  const mousePosition$ = mouseMove$
    .startWith({ x: 0, y: 0 })
    .throttleTime(200)
    .map(e => ({ x: e.clientX, y: e.clientY }));

  return props$
    .map(props => ({ ...props, mouseMove }))
    .bineLatest(mousePosition$, (props, mousePosition) => ({ ...props, ...mousePosition }));
});

const DumbComponent = ({ x, y, mouseMove }) => (
  <div
    onMouseMove={mouseMove}
  >
    <span>{x}, {y}</span>
  </div>
);

const DumbComponentWithMouseMove = WithMouseMove(DumbComponent);

Original Post

For a slightly updated answer to the OP's updated answer, using rxjs5, I came up with the following:

class SomeComponent extends React.Component {
  constructor(props) {
    super(props);

    this.mouseMove$ = new Rx.Subject();
    this.mouseMove$.next = this.mouseMove$.next.bind(this.mouseMove$);

    this.mouseMove$
      .throttleTime(1000)
      .subscribe(idx => {
        console.log('throttled mouse move');
      });
  }

  ponentWillUnmount() {
    this.mouseMove$.unsubscribe();
  }

  render() {
    return (
      <div
       onMouseMove={this.mouseMove$.next}
      />
    );
  }
}

Some notable additions:

• onNext() is now next()
• binding the observable next method allows it to be passed directly to the mouseMove handler
• streams should be unsubscribed in ponentWillUnmount hook

Furthermore, the subject streams initialized in the ponent constructor hook can be passed as properties to 1+ child ponent(s), which could all push to the stream using any of the observable next/error/plete methods. Here's a jsbin example I put together demonstrating multiple event streams shared between multiple ponents.

Curious if anyone has ideas on how to better encapsulate this logic to simplify stuff like binding and unsubscribing.

One option could be to use Rx.Observable.ofObjectChanges > cf. https://github./Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/ofobjectchanges.md.

However :

• It uses Object.observe which is not a standard feature, hence will have to be polyfilled in some browsers and is actually being removed from inclusion in ecmascript (cf. http://www.infoq./news/2015/11/object-observe-withdrawn). Not the choice for the future, but it is easy to use, so if it is just for your own needs, why not.

Other option is to use a subject in one of the three methods at your disposal according to your use case : shouldComponentUpdate, ponentWillUpdate, ponentDidUpdate. Cf. https://facebook.github.io/react/docs/ponent-specs.html for when each function is executed. In one of these methods, you would check if this.state.val has changed, and emits its new value on the subject if it did.

I am not a reactjs specialist, so I guess they might be other options.

Although a subject will work, I think the best practice is to avoid using a subject when you can use an observable. In this case you can use Observable.fromEvent:

class MouseOverComponent extends React.Component {

  ponentDidMount() {
    this.mouseMove$ = Rx.Observable
      .fromEvent(this.mouseDiv, "mousemove")
      .throttleTime(1000)
      .subscribe(() => console.log("throttled mouse move"));

  }

  ponentWillUnmount() {
    this.mouseMove$.unsubscribe();
  }

  render() {
    return (
      <div ref={(ref) => this.mouseDiv = ref}>
          Move the mouse...
      </div>
    );
  }
}


ReactDOM.render(<MouseOverComponent />, document.getElementById('app'));

Here it is on codepen....

It seems to me that there are other times when a Subject like the best choice, like when a custom React ponent executes a function when an event occurs.

I would highly remend reading this blog post on streaming props to a React ponent using RxJS:

https://medium./@fahad19/using-rxjs-with-react-js-part-2-streaming-props-to-ponent-c7792bc1f40f

It uses FrintJS, and applies the observe higher-order ponent for returning the props as a stream:

import React from 'react';
import { Observable } from 'rxjs';
import { observe } from 'frint-react';

function MyComponent(props) {
  return <p>Interval: {props.interval}</p>;
}

export default observe(function () {
  // return an Observable emitting a props-patible object here
  return Observable.interval(1000)
    .map(x => ({ interval: x }));
})(MyComponent);

You can do it using hooks.

Here is a code sample

import { Observable, Subscription } from 'rxjs';
import { useState, useEffect } from 'react';

export default function useObservable<T = number | undefined>(
    observable: Observable<T | undefined>,
    initialState?: T): T | undefined {
    const [state, setState] = useState<T | undefined>(initialState);

    useEffect(() => {
        const subscription: Subscription = observable.subscribe(
            (next: T | undefined) => {
                setState(next);
            },
            error => console.log(error),
            () => setState(undefined));
        return () => subscription.unsubscribe();
    }, [observable])

    return state;
}