RethinkDB changefeeds with ruby, celluloid, and angelo

README.md

$ brew install rethinkdb
$ gem install angelo
$ gem install rethinkdb

You'll want something for testing:

$ npm install -g wscat
$ brew install httpie

Boot it by simply running it

ruby app.rb

Test it:

$ wscat -c ws://localhost:4567/sockets
$ http -f -v POST localhost:4567 message=cats

app.rb

require 'angelo'
require 'rethinkdb'
include RethinkDB::Shortcuts
conn = r.connect(:host=>"localhost", :port=>28015)

class ChangesActor
  include Celluloid
  include Celluloid::Logger

  def initialize(app)
    @app = app
    async.publish_changes
  end

  def publish_changes
    info 'listening for changes on rethinkdb'
    conn = r.connect(:host=>"localhost", :port=>28015)
    r.table('messages').changes.run(conn).each { |message|
      @app.websockets[:bar].each { |ws| ws.write message['new_val']['text'] }
    }
  end
end

class App < Angelo::Base
  websocket '/sockets' do |ws|
    websockets[:bar] << ws
    conn = r.connect(:host=>"localhost", :port=>28015)
  end

  get '/' do
    'Hello World!'
  end

  post '/' do
    conn = r.connect(:host=>"localhost", :port=>28015)

    r.table('messages').insert({
      'text' => params[:message]
    }).run(conn)

    'added a new message'
  end
end

changes = ChangesActor.new(App)
App.run!