Class: HonestPubsub::Publisher

Inherits:
Object
  • Object
show all
Defined in:
lib/honest_pubsub/publisher.rb

Constant Summary collapse

@@publisher = 
nil

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(exchange = "honest") ⇒ Publisher

Returns a new instance of Publisher.



15
16
17
18
19
20
21
22
23
 
# File 'lib/honest_pubsub/publisher.rb', line 15

def initialize(exchange="honest")
  @exchange = exchange
  @disabled = false
  @logger = ::HonestPubsub::Logger.new()

  @config = Configuration.configuration

  self
end

Instance Attribute Details

#channel ⇒ Object (readonly)

Returns the value of attribute channel.



4
5
6
 
# File 'lib/honest_pubsub/publisher.rb', line 4

def channel
  @channel
end

#publisher ⇒ Object (readonly)

Returns the value of attribute publisher.



3
4
5
 
# File 'lib/honest_pubsub/publisher.rb', line 3

def publisher
  @publisher
end

Class Method Details

.instance ⇒ Object 



8
9
10
11
12
13
 
# File 'lib/honest_pubsub/publisher.rb', line 8

def self.instance
  if @@publisher.nil?
    @@publisher = ::HonestPubsub::Publisher.new
  end
  @@publisher
end

Instance Method Details

#enable(value) ⇒ Object 



25
26
27
28
 
# File 'lib/honest_pubsub/publisher.rb', line 25

def enable(value)
  @disabled = !value
  @disabled
end

#publish(context, key, payload, enabled = true) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/honest_pubsub/publisher.rb', line 57

def publish(context, key, payload, enabled = true)
  routing_key = "#{@exchange}.#{key}"
  envelope = ::HonestPubsub::Message.new.serialize(context, key, payload)

  if @publisher.nil?
    start
  end

  if @disabled || @publisher.nil? || !enabled
    @logger.failed_publish(routing_key, {}, envelope)
  else
    tries = 2
    begin
      @publisher.publish(envelope.to_json, :persistent=>true, :mandatory=>true, :timestamp=>envelope[:ts], :content_type=>"application/json", :routing_key =>routing_key )
      @logger.log_publish(routing_key, envelope)
    rescue => e
      tries -= 1
      teardown
      start
      if tries > 0 && @publisher
        retry
      else
        @logger.failed_publish(routing_key, {}, envelope)
      end
    end

  end

end

#start ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/honest_pubsub/publisher.rb', line 30

def start
  if !@config[:enabled].nil? && @config[:enabled] == false
    @disabled = true
    return
  end

  # grab server configuration from initialization file somewhere
  begin
    @connection = Bunny.new(Configuration.configuration[:connection])
    @connection.start

    @channel = @connection.create_channel
    @publisher = @channel.topic(@exchange, :durable=>true, :auto_delete=>false)

  rescue => e
    Airbrake.notify(e, parameters: {message: e.message}, environment_name: ENV['RAILS_ENV'] )
    return
  end


  @publisher.on_return do |return_info, properties, content|
    # contents are already transformed into message that we want to send
    @logger.failed_publish(return_info[:routing_key], properties, content)
  end

end

#teardown ⇒ Object 



87
88
89
90
 
# File 'lib/honest_pubsub/publisher.rb', line 87

def teardown
  @connection.close
  @publisher = nil
end