Class: LogjamAgent::AMQPForwarder

Inherits:
Object
  • Object
show all
Includes:
Util
Defined in:
lib/logjam_agent/amqp_forwarder.rb

Constant Summary collapse

RETRY_AFTER = 
10.seconds

Constants included from Util

Util::BIG_ENDIAN, Util::FIXNUM_MAX, Util::META_INFO_DEVICE_NUMBER, Util::META_INFO_TAG, Util::META_INFO_VERSION, Util::UINT64

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Util

#augment_connection_spec, #next_fixnum, #pack_info, #pack_uint64_big_endian, #unpack_info, #unpack_uint64_big_endian, #zclock_time

Constructor Details

#initialize(*args) ⇒ AMQPForwarder

Returns a new instance of AMQPForwarder.



10
11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 10

def initialize(*args)
  opts = args.extract_options!
  @app = args[0] || LogjamAgent.application_name
  @env = args[1] || LogjamAgent.environment_name
  @app_env = "#{@app}-#{@env}"
  @config = default_options.merge!(opts)
  @exchanges = {}
  @bunny = nil
  @sequence = 0
  @paused = nil
  ensure_bunny_gem_is_available
end

Instance Attribute Details

#app ⇒ Object (readonly)

Returns the value of attribute app.



6
7
8
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 6

def app
  @app
end

#env ⇒ Object (readonly)

Returns the value of attribute env.



6
7
8
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 6

def env
  @env
end

Instance Method Details

#default_options ⇒ Object 



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

def default_options
  {
    :host                 => "localhost",
    :exchange_durable     => true,
    :exchange_auto_delete => false,
  }
end

#forward(data, options = {}) ⇒ Object

TODO: mutex!



32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 32

def forward(data, options = {})
  return if paused?
  app_env = options[:app_env] || @app_env
  key = options[:routing_key] || "logs.#{app_env.sub('-','.')}"
  if engine = options[:engine]
    key += ".#{engine}"
  end
  msg = LogjamAgent.encode_payload(data)
  publish(app_env, key, msg)
rescue => error
  reraise_expectation_errors!
  pause(error)
end

#publish(app_env, key, data) ⇒ Object 



46
47
48
49
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 46

def publish(app_env, key, data)
  info = pack_info(@sequence = next_fixnum(@sequence))
  exchange(app_env).publish(data, :key => key, :persistent => false, :headers => {:info => info})
end

#reset(exception = nil) ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/logjam_agent/amqp_forwarder.rb', line 51

def reset(exception=nil)
  return unless @bunny
  if exception
    @bunny.__send__(:close_socket)
  else
    @bunny.stop
  end
rescue
  # swallow StandardError
ensure
  # if bunny throws an exception here, its not usable anymore anyway
  @exchanges = {}
  @bunny = nil
end