changes_pub.rb

#!/usr/bin/ruby

# as of 20100101 requires the latest yajl-ruby
# http://github.com/brianmario/yajl-ruby 

require 'rubygems'
require "yajl/http_stream"
require 'em-http'
require 'mq'

Signal.trap('INT') { AMQP.stop{ EM.stop } }
Signal.trap('TERM'){ AMQP.stop{ EM.stop } }

def main
  url = "http://localhost:5984/"
  db = "test"
  feed_type = "continuous"
  since = "0"
  queue = "changes"
  
  Yajl::HttpStream.get("#{url}#{db}/_changes?feed=#{feed_type}&since=#{since}", :symbolize_keys => true) do |hash|
    if hash[:id]
      #p hash
      EM.run {
        amq = MQ.new
        http = EventMachine::HttpRequest.new("#{url}#{db}/#{hash[:id]}").get
        http.callback {
          amq.queue(queue).publish(http.response)
          #p http.response
          AMQP.stop{ EM.stop }
        }
      }
    end
  end
end

main

changes_sub.rb

#!/usr/bin/ruby

require 'rubygems'
require "yajl/http_stream"
require 'em-http'
require 'mq'

Signal.trap('INT') { AMQP.stop{ EM.stop } }
Signal.trap('TERM'){ AMQP.stop{ EM.stop } }

def work(hash)
  # do some processing on the data here
  hash.store("newkey", "somestuff#{hash["someid"]}")
  hash
end

def main
  url = "http://localhost:5984/"
  db = "results"
  queue = "changes"
  
  parser = Yajl::Parser.new
  encoder = Yajl::Encoder.new
  
  EM.run {
    parser.on_parse_complete = (Proc.new { |hash|
      result = work(hash)
      EventMachine::HttpRequest.new("#{url}#{db}").post :body => encoder.encode(result)
    })
    
    MQ.queue(queue).subscribe do |msg|
      parser.parse(msg)
    end
  }
end

main