Class: Msgr::Connection

Inherits:
Object
  • Object
show all
Includes:
Logging
Defined in:
lib/msgr/connection.rb

Overview

rubocop:disable Metrics/ClassLength

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Logging

#log, #log_name

Constructor Details

#initialize(uri, config, dispatcher) ⇒ Connection

Returns a new instance of Connection.



12
13
14
15
16
17
 
# File 'lib/msgr/connection.rb', line 12

def initialize(uri, config, dispatcher)
  @uri        = uri
  @config     = config
  @dispatcher = dispatcher
  @channels   = []
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



10
11
12
 
# File 'lib/msgr/connection.rb', line 10

def config
  @config
end

#uri ⇒ Object (readonly)

Returns the value of attribute uri.



10
11
12
 
# File 'lib/msgr/connection.rb', line 10

def uri
  @uri
end

Instance Method Details

#bind(routes) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/msgr/connection.rb', line 75

def bind(routes)
  if routes.empty?
    log(:warn) do
      "No routes to bound to. Bind will have no effect:\n" \
      "  #{routes.inspect}"
    end
  else
    bind_all(routes)
  end
end

#bindings ⇒ Object 



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

def bindings
  @bindings ||= []
end

#channel(prefetch: 1) ⇒ Object 



39
40
41
42
43
44
 
# File 'lib/msgr/connection.rb', line 39

def channel(prefetch: 1)
  channel = Msgr::Channel.new(config, connection)
  channel.prefetch(prefetch)
  @channels << channel
  channel
end

#close ⇒ Object 



86
87
88
89
90
 
# File 'lib/msgr/connection.rb', line 86

def close
  @channels.each(&:close)
  connection.close if @connection
  log(:debug) { 'Closed.' }
end

#connect ⇒ Object 



35
36
37
 
# File 'lib/msgr/connection.rb', line 35

def connect
  connection
end

#connection ⇒ Object 



31
32
33
 
# File 'lib/msgr/connection.rb', line 31

def connection
  @connection ||= ::Bunny.new(config).tap(&:start)
end

#delete ⇒ Object 



57
58
59
60
61
62
 
# File 'lib/msgr/connection.rb', line 57

def delete
  return if bindings.empty?
  log(:debug) { "Delete bindings (#{bindings.size})..." }

  bindings.each(&:delete)
end

#exchange ⇒ Object 



46
47
48
 
# File 'lib/msgr/connection.rb', line 46

def exchange
  @exchange ||= channel.exchange
end

#publish(message, opts = {}) ⇒ Object 



23
24
25
26
27
28
29
 
# File 'lib/msgr/connection.rb', line 23

def publish(message, opts = {})
  opts[:routing_key] = opts.delete(:to) if opts[:to]

  exchange.publish message.to_s, opts.merge(persistent: true)

  log(:debug) { "Published message to #{opts[:routing_key]}" }
end

#purge(**kwargs) ⇒ Object 



64
65
66
67
68
69
 
# File 'lib/msgr/connection.rb', line 64

def purge(**kwargs)
  return if bindings.empty?
  log(:debug) { "Purge bindings (#{bindings.size})..." }

  bindings.each {|b| b.purge(**kwargs) }
end

#release ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/msgr/connection.rb', line 50

def release
  return if bindings.empty?
  log(:debug) { "Release bindings (#{bindings.size})..." }

  bindings.each(&:release)
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


19
20
21
 
# File 'lib/msgr/connection.rb', line 19

def running?
  bindings.any?
end