rxDemo.js

'use strict';
const dotenv = require('dotenv')
const geocoder = require('geocoder');
const Rx = require('rx')
const {Observable} = Rx
const {create, fromEvent, fromNodeCallback} = Observable

// load environment variables
dotenv.load()

// config api
const googleTanslate = GoogleTranslate(process.env.GOOGLE_TRANSLATE_API_KEY)

const Twitter = require('twitter')

const client = new Twitter({
  consumer_key: process.env.TWITTER_CONSUMER_KEY,
  consumer_secret: process.env.TWITTER_CONSUMER_SECRET,
  access_token_key: process.env.TWITTER_ACCESS_TOKEN_KEY,
  access_token_secret: process.env.TWITTER_ACCESS_TOKEN_SECRET,
})

var stream = client.stream('statuses/filter', {track: 'trump'});
var tweets = fromEvent(stream, 'data')

var simpleTweets = tweets
  .map(({text, coordinates, user: {location}}) => {return {text, coordinates, location}})
  
var [locatedTweets, unlocatedTweets] = simpleTweets
  .partition(({coordinates}) => coordinates)

var geocodes = fromNodeCallback(geocoder.geocode, geocoder)

var geocodedTweets = unlocatedTweets
  .filter(({location}) => location)
  .flatMap(tweet => {
    return geocodes(tweet.location) 
      .filter(location => location.results.length > 0)
      .filter(location => location.results[0].geometry)
      .map(location => location.results[0].geometry.location)
      .map(geometry => {
        tweet.coordinates = geometry 
        return tweet
      })
  })

Observable.merge(geocodedTweets, locatedTweets)
  .subscribe(console.log)