Feature/initial state fom observable js

src/App.js

@@ -1,7 +1,15 @@
 import React from 'react';
 import axios from 'axios';
-import { BehaviorSubject, combineLatest, timer } from 'rxjs';
-import { flatMap, map, debounce, filter } from 'rxjs/operators';
+import { BehaviorSubject, combineLatest } from 'rxjs';
+import {
+  map,
+  filter,
+  startWith,
+  debounceTime,
+  distinctUntilChanged,
+  switchMap,
+  catchError,
+} from 'rxjs/operators';
 
 import withObservableStream from './withObservableStream';
 
@@ -59,35 +67,34 @@ const query$ = new BehaviorSubject('react');
 const subject$ = new BehaviorSubject(SUBJECT.POPULARITY);
 
 const queryForFetch$ = query$.pipe(
-  debounce(() => timer(1000)),
-  filter(query => query !== ''),
+  debounceTime(1000),
+  distinctUntilChanged(),
+  filter(Boolean),
 );
 
 const fetch$ = combineLatest(subject$, queryForFetch$).pipe(
-  flatMap(([subject, query]) =>
-    axios(`http://hn.algolia.com/api/v1/${subject}?query=${query}`),
+  switchMap(
+    // discard previous requests. Response order not granted.
+    ([subject, query]) =>
+      axios(`http://hn.algolia.com/api/v1/${subject}?query=${query}`),
   ),
   map(result => result.data.hits),
+  startWith([]),
+  catchError(() => []),
 );
 
-export default withObservableStream(
-  combineLatest(
-    subject$,
-    query$,
-    fetch$,
-    (subject, query, stories) => ({
-      subject,
-      query,
-      stories,
-    }),
-  ),
-  {
-    onSelectSubject: subject => subject$.next(subject),
-    onChangeQuery: value => query$.next(value),
-  },
-  {
-    query: 'react',
-    subject: SUBJECT.POPULARITY,
-    stories: [],
-  },
-)(App);
+const state$ = combineLatest(
+  subject$,
+  query$,
+  fetch$,
+  (subject, query, stories) => ({
+    subject,
+    query,
+    stories,
+  }),
+);
+
+export default withObservableStream(state$, {
+  onSelectSubject: subject => subject$.next(subject),
+  onChangeQuery: value => query$.next(value),
+})(App);

src/withObservableStream.js

@@ -1,19 +1,26 @@
-import React from 'react';
+import React, { Component } from 'react';
+import { skip, first, shareReplay } from 'rxjs/operators';
 
-export default (observable, triggers, initialState) => Component => {
-  return class extends React.Component {
+export default (observable, triggers) => InnerComponent => {
+  class Decorated extends Component {
     constructor(props) {
       super(props);
+      this.sharedObservable = observable.pipe(shareReplay(1));
 
-      this.state = {
-        ...initialState,
-      };
+      const initializationSubscription = this.sharedObservable
+        .pipe(first())
+        .subscribe(initialState => {
+          this.state = initialState;
+        });
+      initializationSubscription.unsubscribe();
     }
 
     componentDidMount() {
-      this.subscription = observable.subscribe(newState =>
-        this.setState({ ...newState }),
-      );
+      this.subscription = this.sharedObservable
+        .pipe(skip(1))
+        .subscribe(newState => {
+          this.setState(newState);
+        });
     }
 
     componentWillUnmount() {
@@ -22,8 +29,13 @@ export default (observable, triggers, initialState) => Component => {
 
     render() {
       return (
-        <Component {...this.props} {...this.state} {...triggers} />
+        <InnerComponent {...this.props} {...this.state} {...triggers}/>
       );
     }
-  };
+  }
+
+  Decorated.displayName = `withObservableStream(${InnerComponent.displayName ||
+    InnerComponent.name})`;
+
+  return Decorated;
 };