Class: Pulsar::Consumer

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

Overview

Public consumer API for receiving and acknowledging messages.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic:, subscription:, impl: nil) ⇒ Consumer

Returns a new instance of Consumer.



8
9
10
11
12
13
 
# File 'lib/pulsar/consumer.rb', line 8

def initialize(topic:, subscription:, impl: nil)
  @topic = String(topic)
  @subscription = String(subscription)
  @impl = impl
  @closed = false
end

Instance Attribute Details

#subscription ⇒ Object (readonly)

Returns the value of attribute subscription.



6
7
8
 
# File 'lib/pulsar/consumer.rb', line 6

def subscription
  @subscription
end

#topic ⇒ Object (readonly)

Returns the value of attribute topic.



6
7
8
 
# File 'lib/pulsar/consumer.rb', line 6

def topic
  @topic
end

Instance Method Details

#ack(message_or_message_id) ⇒ Object

Raises:



22
23
24
25
26
27
 
# File 'lib/pulsar/consumer.rb', line 22

def ack(message_or_message_id)
  ensure_open!
  raise UnsupportedFeatureError, 'consumer ack is not implemented yet' unless @impl

  @impl.ack(message_or_message_id)
end

#close ⇒ Object 



29
30
31
32
33
34
35
 
# File 'lib/pulsar/consumer.rb', line 29

def close
  return if closed?

  @impl&.close
  @closed = true
  nil
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


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

def closed?
  @closed
end

#receive(timeout: nil) ⇒ Object

Raises:



15
16
17
18
19
20
 
# File 'lib/pulsar/consumer.rb', line 15

def receive(timeout: nil)
  ensure_open!
  raise UnsupportedFeatureError, 'consumer receive is not implemented yet' unless @impl

  @impl.receive(timeout: timeout)
end