Class: RockQueueSmpp::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/rock-queue-smpp/worker.rb

Constant Summary collapse

@@mt_id = 
0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Worker

Initialize connection to queue server



21
22
23
24
25
26
27
28
 
# File 'lib/rock-queue-smpp/worker.rb', line 21

def initialize
  puts "=> Initializing..."
  config = RockQueueSmpp::Config.settings
  @queue = RockQueueSmpp::Base.new config.adapter, {
          :server => config.host,
          :port   => config.port
  }
end

Instance Attribute Details

#verbose ⇒ Object

Whether the worker should log basic info to STDOUT



18
19
20
 
# File 'lib/rock-queue-smpp/worker.rb', line 18

def verbose
  @verbose
end

Class Method Details

.send_mt(*args) ⇒ Object 



30
31
32
33
34
 
# File 'lib/rock-queue-smpp/worker.rb', line 30

def self.send_mt(*args)
  @@mt_id += 1
  puts '=> sending message'
  @@tx.send_mt(@@mt_id, *args)
end

Instance Method Details

#bound(transceiver) ⇒ Object 



86
87
88
 
# File 'lib/rock-queue-smpp/worker.rb', line 86

def bound(transceiver)
  puts "=> Delegate: transceiver bound"
end

#delivery_report_received(transceiver, msg_reference, stat, pdu) ⇒ Object 



78
79
80
 
# File 'lib/rock-queue-smpp/worker.rb', line 78

def delivery_report_received(transceiver, msg_reference, stat, pdu)
  puts "=> Delegate: delivery_report_received: ref #{msg_reference} stat #{stat} pdu #{pdu}"
end

#message_accepted(transceiver, mt_message_id, smsc_message_id) ⇒ Object 



82
83
84
 
# File 'lib/rock-queue-smpp/worker.rb', line 82

def message_accepted(transceiver, mt_message_id, smsc_message_id)
  puts "=> Delegate: message_sent: id #{mt_message_id} smsc ref id: #{smsc_message_id}"
end

#mo_received(transceiver, source_addr, destination_addr, short_message) ⇒ Object 



74
75
76
 
# File 'lib/rock-queue-smpp/worker.rb', line 74

def mo_received(transceiver, source_addr, destination_addr, short_message)
  puts "=> Delegate: mo_received: from #{source_addr} to #{destination_addr}: #{short_message}"
end

#unbound(transceiver) ⇒ Object 



90
91
92
93
 
# File 'lib/rock-queue-smpp/worker.rb', line 90

def unbound(transceiver)
  puts "=> Stopping EM Loop"
  EventMachine::stop_event_loop
end

#work ⇒ Object

Main worker loop where all jobs are beeing pulled of the queue. This is also a place where every job starts and ends it’s lifecycle.



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/rock-queue-smpp/worker.rb', line 38

def work
  config = RockQueueSmpp::Config.settings
  puts "=> Worker ready. Hold your horses!"
  loop do
    EventMachine::run do
      @@tx = EventMachine::connect(
              config.smpp_config[:host],
              config.smpp_config[:port],
              Smpp::Transceiver,
              config.smpp_config,
              self    # delegate that will receive callbacks on MOs and DRs and other events
      )
      @queue.receive do |queue|
        puts "=> In Queue Loop"
        if queue
          # code that actually performs the action
          begin
            puts "=> Processing Queue Item"
            args = queue.args.first
            args.empty? ? queue.object.perform : queue.object.perform(args)
          rescue Object => e
            # Add failed processing and retry
            if queue.add_fail(e)
              sleep(queue.get_sleep_time)
              puts "=> Processing fail! Retrying #{queue.fails.length}"
              retry
            end
          end
        end
      end
      puts "=> Disconnected. Reconnecting in 5 seconds"
      sleep 5
    end
  end
end