Module: Insque

Defined in:
lib/insque.rb,
lib/insque/railtie.rb,
lib/insque/version.rb,
lib/generators/insque/initializer_generator.rb

Defined Under Namespace

Modules: Generators Classes: Railtie

Constant Summary collapse

VERSION = 
"0.4.3"

Class Method Summary collapse

Class Method Details

.broadcast(message, params = nil, recipient = :any) ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/insque.rb', line 23

def self.broadcast message, params = nil, recipient = :any
  keys = []
  case recipient
  when :any
    keys = @redis.smembers 'insque_inboxes'
  when :self
    keys = [@inbox]
  else
    keys = recipient.is_a?(Array) ? recipient : [recipient]
  end
  value = { :message => "#{@sender}_#{message}", :params => params, :broadcasted_at => Time.now.utc }
  log "SENDING: #{value.to_json} TO #{keys.to_json}" if @debug
  @redis.multi do |r|
    keys.each {|k| r.lpush k, value.to_json}
  end
end

.debug=(debug) ⇒ Object 



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

def self.debug= debug
  @debug = debug
end

.janitor ⇒ Object 



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
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/insque.rb', line 61

def self.janitor
  redis = Redis.new @redis_config
  redis.select 7

  loop do
    redis.watch @processing
    errors = []
    restart = []
    delete = []
    redis.lrange(@processing, 0, -1).each do |m|
      begin
        parsed_message = JSON.parse(m)
        if parsed_message['restarted_at'] && DateTime.parse(parsed_message['restarted_at']) < 1.hour.ago.utc
          errors << parsed_message 
          delete << m
        elsif DateTime.parse(parsed_message['broadcasted_at']) < 1.hour.ago.utc
          restart << parsed_message.merge(:restarted_at => Time.now.utc)
          delete << m
        end
      rescue
        log "========== JANITOR_BROKEN_MESSAGE: #{m} =========="
      end
    end
    result = redis.multi do |r|
      restart.each {|m| r.lpush @inbox, m.to_json }
      delete.each {|m| r.lrem @processing, 0, m }
    end
    if result
      errors.each {|m| log "ERROR: #{m.to_json}" }
      restart.each {|m| log "RESTART: #{m.to_json}" }
      log "CLEANING SUCCESSFULL"
    else
      log "CLEANING FAILED"
    end
    sleep(Random.rand * 300)
  end
end

.listen(worker_name = '') ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/insque.rb', line 40

def self.listen worker_name=''
  redis = Redis.new @redis_config
  redis.select 7

  redis.sadd 'insque_inboxes', @inbox
  log "#{worker_name} START LISTENING #{@inbox}"
  loop do
    message = redis.brpoplpush(@inbox, @processing, 0)
    log "#{worker_name} RECEIVING: #{message}" if @debug
    begin
      parsed_message = JSON.parse message
      send(parsed_message['message'], parsed_message) if self.respond_to? parsed_message['message']
    rescue => e
      log "#{worker_name} ========== BROKEN_MESSAGE: #{message} =========="
      log e.inspect
      log e.backtrace
    end
    redis.lrem @processing, 0, message
  end
end

.redis_config=(redis) ⇒ Object 



10
11
12
13
14
 
# File 'lib/insque.rb', line 10

def self.redis_config= redis
  @redis_config = redis
  @redis = Redis.new @redis_config
  @redis.select 7
end

.sender=(sender) ⇒ Object 



16
17
18
19
20
21
 
# File 'lib/insque.rb', line 16

def self.sender= sender
  @sender = sender
  @inbox = "insque_inbox_#{sender}"
  @processing = "insque_processing_#{sender}"
  create_send_later_handler
end