Use change-emitter to keep track of observers (#160)

* Use change-emitter to keep track of observers

* Add test for multiple observers

src/packages/rx-recompose/__tests__/mapPropsStream-test.js

@@ -117,3 +117,22 @@ test('mapPropsStream renders null until stream of props emits value', t => {
   props$.onNext({ foo: 'bar' })
   t.is(wrapper.find('div').prop('foo'), 'bar')
 })
+
+
+test('handler multiple observers of props stream', t => {
+  const Container = mapPropsStream(props$ =>
+    // Adds three observers to props stream
+    props$.combineLatest(
+      props$, props$,
+      props1 => props1
+    )
+  )('div')
+
+  const wrapper = mount(<Container value={1} />)
+  const div = wrapper.find('div')
+
+  t.is(div.prop('value'), 1)
+  // Push onto props stream
+  wrapper.setProps({ value: 2 })
+  t.is(div.prop('value'), 2)
+})

src/packages/rx-recompose/createComponent.js

@@ -1,15 +1,17 @@
 import { Component } from 'react'
 import { Observable } from 'rx'
+import { createChangeEmitter } from 'change-emitter'
 
 const createComponent = propsToVdom =>
   class RxComponent extends Component {
     state = {};
 
+    propsEmitter = createChangeEmitter();
+
     // Stream of props
-    props$ = Observable.create(observer => {
-      this.propsObserver = observer
-      this.propsObserver.onNext(this.props)
-    });
+    props$ = Observable.create(observer =>
+      this.propsEmitter.listen(props => observer.onNext(props))
+    );
 
     // Stream of vdom
     vdom$ = propsToVdom(this.props$);
@@ -29,6 +31,8 @@ const createComponent = propsToVdom =>
             : this.setState({ vdom })
         }
       )
+
+      this.propsEmitter.emit(this.props)
     }
 
     componentDidMount() {
@@ -37,7 +41,7 @@ const createComponent = propsToVdom =>
 
     componentWillReceiveProps(nextProps) {
       // Receive new props from the owner
-      this.propsObserver.onNext(nextProps)
+      this.propsEmitter.emit(nextProps)
     }
 
     shouldComponentUpdate(nextProps, nextState) {

src/packages/rx-recompose/createEventHandler.js

@@ -1,16 +1,13 @@
 import { Observable } from 'rx'
+import { createChangeEmitter } from 'change-emitter'
 
 const createEventHandler = () => {
-  const observers = []
-  const stream = Observable.create(observer => {
-    observers.push(observer)
-    return () => {
-      const i = observers.indexOf(observer)
-      observers.splice(i, 1)
-    }
-  })
+  const emitter = createChangeEmitter()
+  const stream = Observable.create(observer =>
+    emitter.listen(value => observer.onNext(value))
+  )
   return {
-    handler: value => observers.forEach(observer => observer.onNext(value)),
+    handler: emitter.emit,
     stream
   }
 }

src/packages/rx-recompose/package.json

@@ -15,6 +15,7 @@
     "composition"
   ],
   "dependencies": {
+    "change-emitter": "^0.1.2",
     "lodash": "^4.0.0"
   },
   "peerDependencies": {