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 { path } 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 geodist from 'geodist';

const getGeoDist = begin => end => geodist(begin, end, { exact: true });

const placesSubject = new ReplaySubject();

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

filter$.subscribe(() => emitter(null));

foodItems$
	.mergeMap((foodItems = Map()) => Observable.from(foodItems.toArray()))
	.mergeMap(getPlaceDetails)
	.map(buildPlaceRecord)
	.subscribe(emitter);

location$
	.combineLatest(filter$)
	.debounceTime(200)
	.mergeMap(([location, filter]: [?Position, FilterRecord]) => {
		return findNearbyPlaces({
			location,
			radius: filter.radius,
			search: filter.search,
		});
	})
	.map((val: { location: ?Position, places: ?Array<GooglePlaceObj> }) => {
		if (!val) {
			return;
		}
		const { places } = val;
		const coords = path(['location', 'coords'], val) || {};
		const getDist = getGeoDist({ lat: coords.latitude, lon: coords.longitude });

		return (places || []).map((place: GooglePlaceObj) =>
			buildPlaceRecord({
				...place,
				distance: getDist({ lat: place.geometry.location.lat, lon: place.geometry.location.lng }),
			})
		);
	})
	.subscribe(places => places && places.map(emitter));

export default placesSubject.scan((places, place) => {
	if (!place) {
		return null;
	}
	return setById(places || new Map(), place);
});