redux-observable / redux-observable

RxJS middleware for action side effects in Redux using "Epics"

RxJS-based middleware for Redux. Compose and cancel async actions to create side effects and more.

https://redux-observable.js.org

Note: this project is quite stable, but is currently in maintenance mode. Critical fixes will still be released, but for now no additional features or changes will be considered as the maintainer @jayphelps is not actively working on any apps that use redux-observable (or any UI apps at all, actually.) If you would like to become a maintainer, please reach out to @jayphelps.

redux-observable's Issues

Add ability to add new delegator

In an app which uses code splitting, reducers might be loaded lazily as the user navigates. It would nice to be able to add delegators to the root delegator after the app is run initially to support these kinds of applications.

Subscribe on Epic

Hi!

Since trunkservables are deprecated, how one can dispatch and subscribe?

i.e. in onEnter() of react-router I want to load some data asynchronously with redux-observable and then call callback to load the view (i.e. if user is authorized)

It's a bit unclear how to wait until Epic is completed (it was easy with trunkservables because they were returning an Observable)

Thanks!

Shouldn't publish the resources directory to npm

Right now we're publishing the images and everything in the resources directory to npm. That's unnecessary

Add error handling examples

Hi,

I think it's not clear right now how to handle errors with this middleware. There is no example of it in the basic example nor the medium article.

Lets say I have this action creator:

function createRequestAction({request, success, failure, abort}, fetch) {
    return payload => actions => Rx.Observable
            .fromPromise(fetch(payload))
            .map(success)
            .takeUntil(actions.ofType(abort.toString()))
            .startWith(request());
}

const getUser = createRequestAction(userActions, fetchUser);

// ...
store.dispatch(getUser({id: 123}));

If the promise returned by fetchUser fail, it will just throw an error and I don't see any way to handle it in a proper RxJS fashion. In RxJS v4 there was a catch method, but now there isn't. How should i properly dispatch the failure error when the request fail?

Effects as data

Do you want to request a feature or report a bug?
feature

What is the current behavior?
I really like the idea of background epics because it provides an intuitive way to handle multiple actions like debouncing with the powerful operators of Rx. However, tests for epics require mocking HTTP requests or other side effects.

To address the same problem, redux-saga allows sagas to yield effects as data, such as call, executes the effects, and feed the results to sagas. This keeps sagas side-effect-free and testable without mocking.

If the current behavior is a bug, please provide the steps to reproduce and if possible a minimal demo of the problem via https://jsbin.com or similar.
It's not a bug.

What is the expected behavior?

To employ the same approach to redux-observable, the straight forward way would be express effects as actions. We can have a dedicated middleware to receive side effect requests, execute them, and dispatches results as actions so that we can throw out side effects of epics.

import { call } from 'redux-observable/effects';

function epic(action$, store) {
  const fooReq$ = action$.ofType('FOO')
    .map(action => call('FOO_REQ', webapi.getFoo, action.payload.id));
  const foo$ = action$.ofType('FOO_REQ')
    .map(foo => ({ type: 'FOO_FETCHED', payload: foo }));

  return Observable.merge(
    fooReq$,
    foo$
  );
}

The example above should work but it's tedious to have pairs of request observable and response observable. I want to write the pair as a single Observable. A possible solution would be creating an operator that emits effect requests to a Subject and returns an Observable that emits effect responses. Let's call it as effect operator. Here I assumed that effect returns an Observable of Observables in order to express effect's result as an Observable.

import { call } from 'redux-observable/effects';

function epic(action$, store) {
  const effect$ = new Subject();

  const foo$ = action$.ofType('FOO')
    .effect(action$, effect$, action => call(webapi.getFoo, action.payload.id))
    .switch()
    .map(foo => ({ type: 'FOO_FETCHED', payload: foo }));

  return Observable.merge(
    effect$,
    foo$
  );
}

I wrote quick sketches for the effect middleware and operator.

const EFFECT_REQUEST = Symbol('EFFECT_REQUEST');
const EFFECT_RESPONSE = Symbol('EFFECT_RESPONSE');

const effectMiddleware = store => next => action => {
  if (typeof action[EFFECT_REQUEST] !== 'object') {
    return next(action);
  }

  const { id, payload } = action[EFFECT_REQUEST];
  const { type, args } = payload;
  switch (type) {
    case 'call':
      try {
        const returnValue = args[0].apply(null, args.slice(1));
        if (isObservable(returnValue)) {
          next({
            [EFFECT_RESPONSE]: { id, type, payload: returnValue }
          });
        } else if (typeof returnValue.then === 'function') {
          returnValue
            .then(result => next({
              [EFFECT_RESPONSE]: { id, type, payload: Observable.of(result) }
            }))
            .catch(error => next({
              [EFFECT_RESPONSE]: { id, type, payload: Observable.throw(error) }
            }));
        } else {
          next({
            [EFFECT_RESPONSE]: { id, type, payload: Observable.of(returnValue) }
          });
        }
      catch (error) {
        next({
          [EFFECT_RESPONSE]: { id, type, payload: Observable.throw(error) }
        });
      }
    default:
      // TODO: Handle other effect types.
      return next(action);
  }
};

function effectOperator(action$, effect$, callback) {
  return Observable.create(subscriber => {
    const id = uuid();
    const source = this;
    const responseSub = action$
      .filter(a => a[EFFECT_RESPONSE] && a[EFFECT_RESPONSE].id === id)
      .map(a => a[EFFECT_RESPONSE].payload)
      .subscribe(subscriber);
    const requestSub = source.subscribe(
      value => {
        try {
          effect$.next({
            [EFFECT_REQUEST]: { id, payload: callback(value) }
          });
        } catch (e) {
          subscriber.error(e);
        }
      },
      err => subscriber.error(err),
      () => subscriber.complete()
    );
    // https://github.com/ReactiveX/rxjs/issues/1583
    return new Subscription(() => {
      [responseSub, requestSub].forEach(s => s.unsubscribe());
    });
  });
}
Observable.prototype.effect = effectOperator;

What do you think about this idea?

There are still some things to consider:

• Awkwardness of passing action$ and effect$
• Awkwardness of explicitly creating a Subject
• Epic composition

Which versions of redux-observable, and which browser and OS are affected by this issue? Did this work in previous versions of redux-observable?
It's not a bug.

Add support for Observable emitting additional "thunkservables"?

Okay... so I just coined a term "thunkservable"... which is a function that returns an observable.

Anyhow, should we support something like this?

let secondAction = () => Observable.of({ type: 'ACTIONS_DONE' });
let firstAction = () => Observable.interval(1000).mapTo(secondAction);

dispatch(firstAction);

Add more recipes

Everyone,

Let's add a bunch of thoughts here about Recipes people would like to see added to the docs. If you can, please feel free to just go ahead and PR any you think of or that are on this list! 🚀

• How RxJS makes things like debouncing trivial--Prominently show this example in the README and other places cause it's such amaze.
• Router navigation (related)
• Logging analytics (like gaq, etc)
• Patterns with common libs like redux-form (related)
• Best practices/anti-patterns
• Polling things (example by @winkerVSbecks)
• WebSocket streaming (prolly using Observable.webSocket())
• forking short-lived Epics (example)
• Global error handler in root Epic example

Add support for ActionsObservable.of(...actions) etc

This would make it far easier to create them for tests and such:

const action$ = ActionsObservable.of({ type: FIRST }, { type: SECOND });

// instead of the current:

const input$ = Observable.of({ type: FIRST }, { type: SECOND });
const action$ = new ActionsObservable(input$);

I was waiting on ReactiveX/rxjs#1876 but it might be a while for that is resolved so we probably should provide our implementation in the mean time.

TypeError: action$.ofType(...).delay is not a function

Do you want to request a feature or report a bug?
Bug

What is the current behavior?
TypeError: action$.ofType(...).delay is not a function

whenever I pass an Epic to the middleware creator

If the current behavior is a bug, please provide the steps to reproduce and if possible a minimal demo of the problem via https://jsbin.com or similar.

Here's the configureStore

https://github.com/imranismail/test-redux-observable/blob/master/src/utils/configure-store.js

What is the expected behavior?

No error and middleware should be able to capture the actions

Which versions of redux-observable, and which browser and OS are affected by this issue? Did this work in previous versions of redux-observable?

Chrome 51
OSX
[email protected]

HELP: what do we name this new feature?

Edit: so far we've decided on "process manager" instead of "delegator". They're called "Epics".

Although undocumented at the moment, we've added the ability to compose and inject static action handlers directly into the middleware. That is, they are not dispatched, instead you dispatch redux action POJOs and react to those actions. This keeps your actions always pure objects and allows you to do debouncing and action composition easier in many ways.

Here are two examples, fetching a user by id (as a trivial example) and a fairly non-trivial search example that debounces value changes, cancels in-flight requests, server error response handling, and even supports pagination.

// FETCH USER

const fetchUser = id => ({ type: FETCH_USER, payload: id });

const fetchUserDelegator = actions =>
  actions.ofType(FETCH_USER)
    .switchMap(action =>
      Observable.ajax.getJSON(`/api/users/${action.payload}`)
        .map(payload => ({ type: FETCH_USER_FULFILLED, payload }))
        .startWith({ type: FETCH_USER_PENDING })
    );

// SEARCH

const setPage = value => ({ type: SET_PAGE, value });
const searchValueChanged = value => ({ type: SEARCH_VALUE_CHANGED, value });
const searchFulfilled = value => ({ type: SEARCH_RESULTS_FULFILLED, value });
const searchErrored = value => ({ type: SEARCH_RESULTS_ERRORED, value });
const searchPending = () => ({ type: SEARCH_RESULTS_PENDING });

const searchDelegator = (actions, { getState }) =>
  Observable.combineLatest(
    actions.ofType(SEARCH_VALUE_CHANGED)
      .debounceTime(250)
      .map(action => action.value)
      .startWith(getState().search.value)
      .filter(value => !!value),
    actions.ofType(SET_PAGE)
      .map(action => action.value)
      .startWith(getState().search.page),
    (q, page) => `https://api.github.com/search/repositories?q=${q}&page=${page}&per_page=${resultsPerPage}`
  )
    .switchMap(url =>
      Observable.ajax.getJSON(url)
        .map(searchFulfilled)
        .catch(({ xhr }) => Observable.of(searchErrored(xhr.response.message)))
        .startWith(searchPending())
    );

// STORE SETUP

const rootDelegator = combineDelegators(
  fetchUserDelegator,
  searchDelegator
);

const store = createStore(rootReducer,
  applyMiddleware(
    reduxObservable(rootDelegator)
  )
);

So far we're very happy with this new approach, although it's a bummer that you have to import and combine all of them ahead of time, just like the reducers.

THAT SAID. We don't have a super great name for this new pattern. We discussed internally a lot and the best name we could come up with is "action delegator" or just delegator for short. We're open to suggestions on what to name these, before we create the documentation for them and start suggesting them. In some ways, they are middleware for the middleware but that's obviously not a name. redux-saga calls their similar thing a "saga", basically punting on having a technically descriptive name in favor of an abstract one. I'm probably fine with using an abstract name, if someone has a clever one. Apparently, the term "saga" is a real thing, outside of that redux-saga, but I feel like this is stretching that definition..chime in if you can provide a good argument in favor of using the term saga as well.

Add optional default scheduler argument to middleware factory

The idea is provide the option to schedule observation with a scheduler passed to the middleware factory. This might prove useful for integration testing and possibly other uses.

Process Manager dispatch multiple actions

I tried to return an array of action objects from the observable, Uncaught Error: Actions must be plain objects. Use custom middleware for async actions.

It would be great if it worked, then I could dispatch to a logger or notification component, for success and errors.

Add tests to the example app

Any plans to add tests to the example?

Merge strategy factories (higher-order thunkservables)

Problem

Basically, just need a cleaner way to handle composed cancellation via something like a "switch" map. Generally in Rx these are referred to as "merge strategies".

Conceptually they'd look like this:

const switch = (thunkservable)  => {
  let subject = new Subject();
  let subscription;
  return (actions, store) => {
    if (subscription) {
      subscription.unsubscribe() 
    }
    subscription = thunkservable(actions,store).subscribe(::subject.next);
    return subject;
  };
}

const merge = (thunkservable, concurrency = Number.POSITIVE_INFINITY) => {
  let subject = new Subject();
  let merged = subject.mergeAll(concurrency);
  return (actions, store) => {
    subject.next(thunkservable(actions, store));
    return merged;
  };
}

const concat = (thunkservable) => {
  return merge(thunkservable, 1);
}

And would be used as such:

let sendAutoComplete = (q) => switch(() => Observable.ajax.get(`some/endpoint${q}`));

Redux Chrome Dev Tools

Do you want to request a feature or report a bug?

What is the current behavior?

When dispatching from the redux devtools in chrome, the action is triggered but the Epic is not.

If the current behavior is a bug, please provide the steps to reproduce and if possible a minimal demo of the problem via https://jsbin.com or similar.

Having redux devtools installed you can dispatch {type: 'PING'} and the UI updates, however, the Epic to trigger 'PONG' is not invoked.

https://output.jsbin.com/vuviliveka/1

What is the expected behavior?

When sending dispatches from devtools Epics are able to react to them.

Which versions of redux-observable, and which browser and OS are affected by this issue? Did this work in previous versions of redux-observable?

OS: OS X
latest version of everything

Not sure if this worked before, just now diving into all this.

Compatability with thunk

Currently both redux-observable and thunk are doing a typeof action === 'function' - was wondering if taking a cue from redux-rxjs and checking if the action is an observable instead of just a function would allow for this to be compatible with thunk?

Or is there something else going on due to the support for teardown/cancelling an observable that I'm missing that wouldn't allow this to be possible?

Document processManager

Edit: WIP docs are up, please review! #55

I strongly feel that the best and most powerful way to use redux-observable is with processManagers rather than thunkservables. Unfortunately, right now we're only documenting the thunkservable methodology.

I, personally, would like to deprecate the thunkservable method prior to RC, but I feel like @jayphelps might have different opinions.

Either way, we really need to document it in a first class way.

Canceling of all async actions?

is there some way to cancel all asynchronous actions when you want to destroy the store? If not, please consider this a feature request 😊. For example, a special action type that gets used with takeUntil at subscription time here.

redux-epic and redux-observable

Hello Everyone,

I'm the creator of Redux-Epic, which is a redux library in similar fashion to redux-saga but built using RxJs with first class support for SSR.

In BerkeleyTrue/redux-epic#13 it has been brought to my attention that this library has arrived at the same path as redux-epic, triggering observable in middleware a la redux-saga.

But there are also various differences that I go into detail in the issue above.

I'm opening this issue to discuss whether merging the two products would benefit everybody and whether the goals I've laid out for redux-epic are compatible with the goals of redux-observable.

• First class support for SSR
• Designed for Universal JavaScript (read: Isomorphic)
• Ability to inject dependencies
• Disposable Epics

Add documentation suggesting users new to RxJS start with something like redux-thunk

I'd like to start prominently suggesting that, if you're not familiar with RxJS, you start off with something like redux-thunk to be productive and only differ to redux-observable for the complex async stuff it truly shines the most at.

It's perfectly acceptable to use redux-observable for all of it (we do) but if you're new to RxJS or even just not super comfortable with it, we don't want you frustrated and non-productive having to learn RxJS for trivial async tasks. It's a much better idea to use these sorts of things together in harmony.

Right now too many people assume they should choose one or the other, so they may either shy away from redux-observable entirely or commit to using it for everything and be in over their head and use use anti-patterns like imperatively manage subscriptions and stuff.

Testing story

We need to improve the documentation around testing. Partially the problem is that testing RxJS v5 isn't very well documented either, but we can't use that as an excuse forever.

---

Edit: this thread contains plenty of great ideas, we just need someone willing to PR better documentation around them.

reduxObservable.js Uncaught TypeError: Cannot read property 'subscribe' of undefined

Hi.

This maybe me, but I keep getting this error

reduxObservable.js:21 Uncaught TypeError: Cannot read property 'subscribe' of undefined

My manager

import Rx from 'rxjs'
import { todoHz } from '../db' // I'm using horizon.io

import {
  getTodosSuccessful
} from '../actions/todos'

const { ajax, of } = Rx.Observable;

const todosManager = action$ => {
  action$.filter(action => console.log('action:', action))
  console.log(action$)
  action$
    .ofType('request todos')
    .merge(todoHz.fetch())
    .map(todos => {
      console.log('todos:', todos)
      getTodosSuccessful(todos)
    })
    .catch(err => console.log(err))
}

export default todosManager

this is then combined (as I intend to have different managers in different files going forward)

import { combineDelegators } from 'redux-observable'

import todosManager from './todos'

export default combineDelegators(todosManager)

My store

import { createStore, applyMiddleware, compose } from 'redux'

import invariant from 'redux-immutable-state-invariant'
import devTools from 'remote-redux-devtools'

import { reduxObservable } from 'redux-observable';
import { persistStore, autoRehydrate } from 'redux-persist'

import localforage from 'localforage'

import { actionLogger } from '../middleware'

import rootReducer from '../reducers'
import rootManager from '../managers'

export default function configure(initialState) {
  const middleware = [invariant(), reduxObservable(rootManager), actionLogger]

  const enhancer = compose(
    applyMiddleware(...middleware),
    autoRehydrate(),
    devTools({
      name: 'todo_devtool',
      realtime: true,
      hostname: 'localhost',
      port: 8000,
      maxAge: 30,
      filters: { blacklist: ['EFFECT_RESOLVED'] }
    })
  )

  const store = createStore(rootReducer, initialState, enhancer)

  localforage.config({
    name: 'todo_dev',
    version: 1.0,
    storeName: 'todo_dev_app', // Should be alphanumeric, with underscores.
    description: 'Todo Dev'
  })

  persistStore(store, {
    storage: localforage
  })

  if (module.hot) {
    module.hot.accept('../reducers', () => {
      const nextReducer = require('../reducers')
      store.replaceReducer(nextReducer)
    })
  }

  return store
}

Can you help me.

PS. Looking forward to the Process Managers / Action Managers

Provide mechanism to run process managers before reducers are hit

Following up on the comment here:

I'm using CouchDB (a document store) to sync data between client and server in react-native. The client database is considered the source of truth for all UI on the client. When a "data synced" event is emitted, I'm running a few queries against the db using Observable process managers, and emitting redux actions accordingly. The actions all follow the template of UPDATE_X with an array of X objects, where X might be users, for example. redux-optimist doesn't handle these types of actions well. Instead of UPDATE_X with an array of 5 objects, it prefers to see a sequence of 5 actions like ADD_X. That's not efficient enough for my needs, and I don't feel that library should add logic to handle list updates in that manner (too domain-specific IMO).

My way around it currently is to wrap all UPDATE_X action creators with a thunkservable that reverts all pending optimistic actions, runs the wrapped action, then replays the optimistic actions. I'm unable to do this with the current process managers because the reducers have been hit before the process manager runs.

Add TypeScript type definitions

Do you plan to add typings for this library?

RFC: onErrorReturn() operator

Some Rx implementations contain an onErrorReturn operator. http://reactivex.io/documentation/operators/catch.html (scroll down and select it from RxJava list)

instructs an Observable to emit a particular item when it encounters an error, and then terminate normally

The onErrorReturn method returns an Observable that mirrors the behavior of the source Observable, unless that Observable invokes onError in which case, rather than propagating that error to the observer, onErrorReturn will instead emit a specified item and invoke the observer’s onCompleted method. javadoc

This operator is usually similar to .catch() but that it expects your provided selector to return a single value instead of an Observable.

In practice this is exactly what people do 99% of the time in redux-observable.

Using .catch()

// return an Observable
.catch(error => Observable.of({
  type: FETCH_USER_REJECTED,
  payload: error.xhr.response,
  error: true
}))

const fetchUserEpic = action$ =>
  action$.ofType(FETCH_USER)
    .mergeMap(action =>
      ajax.getJSON(`/api/users/${action.payload}`)
        .map(fetchUserFulfilled)
        .catch(error => Observable.of({
            type: FETCH_USER_REJECTED,
            payload: error.xhr.response,
            error: true
        }))
    );

Using .onErrorReturn()

// Return a value
.onErrorReturn(error => ({
  type: FETCH_USER_REJECTED,
  payload: error.xhr.response,
  error: true
})

const fetchUserEpic = action$ =>
  action$.ofType(FETCH_USER)
    .mergeMap(action =>
      ajax.getJSON(`/api/users/${action.payload}`)
        .map(fetchUserFulfilled)
        .onErrorReturn(error => ({
            type: FETCH_USER_REJECTED,
            payload: error.xhr.response,
            error: true
        })
    );

Or more likely with action creators:

.onErrorReturn(error => fetchUserRejected(error.xhr.response))

I've noticed most people don't fully understand what .catch() is doing--that you're returning an observable that should be switched to, on error, which itself may or may not terminate. While .catch() is certainly handy, onErrorReturn on the other hand has more obvious semantics for redux-observable use cases.

This probably belongs in RxJS v5 core, but before I pitch it there I wanted to confirm the community agrees this would be super helpful. Worst case, we can add onErrorReturn to the ActionsObservable prototype, but I'm confident core will want it too.

Cc/ @Blesh

[TypeScript] combineEpics is missing in index.d.ts.

I got an TSC error using [email protected]

(6,10): error TS2305: Module '"/path/to/project/node_modules/redux-observable/index"' has no exported member 'combineEpics

Mostly because it's missing in index.d.ts

RFC: provide state$ as a stream argument

The store provided by redux (and given to Epics) isn't a full store, so it doesn't have the store[Symbol.observable]() interop point to support Observable.from(store). This is by their design.

Let's consider changing the Epic signature to function (action$: ActionsObservable, state$: BehaviorSubject<State>, store: Store), making a stream of stage changes the second argument and moving the store to the third argument.

const fetchUserEpic = (action$, state$) =>
  action$.ofType(FETCH_USER)
    .mergeMap(action =>
      getJson(`/users/${action.id}`, { 'Authorization': `Bearer ${state$.value.authToken}` })
        .map(respose => fetchUserFulfilled(response))
    );

// or the "reactive" way, but more verbose

const fetchUserEpic = (action$, state$) =>
  action$.ofType(FETCH_USER)
    .withLatestFrom(state$.pluck('authToken'))
    .mergeMap(([action, authToken]) =>
      getJson(`/users/${action.id}`, { 'Authorization': `Bearer ${authToken}` })
        .map(respose => fetchUserFulfilled(response))
    );

const localStorageStateEpic = (action$, state$) =>
  state$
    .filter(state => state.someState)
    .distinctUntilChanged() // state is supposed to be immutable, so this should be safe
    .throttleTime(500) // may or may not want to do something like this for perf reasons
    .do(state => localStorage.setItem('someState', JSON.stringify(state)))
    .ignoreElements(); // we have no actions to emit

I'd want to keep the store still, since while not necessarily idiomatic to use, is a handy escape hatch to "get shit done".

If this is a good idea in general, we'd need to decide what kind of observable the state changes were in. My initial thoughts were a BehaviorSubject so that the imperative value property was available for the same common cases people use store.getState() for or they can operate on it and it emit the last value right away. Because it's an observable, it feels slightly more idiomatic and then people don't need to learn about the fact that Redux's store supports the Symbol.observable interop so you could just do Observable.from(store), which isn't immediately obvious.

Obviously, this is not really all that different than existing solution..it's more whether we should provide a obvious "Rx way" or not. Please feel free to debate this.

Cc/ @Blesh

debounce ajax managers?

I know it's simple to debounce an input field to throttle ajax calls, but is it possible to debounce this function directly and bypassing to implement input debounce?

const fetchUserManager = action$ =>
  action$.ofType(FETCH_USER)
    .switchMap(action =>
      ajax.getJSON(`https://api.github.com/users/${action.payload}`)
        .map(fetchUserFulfilled)
        .catch(({ xhr }) => of(fetchUserRejected(xhr.response)))
    );

Consider adding checks for process managers that don't actually return an observable

#64 demos that it's easy for people to not actually return an observable on accident (or intentionally, not realizing they must).

We should consider adding checks to the combine helper and the middleware itself to confirm what they're returning is subscribable.

Sequencing actions together

This is a fantastic library. Thank you for sharing!

Would it possible to provide an example that composes two asynchronous actions together into one asynchronous action that is cancelable?

I believe it is possible however struggling at the movement to get my head around it.

angular/ember/etc examples?

Nothing about this is React specific at all, and with momentum growing with angular-redux and ember-redux it would be very beneficial for usage examples with them added to the docs as well as in the examples/ directory.

https://github.com/angular-redux/ng-redux (angular 1)
https://github.com/angular-redux/ng2-redux (angular 2)
https://github.com/toranb/ember-redux

Best practice question with a flatmap-debounced-search example

Thank you for this library – it's awesome!

I was trying to test out a more complex example. Something that replicates a search input which makes a debounced api call onChange, etc.

This is what I came up with, it's a complete example so you can clone and run it:
https://github.com/winkerVSbecks/redux-obsv-search-example/blob/master/src/actions/index.js#L10

Is this the right way to go about it? My primary concern is whether you need that init search action or can this somehow be achieved using only one onChange action.

Overlap and differences of redux-observable and redux-stream

Hey guys,

It's me, that guy that wrote and experimented a bit with state streams. Thanks again @Blesh for your feedback, it was much appreciated seeing as I'm a newcomer to the open source world. I actually took your feedback to heart about how there is space for both RxJS and Redux and how Observable.from(store) just works™. Long story short I revisited Redux again and realised that I could write a Redux store enhancer, which I have been experimenting with for the past 2 weeks. Recently I saw you guys over here at redux-observable have deprecated your thunk-servable approach, to be replaced by the action managers. This is actually very interesting as there is a lot of overlap with the Redux store enhancer approach I am experimenting with.

The experimental library is called redux-stream and can be found over here: https://github.com/mcoetzee/redux-stream

Its main focus is on composing side effect streams (as apposed to state streams). What you guys call action managers I called side effect producers. Small extract from the readme:

Use RxJS 5 to compose side effect streams with Redux.

Features:

• Works alongside middleware
• Gradually introduce redux-stream into an application that makes use of thunk-middleware
• Side effect composing of:
  • Ajax
  • Inter module actions (eg. selecting a search filter triggers a search)
  • Independent modules' state changes
  • Independent modules' actions
• Dispatch thunks to cater for conditional dispatching

I believe the biggest difference between our two approaches is that the store enhancer can work alongside thunk-middleware. I'm of the opinion that an approach that works with redux-thunk will really help drive adoption, as you can take it for a test drive and incrementally add side effect streams to your Redux app.

I would really like to hear what you guys think w.r.t. the differences in our approaches etc.

P.S. I also have some concerns that I would like to discuss

RFC: default error handling of Epics

The middleware is intentionally designed to accept a single, root Epic. This means that all your Epics are typically composed together using combineEpics(...epics).

The problem with this approach is that if an exception is thrown inside one of your epics and it isn't caught by you, it will "bubble" up all the way to the root epic and cause the entire output stream to terminate aka error(). As an example, Observable.ajax() currently throws exceptions for non-200 responses. So one bad AJAX request can terminate all of your epics and your app no longer listens for actions to handle side effects.

You can demo this sort of behavior here: http://jsbin.com/yowiduh/edit?js,output notice that it works the first time, errors the second, and any subsiquent time does nothing because the epic has terminated!

This is not unlike most programming models (where an uncaught exception terminates the "process"). In fact, this behavior is the same as if you were doing the equivalent in imperative JS with generators or similar. So RxJS is "correct" in the implementation.

However, because this has critical implications, we've been discussing adding some default behavior to mitigate this; especially for people who may be fairly new to Rx and not realize the implications.

Here are a couple options:

1. .catch() it and and resubscribe to the root epic, basically "restarting" epics to continue listening.
   • Epics that maintain internal state will have lost that state completely, even ones that didn't cause the exception.
2. or have combineEpics() listen for individual epics throwing exceptions and restart only the one that produced the error.

Alternative suggestions are welcome.

---

If we proof-of-concept option 1. it might look something like this:

http://jsbin.com/kalite/edit?js,output

const output$ = rootEpic(action$, store).catch((error, source) => {
  Observable.throw(error, Scheduler.async).subscribe();
  return source;
});

output$.subscribe(store.dispatch);

I'm not sure we want to do Observable.throw(error, Scheduler.async).subscribe(); but I used it here because I do want the exception to be thrown so window.onerror sees it and it shows up in the console as expected, but we also need to return the original source observable so I'm using Scheduler.async to "defer" the exception. There's prolly a better way to do this. A simple setTimeout prolly is ok instead.

Cc/ @Blesh

implicit return-type error during typescript compilation

When I try to compile my typescript project which uses redux-observable, the following error messages result:

node_modules/redux-observable/index.d.ts(7,3): error TS7010: 'lift', which lacks return-type annotation, implicitly has an 'any' return type.
node_modules/redux-observable/index.d.ts(8,3): error TS7010: 'ofType', which lacks return-type annotation, implicitly has an 'any' return type.

Question re: SSR

If I use React Router's match function, and trigger actions from all matched components, is there a good way to coordinate when all these actions (Observables) are completed? This would be essential for SSR, and is normally coordinated with Promises and Promise.all.

process manager, wait for another process manager

I haven't figured out a way to wait for another process manager inside a process manager,
store dispatch simply returns the action object which is of no use.

const fetchWhatEverManager = (action$, store) =>
  action$.ofType(FETCH_USER)
    .switchMap(action => {
      const maybeData = store.getState().getIn([ 'myReducer', 'myData' ]);
      if (!maybeData) {
        // Data has not been fetched to continue, need to fetch it right here
        // What to do here??? Dispatch does not wait for action
        const onlyActionObj = store.dispatch({ type: 'FETCH_MY_DATA_AND_POPULATE_REDUCER' });
      } else {
        // Did not need to fetch, do something with available data and return payload
        const payload = maybeData................
        return Observable.of({ type: 'MY_ACTION', payload });
      }
    });

store doesn't have subscribe?

How come the store send into the epic constructor doesn't contain a subscribe? I would need that in order to create a store$.

Ben, please enable greenkeeper

A new eslint release today broke up build, as described in #14. https://greenkeeper.io is free for open source and will automatically check for these kinds of things and warn us via PR. Here's an example: freeCodeCamp/freeCodeCamp#8617

It's pretty amazing. Only the repo owner (not collaborators) can enable it, so you'll need to do it. Instructions are easy.

https://greenkeeper.io/#getting-started

GETTING STARTED

Three simple commands: Open your terminal, and

$ npm install -g greenkeeper
$ greenkeeper login

Your browser will open and ask you to log in to GitHub. Please do that.

Then, inside the repo you want to keep up to date, do

$ greenkeeper enable

and you’re done! Your first greenkeeper PR will arrive shortly!

Hot Reloading?

How does this work with hot reloading. Do we need some kind of replaceEpic similar to:

  if (module.hot) {
    module.hot.accept('./reducers', () => {
      const reducers = require('./reducers').default
      store.replaceReducer(reducers)
    })
  }

Websocket integration error

I'm trying to integrate with a websocket, but get the following error and don't know why: Uncaught Error: Actions must be plain objects. Use custom middleware for async actions.

Any ideas? Redux Observable seems to be working for everything else just fine.

export function startWebsocket() {
  return (actions, store) => Observable
    .webSocket({
      url: 'wss://example.com/ws/data',
      resultSelector: (msgEvent) => msgEvent.data
    })
    .map((payload) => ({ type: WEBSOCKET_DATA, payload }))
    .takeUntil(actions.ofType(STOP_WEBSOCKET))
    .catch((err) => {
      console.log('ws error', err);

      return Observable.of({type: 'DUMMY'});
    });

Pass function to ofType

When a function (like the ones returned by redux-actions createAction()) are passed to ofType(), they should be converted via toString() automatically. String action specifiers would be passed onward, unmodified.

Current syntax:

const queueSomethingEpic = createAction('app/activities/epic')
const doSomethingEpic = action$ => action$.ofType(queueSomethingEpic.toString())

Proposed syntax:

const doSomethingEpic = action$ => action$.ofType(queueSomethingEpic)

This seems to be an emerging method in the react/redux community. I've found it a useful pattern. But does it belong in redux-observable?

epicMiddleware.replaceEpic(nextEpic) adds nextEpic instead of replacing

To reproduce, simply call epicMiddleware.replaceEpic(nextEpic) mutiple times. All actions then get received multiple times by your epics.

Request to move ActionsObservable into a separate module

@ngrx/effects and redux-observable both have separate implementations of an actions observable that includes a custom operator for filtering actions by type (see StateUpdates#whenAction in @ngrx/effects and ActionsObservable#ofType in redux-observable).

It would be great if redux-observable's ActionsObservable could be moved into a separate module so that both of these libraries can share a common implementation.

cc @Blesh

More examples?

Hey, so as you know it'd be better to have more examples and I am willing to write some, just not sure which ones you want.

For example, here are the Redux Saga examples and here are the regular Redux examples.

Which ones would be beneficial to the community? I was going to do some of the counter ones but conceptually they'd be very similar to the basic example.

TypeError: Cannot read property 'Symbol(Symbol.iterator)' of undefined

Hi @jayphelps thanks for the great work here and EPICS sounds just about right.

Here's another issue. I guess this is me as well, but ...

I've updated my epics (I switched immediately) based on our last conversation #64

import Rx from 'rxjs'
import { todoHz } from '../db'

import {
  getTodosSuccessful
} from '../actions/todos'

const watchTodosEpic = action$ =>
  action$
    .ofType('request todos')
    .merge(todoHz.watch())
    .do(todos => delete todos.type)
    .map(todos => getTodosSuccessful(todos))
    .catch(err => console.log(err)) // eslint-disable-line

const addTodoEpic = action$ =>
  action$
    .ofType('request post todo')
    .takeUntil(action$.ofType('cancel requests'))
    .mergeMap(action => { todoHz.store(action.payload) })
    .catch(err => console.log(err)) // eslint-disable-line

const deleteTodoEpic = action$ =>
  action$
    .ofType('request delete todo')
    .takeUntil(action$.ofType('cancel requests'))
    .mergeMap(action => { todoHz.remove(action.payload) })
    .catch(err => console.log(err)) // eslint-disable-line

export default [watchTodosEpic, addTodoEpic, deleteTodoEpic]

The error

TypeError: Cannot read property 'Symbol(Symbol.iterator)' of undefined

seems to come from the mergeMap. Returning from that line breaks future actions, not returning works, but that error. Another check hopefully.

Cheers

subscribeToResults:45 - TypeError: Cannot read property 'Symbol(Symbol.iterator)' of undefined

Hello,

this is probably a bug. I am trying to run simple app with this library. Having the basic setup I run into following issue:

subscribeToResult.js:45 Uncaught TypeError: Cannot read property 'Symbol(Symbol.iterator)' of undefined

I cannot provide a jsbin (nor similar) as I am behind a firewall and they won't let me out.

My versions are:

"redux": "3.5.2",
"rxjs": "5.0.0-beta.10",
"redux-observable": "0.8.0"

In my index.jsx I have imported import 'rxjs';

If the root epic is empty then it's working fine (doing nothing :) :

rootEpic = (action$, store) => {
  // empty
}

When it contains something (an epic, console.log) the app ends up with the above error:

rootEpic = (action$, store) => {
  console.log("xxx");
}

do you have an estimate about stable version?

Rx.Observable.ajax is not a function

Hi guys,

I follow exactly the basic example and when I try to have .ajax() instead of .of()
I get this error:
Uncaught TypeError: Rx.Observable.ajax is not a function

Any idea? It looks like RxJS made few changes (ReactiveX/rxjs#1506) impacting ajax() but whatever I try, I can't manage to have redux-observable working.

I tried to install rx-dom package and use Rx.DOM.ajax() instead but redux-observable says it doesn't return an Observable object. I'm confused.

Feature Suggestion: Accept an object in combineEpics

Like combineReducers, something like this

if (epics.length === 1 && typeof epics[0] !== 'function') {
  const obj = epics[0];
  const epics = Object.keys(obj)
    .filter(key => typeof obj[key] === 'function')
    .map(key => obj[key]);
}

This would allow a pattern like:

// epics.js
export const incrementIfOddEpic = ...;

// elsewhere
import * as epics from './epics';
const rootEpic = combineEpics(epics);