Class: Krakow::Producer

Inherits:

Object

• Object
• Krakow::Producer

show all

Includes:

Celluloid, Utils::Lazy

Defined in:

lib/krakow/producer.rb,
lib/krakow/producer/http.rb

Overview

Defined Under Namespace

Classes: Http

Instance Attribute Summary (collapse)

• - (Object) connection readonly

  Returns the value of attribute connection.

Attributes (collapse)

• - (Hash) connection_options

  The connection_options attribute.

• - (TrueClass, FalseClass) connection_options?

  Truthiness of the connection_options attribute.

• - (String) host

  The host attribute.

• - (TrueClass, FalseClass) host?

  Truthiness of the host attribute.

• - ([String, Integer]) port

  The port attribute.

• - (TrueClass, FalseClass) port?

  Truthiness of the port attribute.

• - (Integer) reconnect_interval

  The reconnect_interval attribute.

• - (TrueClass, FalseClass) reconnect_interval?

  Truthiness of the reconnect_interval attribute.

• - (Integer) reconnect_retries

  The reconnect_retries attribute.

• - (TrueClass, FalseClass) reconnect_retries?

  Truthiness of the reconnect_retries attribute.

• - (String) topic

  The topic attribute.

• - (TrueClass, FalseClass) topic?

  Truthiness of the topic attribute.

Instance Method Summary (collapse)

• - (Object) connect

  Establish connection to configured `host` and `port`.

• - (TrueClass, FalseClass) connected?

  Currently connected to server.

• - (TrueClass) connection_failure(*args)

  Process connection failure and attempt reconnection.

• - (Object) goodbye_my_love!

  Instance destructor.

• - (Producer) initialize(args = {}) constructor

  A new instance of Producer.

• - (Logger^?) log(*args) included from Utils::Logging

  Log message.

• - (String) to_s

  Stringify object.

• - (Krakow::FrameType::Error^?) write(*message)

  Write message to server.

37
38
39
40
41
42
43

# File 'lib/krakow/producer.rb', line 37

def initialize(args={})
  super
  arguments[:connection_options] = {:features => {}, :config => {}}.merge(
    arguments.fetch(:connection_options, {})
  )
  connect
end

19
20
21

# File 'lib/krakow/producer.rb', line 19

def connection
  @connection
end

48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

# File 'lib/krakow/producer.rb', line 48

def connect
  info "Establishing connection to: #{host}:#{port}"
  begin
    @connection = Connection.new(
      :host => host,
      :port => port,
      :features => connection_options[:features],
      :features_args => connection_options[:config]
    )
    connection.init!
    self.link connection
    info "Connection established: #{connection}"
    nil
  rescue => e
    abort e
  end
end

72
73
74

# File 'lib/krakow/producer.rb', line 72

def connected?
  !!(connection && connection.alive? && connection.connected?)
end

79
80
81
82
83
84
85
86
87
88
89
90

# File 'lib/krakow/producer.rb', line 79

def connection_failure(*args)
  @connection = nil
  begin
    warn "Connection failure detected for #{host}:#{port}"
    connect
  rescue => e
    warn "Failed to establish connection to #{host}:#{port}. Pausing #{reconnect_interval} before retry"
    sleep reconnect_interval
    connect
  end
  true
end

33

# File 'lib/krakow/producer.rb', line 33

attribute :connection_options, Hash, :default => ->{ Hash.new }

33

# File 'lib/krakow/producer.rb', line 33

attribute :connection_options, Hash, :default => ->{ Hash.new }

94
95
96
97
98
99
100
101
102

# File 'lib/krakow/producer.rb', line 94

def goodbye_my_love!
  debug 'Tearing down producer'
  if(connection && connection.alive?)
    connection.terminate
  end
  @connection = nil
  info 'Producer torn down'
  nil
end

28

# File 'lib/krakow/producer.rb', line 28

attribute :host, String, :required => true

28

# File 'lib/krakow/producer.rb', line 28

attribute :host, String, :required => true

29

# File 'lib/krakow/producer.rb', line 29

attribute :port, [String, Integer], :required => true

29

# File 'lib/krakow/producer.rb', line 29

attribute :port, [String, Integer], :required => true

32

# File 'lib/krakow/producer.rb', line 32

attribute :reconnect_interval, Integer, :default => 5

32

# File 'lib/krakow/producer.rb', line 32

attribute :reconnect_interval, Integer, :default => 5

31

# File 'lib/krakow/producer.rb', line 31

attribute :reconnect_retries, Integer, :default => 10

31

# File 'lib/krakow/producer.rb', line 31

attribute :reconnect_retries, Integer, :default => 10

67
68
69

# File 'lib/krakow/producer.rb', line 67

def to_s
  "<#{self.class.name}:#{object_id} {#{host}:#{port}} T:#{topic}>"
end

30

# File 'lib/krakow/producer.rb', line 30

attribute :topic, String, :required => true

30

# File 'lib/krakow/producer.rb', line 30

attribute :topic, String, :required => true

109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

# File 'lib/krakow/producer.rb', line 109

def write(*message)
  if(message.empty?)
    abort ArgumentError.new 'Expecting one or more messages to send. None provided.'
  end
  if(connection && connection.alive?)
    if(message.size > 1)
      debug 'Multiple message publish'
      connection.transmit(
        Command::Mpub.new(
          :topic_name => topic,
          :messages => message
        )
      )
    else
      debug 'Single message publish'
      connection.transmit(
        Command::Pub.new(
          :message => message.first,
          :topic_name => topic
        )
      )
    end
  else
    abort Error::ConnectionUnavailable.new 'Remote connection is unavailable!'
  end
end