Module: ASS

Defined in:
lib/ass.rb,
lib/ass.rb,
lib/ass/serializers.rb

Defined Under Namespace

Modules: BSON, JSON, Marshal Classes: Actor, CallbackFactory, Client, RPC, Server, Topic

Class Method Summary collapse

Class Method Details

.actor(name, opts = {}, &block) ⇒ Object 



53
54
55
56
57
58
59
 
# File 'lib/ass.rb', line 53

def actor(name,opts={},&block)
  s = ASS::Actor.new(name,opts)
  if block
    s.react(&block)
  end
  s
end

.call(name, method, data, opts, meta) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/ass.rb', line 75

def call(name,method,data,opts,meta)
  # make sure the payload hash use string
  # keys. Serialization format might not
  # preserve type.
  payload = {
    #:type => type,
    "method" => method,
    "data" => data,
    "meta" => meta,
  }
  payload.merge("version" => opts[:version]) if opts.has_key?(:version)
  payload.merge("meta" => opts[:meta]) if opts.has_key?(:meta)
  dummy_exchange(name).publish(ASS.serializer.dump(payload),opts)
  true
end

.cast(name, method, data, opts, meta) ⇒ Object

maybe move cast and call into ASS::Server’s class methods



71
72
73
 
# File 'lib/ass.rb', line 71

def cast(name,method,data,opts,meta)
  call(name,method,data,opts.merge(:reply_to => nil),meta)
end

.client(opts = {}) ⇒ Object

the opts is used to initiate an RPC



66
67
68
 
# File 'lib/ass.rb', line 66

def client(opts={})
  ASS::Client.new(opts)
end

.dummy_exchange(name) ⇒ Object

this would create a dummy MQ exchange object for the sole purpose of publishing the message. Will not clobber existing server already started in the process.



95
96
97
 
# File 'lib/ass.rb', line 95

def dummy_exchange(name)
  @mq.direct(name,:no_declare => true)
end

.mq ⇒ Object 



37
38
39
 
# File 'lib/ass.rb', line 37

def mq
  @mq
end

.rpc(opts = {}) ⇒ Object 



61
62
63
 
# File 'lib/ass.rb', line 61

def rpc(opts={})
  ASS::RPC.new(opts)
end

.serializer ⇒ Object 



41
42
43
 
# File 'lib/ass.rb', line 41

def serializer
  @serializer
end

.server(name, opts = {}, &block) ⇒ Object 



45
46
47
48
49
50
51
 
# File 'lib/ass.rb', line 45

def server(name,opts={},&block)
  s = ASS::Server.new(name,opts)
  if block
    s.react(&block)
  end
  s
end

.start(settings = {}) ⇒ Object

MQ = nil



21
22
23
24
25
26
27
28
29
30
 
# File 'lib/ass.rb', line 21

def start(settings={})
  raise "should have one ASS per eventmachine" if EM.reactor_running? == true # allow ASS to restart if EM is not running.
  EM.run {
    @serializer = settings.delete(:format) || ::Marshal
    raise "Object Serializer must respond to :load and :dump" unless @serializer.respond_to?(:load) && @serializer.respond_to?(:dump)
    @mq = ::MQ.new(AMQP.start(settings))
    # ASS and its worker threads (EM.threadpool) should share the same MQ instance.
    yield if block_given?
  }
end

.stop ⇒ Object 



32
33
34
35
 
# File 'lib/ass.rb', line 32

def stop
  AMQP.stop{ EM.stop }
  true
end