Class: Ons::Consumer

Inherits:
Object
  • Object
show all
Defined in:
lib/ons/consumer.rb

Overview

the ONS Consumer

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(access_key, secret_key, consumer_id, options = {}) ⇒ Consumer

Create a new aliyun ONS Consumer instance.

Parameters:

  • access_key (String) —

    the access key to aliyun ONS

  • secret_key (String) —

    the secret key to aliyun ONS

  • consumer_id (String) —

    the consumer ID

  • options (Hash{String, Symbol => String}) (defaults to: {})

Options Hash (options):

  • :namesrv_addr (String) —

    the nameserver used to fetching ons_addr

  • :ons_addr (String) —

    the ONS server address

  • :thread_num (String, Fixnum, Bignum) —

    the consumer thread numbers



20
21
22
23
24
25
26
27
28
 
# File 'lib/ons/consumer.rb', line 20

def initialize(access_key, secret_key, consumer_id, options = {})
  @consumer = Internal::Consumer.new(access_key, secret_key, consumer_id, options)

  # register instance
  self.class.instances << self

  # start the global LMFAO event thread, so it could consume messages properly
  LMFAO.start_event_thread
end

Class Method Details

.instances ⇒ <Consumer>

Get all the Consumer instances.

Returns:

  • (<Consumer>) —

    all the Consumer instances



7
8
9
 
# File 'lib/ons/consumer.rb', line 7

def self.instances
  @instances ||= []
end

Instance Method Details

#shutdown ⇒ void

Note:

this method should be called before program exit, otherwise it would case a memory leak.

This method returns an undefined value.

Shutdown the Consumer instance.

Please see also Ons.register_cleanup_hooks if you want call it automatically.



67
68
69
 
# File 'lib/ons/consumer.rb', line 67

def shutdown
  @consumer.shutdown
end

#start ⇒ self

Note:

this method should be called after subscribe.

Note:

thie method will not block the thread, please see also Ons.loop_forever.

Start the Consumer instance.

Returns:

  • (self) —

    returns itself



56
57
58
59
 
# File 'lib/ons/consumer.rb', line 56

def start
  @consumer.start
  self
end

#subscribe(topic, expression, handler = nil) ⇒ self

Subsribe a topic.

Examples:

subscribe tag :tagA under topic :TopicTestMQ

consumer.subscribe('TopicTestMQ', 'tagA') {}

subscribe tag :tagA and :tagB under topic :TopicTestMQ

consumer.subscribe('TopicTestMQ', 'tagA || tagB') {}

subscribe all tags under topic :TopicTestMQ

consumer.subscribe('TopicTestMQ', '*') {}

Parameters:

  • topic (String) —

    the message topic

  • expression (String) —

    the subsribe expression used to filter messages

  • handler (#call) (defaults to: nil) —

    the handler which will handle the message

Returns:

  • (self) —

    returns itself



45
46
47
48
 
# File 'lib/ons/consumer.rb', line 45

def subscribe(topic, expression, handler = nil)
  @consumer.subscribe(topic, expression, handler || Proc.new)
  self
end