import { Observable, Observer } from 'rxjs' import { combineEpics, ActionsObservable } from 'redux-observable' import { Store } from 'redux' import * as api from './api' import * as actions from './actions' import * as selectors from './selectors' import * as PopupActions from '../popup/actions' import * as PopupSelectors from '../popup/selectors' import * as AuthSelectors from '../auth/selectors' import { SocketCluster, SocketChannel, } from '../socket/constants' import { makeJoinChannelPendingEpic, makeJoinChannelSuccessEpic, } from '../socket/epic' const socketPendingEpic = makeJoinChannelPendingEpic(SocketCluster.MESSAGES, SocketChannel.USER) const socketSuccessEpic = makeJoinChannelSuccessEpic(SocketCluster.MESSAGES, SocketChannel.USER) export const triggerFetchGroupedActivitiesEpic = (action$: ActionsObservable, store: Store) => { return socketSuccessEpic(action$) .switchMap(() => { // delay on reconnect to avoid herding if (selectors.isLoaded(store.getState())) { const delayMs = Math.random() * 60000 return Observable.timer(delayMs) .first() .map(() => actions.fetchActivityGroups()) } else { return Observable.of(actions.fetchActivityGroups()) } }) } export const fetchGroupedActivities = (action$: ActionsObservable, store: Store) => { return action$.ofType(actions.fetchActivityGroups.toString()) .pluck('payload') .map(() => ({ accessToken: AuthSelectors.accessToken(store.getState()) })) .mergeMap((options) => { return api.fetchActivityGroups(options) .map(activityGroups => actions.fetchActivityGroupsSuccess(activityGroups)) .startWith(actions.fetchActivityGroupsPending()) .catch(() => Observable.of(actions.fetchActivityGroupsFailure())) }) } export const fetchGroupedActivitiesByEmail = (action$: ActionsObservable, store: Store) => { return action$.ofType(PopupActions.updateSearchQuery.toString()) .pluck('payload') .filter(query => !!query) .map((query: string) => ({ email: query, accessToken: AuthSelectors.accessToken(store.getState()) })) .switchMap((options) => { return api.fetchActivityGroups(options) .map(activities => actions.fetchActivityGroupsByEmailSuccess(activities)) .startWith(actions.fetchActivityGroupsByEmailPending()) .catch(() => Observable.of(actions.fetchActivityGroupsByEmailFailure())) }) } interface ChannelPayload { body: any } export const joinUserChannelPendingEpic = (action$: ActionsObservable, store: Store) => { return socketPendingEpic(action$) .switchMap(({ channel }) => { return Observable.create((observer: Observer) => { channel.on('outbound-group', ({ body }: ChannelPayload) => { observer.next(actions.receiveActivityGroups([body])) const searchQuery = PopupSelectors.searchQuery(store.getState()) if (searchQuery && body.email.indexOf(searchQuery) !== -1) { observer.next(actions.receiveActivityGroupsByEmail([body])) } }) }) }) } export const epic = combineEpics( triggerFetchGroupedActivitiesEpic, fetchGroupedActivities, fetchGroupedActivitiesByEmail, joinUserChannelPendingEpic, )