Class: Warren::Broadcast

Inherits:
Object
  • Object
show all
Defined in:
lib/warren/broadcast.rb

Overview

Class Warren::Broadcast provides a connection pool of threadsafe RabbitMQ channels for broadcasting messages

Defined Under Namespace

Classes: Channel

Instance Method Summary collapse

Constructor Details

#initialize(server: {}, exchange:, pool_size: 14) ⇒ Broadcast

Creates a warren but does not connect.

Parameters:

  • server (Hash)

    Server config options passes straight to Bunny

  • exchange (String)

    The name of the exchange to connect to

  • pool_size (Integer)

    The connection pool size


35
36
37
38
39
# File 'lib/warren/broadcast.rb', line 35

def initialize(server: {}, exchange:, pool_size: 14)
  @server = server
  @exchange_name = exchange
  @pool_size = pool_size
end

Instance Method Details

#<<(message) ⇒ Warren::Broadcast

Borrows a RabbitMQ channel, sends a message, and immediately returns it again. Useful if you only need to send one message.

Parameters:

  • message (Warren::Message)

    The message to broadcast. Must respond to #routing_key and #payload

Returns:

  • (Warren::Broadcast)

    Returns itself to allow chaining. But you're probably better off using #with_chanel inthat case


80
81
82
83
# File 'lib/warren/broadcast.rb', line 80

def <<(message)
  with_chanel { |chanel| chanel << message }
  self
end

#connecttrue

Opens a connection to the RabbitMQ server. Will need to be re-initialized after forking.

Returns:

  • (true)

    We've connected!


46
47
48
49
# File 'lib/warren/broadcast.rb', line 46

def connect
  reset_pool
  start_session
end

#disconnecttrue

Closes the connection. Call before forking to avoid leaking connections

Returns:

  • (true)

    We've disconnected


57
58
59
# File 'lib/warren/broadcast.rb', line 57

def disconnect
  close_session
end

#with_chanel(&block) ⇒ void

This method returns an undefined value.

Yields an exchange which gets returned to the pool on block closure

Yield Returns:


68
69
70
# File 'lib/warren/broadcast.rb', line 68

def with_chanel(&block)
  connection_pool.with(&block)
end