Add server, pub/sub and SSE.

.rvmrc

@@ -0,0 +1,81 @@
+#!/usr/bin/env bash
+
+# This is an RVM Project .rvmrc file, used to automatically load the ruby
+# development environment upon cd'ing into the directory
+
+# First we specify our desired <ruby>[@<gemset>], the @gemset name is optional.
+environment_id="ruby-1.9.2-p290@heathub"
+
+#
+# Uncomment the following lines if you want to verify rvm version per project
+#
+# rvmrc_rvm_version="1.10.2" # 1.10.1 seams as a safe start
+# eval "$(echo ${rvm_version}.${rvmrc_rvm_version} | awk -F. '{print "[[ "$1*65536+$2*256+$3" -ge "$4*65536+$5*256+$6" ]]"}' )" || {
+#   echo "This .rvmrc file requires at least RVM ${rvmrc_rvm_version}, aborting loading."
+#   return 1
+# }
+#
+
+#
+# Uncomment following line if you want options to be set only for given project.
+#
+# PROJECT_JRUBY_OPTS=( --1.9 )
+#
+# The variable PROJECT_JRUBY_OPTS requires the following to be run in shell:
+#
+#    chmod +x ${rvm_path}/hooks/after_use_jruby_opts
+#
+
+#
+# First we attempt to load the desired environment directly from the environment
+# file. This is very fast and efficient compared to running through the entire
+# CLI and selector. If you want feedback on which environment was used then
+# insert the word 'use' after --create as this triggers verbose mode.
+#
+if [[ -d "${rvm_path:-$HOME/.rvm}/environments" \
+  && -s "${rvm_path:-$HOME/.rvm}/environments/$environment_id" ]]
+then
+  \. "${rvm_path:-$HOME/.rvm}/environments/$environment_id"
+
+  if [[ -s "${rvm_path:-$HOME/.rvm}/hooks/after_use" ]]
+  then
+    . "${rvm_path:-$HOME/.rvm}/hooks/after_use"
+  fi
+else
+  # If the environment file has not yet been created, use the RVM CLI to select.
+  if ! rvm --create use  "$environment_id"
+  then
+    echo "Failed to create RVM environment '${environment_id}'."
+    return 1
+  fi
+fi
+
+#
+# If you use an RVM gemset file to install a list of gems (*.gems), you can have
+# it be automatically loaded. Uncomment the following and adjust the filename if
+# necessary.
+#
+# filename=".gems"
+# if [[ -s "$filename" ]]
+# then
+#   rvm gemset import "$filename" | grep -v already | grep -v listed | grep -v complete | sed '/^$/d'
+# fi
+
+# If you use bundler, this might be useful to you:
+# if [[ -s Gemfile ]] && ! command -v bundle >/dev/null
+# then
+#   printf "%b" "The rubygem 'bundler' is not installed. Installing it now.\n"
+#   gem install bundler
+# fi
+# if [[ -s Gemfile ]] && command -v bundle
+# then
+#   bundle install
+# fi
+
+if [[ $- == *i* ]] # check for interactive shells
+then
+    echo "Using: $(tput setaf 2)$GEM_HOME$(tput sgr0)" # show the user the ruby and gemset they are using in green
+else 
+	echo "Using: $GEM_HOME" # don't use colors in interactive shells
+fi
+

Gemfile

@@ -0,0 +1,13 @@
+source 'http://rubygems.org'
+
+gem 'eventmachine', :git => 'git://github.com/eventmachine/eventmachine.git'
+gem 'em-http-request', :git => 'git://github.com/igrigorik/em-http-request.git'
+
+gem 'yajl-ruby', :require => 'yajl'
+gem 'em-stathat'
+gem 'log4r'
+gem 'placefinder'
+gem 'em-mongo'
+gem "bson_ext", "~> 1.6.2"
+gem 'mongo'
+gem 'goliath'

Gemfile.lock

@@ -0,0 +1,85 @@
+GIT
+  remote: git://github.com/eventmachine/eventmachine.git
+  revision: a92b667684ed17602fa91915e9095c1d7710ba43
+  specs:
+    eventmachine (1.0.0.beta.4)
+
+GIT
+  remote: git://github.com/igrigorik/em-http-request.git
+  revision: 6159c1643fc38e9a947c58834221856ed6fdf392
+  specs:
+    em-http-request (1.0.2)
+      addressable (>= 2.2.3)
+      cookiejar
+      em-socksify
+      eventmachine (>= 1.0.0.beta.4)
+      http_parser.rb (>= 0.5.3)
+
+GEM
+  remote: http://rubygems.org/
+  specs:
+    addressable (2.2.8)
+    async-rack (0.5.1)
+      rack (~> 1.1)
+    bson (1.6.2)
+    bson_ext (1.6.2)
+      bson (~> 1.6.2)
+    cookiejar (0.3.0)
+    em-mongo (0.4.2)
+      bson (>= 1.1.3)
+      eventmachine (>= 0.12.10)
+    em-socksify (0.2.0)
+      eventmachine (>= 1.0.0.beta.4)
+    em-stathat (0.1.0)
+      em-http-request (~> 1.0)
+    em-synchrony (1.0.1)
+      eventmachine (>= 1.0.0.beta.1)
+    goliath (0.9.4)
+      async-rack
+      em-synchrony (>= 1.0.0)
+      eventmachine (>= 1.0.0.beta.3)
+      http_parser.rb
+      http_router (~> 0.9.0)
+      log4r
+      multi_json
+      rack (>= 1.2.2)
+      rack-contrib
+      rack-respond_to
+    http_parser.rb (0.5.3)
+    http_router (0.9.7)
+      rack (>= 1.0.0)
+      url_mount (~> 0.2.1)
+    httparty (0.8.3)
+      multi_json (~> 1.0)
+      multi_xml
+    log4r (1.1.10)
+    mongo (1.6.2)
+      bson (~> 1.6.2)
+    multi_json (1.3.4)
+    multi_xml (0.5.1)
+    placefinder (0.2.2)
+      httparty (>= 0.7.3)
+    rack (1.4.1)
+    rack-accept-media-types (0.9)
+    rack-contrib (1.1.0)
+      rack (>= 0.9.1)
+    rack-respond_to (0.9.8)
+      rack-accept-media-types (>= 0.6)
+    url_mount (0.2.1)
+      rack
+    yajl-ruby (1.1.0)
+
+PLATFORMS
+  ruby
+
+DEPENDENCIES
+  bson_ext (~> 1.6.2)
+  em-http-request!
+  em-mongo
+  em-stathat
+  eventmachine!
+  goliath
+  log4r
+  mongo
+  placefinder
+  yajl-ruby

README.md

@@ -1,4 +1,16 @@
 heathub
 =======
 
-Source code for heathub.co
+Source code for heathub.co
+
+
+Running
+-------
+
+      bundlex ruby server.rb -v
+
+Then visit
+
+        http://localhost:9000/index.html
+
+Profit.

config/server.rb

@@ -0,0 +1,96 @@
+@log = Log4r::Logger.new('github')
+@log.add(Log4r::StdoutOutputter.new('console', {
+  :formatter => Log4r::PatternFormatter.new(:pattern => "[#{Process.pid}:%l] %d :: %m")
+}))
+
+@latest = []
+
+db = EM::Mongo::Connection.new('localhost').db('heathub_development')
+collection = db.collection('push_events')
+cities_collection = db.collection('cities')
+$channel = EM::Channel.new
+
+process = Proc.new do
+  req = EventMachine::HttpRequest.new("https://github.com/timeline.json").get({
+    :head => {
+      'user-agent' => 'abuiles.com'
+    }
+  })
+
+  req.callback do
+    begin
+      latest = Yajl::Parser.parse(req.response)
+      urls = latest.collect {|e| e['url']}
+      new_events = latest.reject {|e| @latest.include? e['url']}
+
+      @latest = urls
+      new_events.each do |event|
+        location = event["actor_attributes"] && event["actor_attributes"]["location"]
+        next unless location && event["type"] == "PushEvent"
+
+        @log.info "Event in city #{location}"
+
+        cursor = cities_collection.find(_id: location)
+        resp = cursor.defer_as_a
+
+        resp.callback do |cities|
+          if city = cities.first
+            @log.info "using pre-store city #{location}"
+            event["location"] = city["location"]
+            event["city"] = location
+
+            @log.info "Pushing event"
+            $channel.push event
+
+            collection.insert(event)
+          else
+            query = {:q => location, :flags => 'J'}
+            http = EventMachine::HttpRequest.new('http://where.yahooapis.com/geocode').get :query => query
+
+            http.errback { @log.error "Request to placefinder failed for #{location} " }
+            http.callback {
+              result = Yajl::Parser.parse(http.response)
+              if result["ResultSet"]["Found"] > 0
+
+                result = result["ResultSet"]["Results"].first
+                event["location"] = {
+                  "longitude" => result["longitude"],
+                  "latitude"  => result["latitude"]
+                }
+
+                event["city"] = location
+
+                city_info = { location: event["location"] }.merge(_id: location)
+
+                cities_collection.insert( city_info)
+                $channel.push event
+                collection.insert(event)
+              end
+            }
+          end
+
+        end
+
+        resp.errback do |err|
+          raise *err
+        end
+      end
+
+      @log.info "Found #{new_events.size} new events"
+
+      if new_events.size >= 25
+        EM.add_timer(1.5, &process)
+      end
+
+    rescue Exception => e
+      @log.error "Processing exception: #{e}, #{e.backtrace.first(5)}"
+      # @log.error "Response: #{req.response_header}, #{req.response}"
+    end
+  end
+
+  req.errback do
+    @log.error "Error: #{req.response_header.status}, header: #{req.response_header}, response: #{req.response}"
+  end
+end
+
+EM.add_periodic_timer(6, &process)

crawler.rb

@@ -23,16 +23,16 @@
   cities_collection = db.collection('cities')
 
 
-  stop = Proc.new do
-    puts "Terminating crawler"
-    EM.stop
-  end
+  # stop = Proc.new do
+  #   puts "Terminating crawler"
+  #   EM.stop
+  # end
 
-  Signal.trap("INT",  &stop)
-  Signal.trap("TERM", &stop)
+  # Signal.trap("INT",  &stop)
+  # Signal.trap("TERM", &stop)
 
   process = Proc.new do
-    req = HttpRequest.new("https://github.com/timeline.json").get({
+    req = EventMachine::HttpRequest.new("https://github.com/timeline.json").get({
       :head => {
         'user-agent' => 'abuiles.com'
       }

public/index.html

@@ -0,0 +1,35 @@
+<!DOCTYPE html>
+<html>
+<body>
+  <h3>Hello SSE!</h3>
+  <script>
+    var source = new EventSource('/events');
+
+    // new connection opened callback
+    source.addEventListener('open', function(e) {
+      console.log('connection opened');
+    }, false);
+
+    // subscribe to unnamed messages
+    source.onmessage = function(e) {
+      console.log(e);
+      document.body.innerHTML += e.data + '<br />';
+    };
+
+    // listen for signup events
+    source.addEventListener('signup', function(e) {
+      last = e;
+      console.log('Data received');
+       document.body.innerHTML += e.data + '<br />';
+    }, false);
+
+    // connection closed callback
+    source.addEventListener('error', function(e) {
+      if (e.eventPhase == EventSource.CLOSED) {
+        console.log('connection closed');
+      }
+    }, false);
+
+  </script>
+</body>
+</html>

server.rb

@@ -0,0 +1,38 @@
+require 'em-http'
+require 'em-mongo'
+require 'yajl'
+require 'log4r'
+require 'goliath'
+
+class EventGenerator < Goliath::API
+  def on_close(env)
+    return unless env['subscription']
+
+    $channel.unsubscribe(env['subscription'])
+    env.logger.info "Stream connection closed."
+  end
+
+  def response(env)
+    env['subscription'] = $channel.subscribe do |event|
+      env.stream_send(["event:signup", "data:signup event #{Yajl::Encoder.encode(event["location"])}\n\n"].join("\n"))
+    end
+
+    streaming_response(200, {'Content-Type' => 'text/event-stream'})
+  end
+end
+
+
+class Server < Goliath::API
+  use Goliath::Rack::Params
+
+  use Goliath::Rack::Render, 'json'
+  use Goliath::Rack::Heartbeat
+  use Goliath::Rack::Validation::RequestMethod, %w(GET)
+
+  use Rack::Static, :urls => ["/index.html"], :root => Goliath::Application.app_path("public")
+
+
+  get "/events" do
+    run EventGenerator.new
+  end
+end