Contents

require "beetle_joevandyk_extensions/version"
require 'beetle'
require 'timeout'

module BeetleJoevandykExtensions
  class BeetleError < RuntimeError; end

  DEFAULT_TIMEOUT = 10

  def setup
    uri = URI.parse(ENV['AMQP_URL'])
    Beetle.config.servers = uri.host
  end

  def rpc name, message, options={}
    setup
    client = Beetle::Client.new
    client.register_message name
    Timeout.timeout(options[:timeout] || DEFAULT_TIMEOUT, BeetleError) do
      status, result = client.rpc(name, message.to_json)
      raise BeetleError.new(result.to_s) if status != "OK"
      JSON.parse(result)
    end
  end

  def listen name, &block
    setup
    client = Beetle::Client.new
    client.register_queue name
    client.register_handler name do |input|
      block.call(JSON.parse(input.data)).to_json
    end
    client.listen do
      puts "Started #{ name } at #{ Time.now }"
    end
  end

  def publish name, message
    setup
    client = Beetle::Client.new
    client.register_message name
    client.publish name, message
  end
end

Beetle.extend(BeetleJoevandykExtensions)

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
beetle_joevandyk_extensions-0.0.10	lib/beetle_joevandyk_extensions.rb