aretherecookies-mobile/js/streams/PlacesStream.js

// @flow
import { ReplaySubject, Observable } from 'rxjs';
import { buildPlaceRecord } from '../records/PlaceRecord';
import { Map } from 'immutable';
import { findNearbyPlaces, getPlaceDetails } from '../apis/GooglePlacesApi';
import { memoize } from 'ramda';
import { type GooglePlaceObj } from '../records/PlaceRecord';
import { setById } from '../helpers/ImmutableHelpers';
import location$ from './LocationStream';
import filter$ from './FilterStream';
import FilterRecord from '../records/FilterRecord';
import foodItems$ from './FoodItemsStream';
import PlaceRecord from '../records/PlaceRecord';
import FoodItemRecord from '../records/FoodItemRecord';

/**
 * always return a promise and swallow exceptions so as not to break the stream
 */
const safeWrapAjax = ajaxFn =>
	memoize((...args) => {
		return ajaxFn(...args).catch(error => {
			console.log('ERROR: ', error); //eslint-disable-line no-console
			return null;
		});
	});

const safeFindNearbyPlaces = safeWrapAjax(findNearbyPlaces);

const safeGetPlaceDetails = safeWrapAjax(getPlaceDetails);

const placesSubject = new ReplaySubject();

export function emitter(val?: ?PlaceRecord) {
	placesSubject.next(val);
}

emitter(null);

foodItems$
	.mergeMap(
		(foodItems: Map<string, FoodItemRecord> = Map()): Observable<string> => {
			return Observable.from(foodItems.toArray().map(foodItem => foodItem.placeId));
		}
	)
	.distinct()
	.mergeMap(safeGetPlaceDetails)
	.map(buildPlaceRecord)
	.subscribe(emitter);

location$
	.combineLatest(filter$)
	.mergeMap(([location, filter]: [?Position, FilterRecord]) =>
		safeFindNearbyPlaces({
			location,
			radius: filter.radius,
			search: filter.search,
		})
	)
	.map((places: Array<GooglePlaceObj>) => {
		return (places || []).map(buildPlaceRecord);
	})
	.subscribe(places => places.map(emitter));

export default placesSubject.distinct().scan(setById, new Map());