Class: EventHub::Processor
- Inherits:
-
Object
- Object
- EventHub::Processor
show all
- Includes:
- Helper
- Defined in:
- lib/eventhub/processor.rb
Instance Attribute Summary collapse
Instance Method Summary
collapse
Methods included from Helper
#class_to_array, #duration, #format_string, #now_stamp
Constructor Details
#initialize(name = nil) ⇒ Processor
Returns a new instance of Processor.
11
12
13
14
15
16
17
18
19
20
21
22
|
# File 'lib/eventhub/processor.rb', line 11
def initialize(name=nil)
@name = name || class_to_array(self.class)[1..-1].join('.')
@pidfile = EventHub::Components::Pidfile.new(File.join(Dir.pwd, 'pids', "#{@name}.pid"))
@exception_writer = EventHub::Components::ExceptionWriter.new
@statistics = EventHub::Statistics.new
@heartbeat = EventHub::Heartbeat.new(self)
@message_processor = EventHub::MessageProcessor.new(self)
@channel_receiver = nil
@channel_sender = nil
@restart = true
end
|
Instance Attribute Details
#exception_writer ⇒ Object
Returns the value of attribute exception_writer.
3
4
5
|
# File 'lib/eventhub/processor.rb', line 3
def exception_writer
@exception_writer
end
|
#name ⇒ Object
Returns the value of attribute name.
3
4
5
|
# File 'lib/eventhub/processor.rb', line 3
def name
@name
end
|
#pidfile ⇒ Object
Returns the value of attribute pidfile.
3
4
5
|
# File 'lib/eventhub/processor.rb', line 3
def pidfile
@pidfile
end
|
#statistics ⇒ Object
Returns the value of attribute statistics.
3
4
5
|
# File 'lib/eventhub/processor.rb', line 3
def statistics
@statistics
end
|
Instance Method Details
#configuration ⇒ Object
24
25
26
|
# File 'lib/eventhub/processor.rb', line 24
def configuration
EventHub::Configuration.instance.data
end
|
#connection_settings ⇒ Object
48
49
50
|
# File 'lib/eventhub/processor.rb', line 48
def connection_settings
{ user: server_user, password: server_password, host: server_host, vhost: server_vhost }
end
|
#handle_message(metadata, payload) ⇒ Object
103
104
105
|
# File 'lib/eventhub/processor.rb', line 103
def handle_message(metadata, payload)
raise 'Please implement method in derived class'
end
|
#heartbeat_cycle_in_s ⇒ Object
68
69
70
|
# File 'lib/eventhub/processor.rb', line 68
def heartbeat_cycle_in_s
configuration.get('processor.heartbeat_cycle_in_s') || 300
end
|
#listener_queues ⇒ Object
52
53
54
55
56
57
58
|
# File 'lib/eventhub/processor.rb', line 52
def listener_queues
Array(
configuration.get('processor.listener_queue') ||
configuration.get('processor.listener_queues') ||
'undefined_listener_queues'
)
end
|
#restart_in_s ⇒ Object
64
65
66
|
# File 'lib/eventhub/processor.rb', line 64
def restart_in_s
configuration.get('processor.restart_in_s') || 15
end
|
#send_message(message, exchange_name = EventHub::EH_X_INBOUND) ⇒ Object
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
# File 'lib/eventhub/processor.rb', line 134
def send_message(message, exchange_name = EventHub::EH_X_INBOUND)
if @channel_sender.nil? || !@channel_sender.open?
@channel_sender = AMQP::Channel.new(@connection, prefetch: 1)
@channel_sender.confirm_select
end
exchange = @channel_sender.direct(exchange_name, :durable => true, :auto_delete => false)
exchange.publish(message.to_json, :persistent => true)
end
|
#server_host ⇒ Object
28
29
30
|
# File 'lib/eventhub/processor.rb', line 28
def server_host
configuration.get('server.host') || 'localhost'
end
|
#server_management_port ⇒ Object
40
41
42
|
# File 'lib/eventhub/processor.rb', line 40
def server_management_port
configuration.get('server.management_port') || 15672
end
|
#server_password ⇒ Object
36
37
38
|
# File 'lib/eventhub/processor.rb', line 36
def server_password
configuration.get('server.password') || 'admin'
end
|
#server_user ⇒ Object
32
33
34
|
# File 'lib/eventhub/processor.rb', line 32
def server_user
configuration.get('server.user') || 'admin'
end
|
#server_vhost ⇒ Object
44
45
46
|
# File 'lib/eventhub/processor.rb', line 44
def server_vhost
configuration.get('server.vhost') || 'event_hub'
end
|
#sleep_break(seconds) ⇒ Object
breaks after n seconds or after interrupt
150
151
152
153
154
155
156
|
# File 'lib/eventhub/processor.rb', line 150
def sleep_break(seconds) while (seconds > 0)
sleep(1)
seconds -= 1
break unless @restart
end
end
|
#start(detached = false) ⇒ Object
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
# File 'lib/eventhub/processor.rb', line 72
def start(detached = false)
daemonize if detached
EventHub.logger.info("Processor [#{@name}] base folder [#{Dir.pwd}]")
Signal.trap('TERM') { EventMachine.add_timer(0) { about_to_stop } }
Signal.trap('INT') { EventMachine.add_timer(0) { about_to_stop } }
while @restart
begin
handle_start_internal
post_start
rescue => e
id = @exception_writer.write(e)
EventHub.logger.error("Unexpected exception: #{e}, see => #{id}. Trying to restart in #{self.restart_in_s} seconds...")
sleep_break self.restart_in_s
end
end
post_stop
EventHub.logger.info("Processor [#{@name}] has been stopped")
ensure
pidfile.delete
end
|
#version ⇒ Object
7
8
9
|
# File 'lib/eventhub/processor.rb', line 7
def version
'1.0.0'
end
|
#watchdog ⇒ Object
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
# File 'lib/eventhub/processor.rb', line 107
def watchdog
self.listener_queues.each do |queue_name|
begin
response = RestClient.get "http://#{self.server_user}:#{self.server_password}@#{self.server_host}:#{self.server_management_port}/api/queues/#{self.server_vhost}/#{queue_name}/bindings", { :content_type => :json}
data = JSON.parse(response.body)
if response.code != 200
EventHub.logger.warn("Watchdog: Server did not answered properly. Trying to restart in #{self.restart_in_s} seconds...")
EventMachine.add_timer(self.restart_in_s) { stop_processor(true) }
elsif data.size == 0
EventHub.logger.warn("Watchdog: Something is wrong with the vhost, queue [#{queue_name}], and/or bindings. Trying to restart in #{self.restart_in_s} seconds...")
EventMachine.add_timer(self.restart_in_s) { stop_processor(true) }
else
EventMachine.add_timer(self.watchdog_cycle_in_s) { watchdog }
end
rescue => e
EventHub.logger.error("Watchdog: Unexpected exception: #{e}. Trying to restart in #{self.restart_in_s} seconds...")
stop_processor
end
end
end
|
#watchdog_cycle_in_s ⇒ Object
60
61
62
|
# File 'lib/eventhub/processor.rb', line 60
def watchdog_cycle_in_s
configuration.get('processor.watchdog_cycle_is_s') || 15
end
|