Class: WatchmonkeyCli::Requeue

Inherits:
Object
  • Object
show all
Defined in:
lib/watchmonkey_cli/hooks/requeue.rb

Class Method Summary collapse

Class Method Details

.hook!(app) ⇒ Object 



3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
56
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
86
87
88
89
 
# File 'lib/watchmonkey_cli/hooks/requeue.rb', line 3

def self.hook!(app)
  app.instance_eval do
    # app options
    @opts[:loop_forever] = true
    @opts[:logfile] = logger_filename # enable logging

    # module options
    @opts[:default_requeue]                   = 60
    # @opts[:default_requeue_ftp_availability]  = 60
    @opts[:default_requeue_mysql_replication] = 30
    @opts[:default_requeue_ssl_expiration]    = 1.hour
    @opts[:default_requeue_unix_defaults]     = false
    # @opts[:default_requeue_unix_df]           = 60
    # @opts[:default_requeue_unix_file_exists]  = 60
    # @opts[:default_requeue_unix_load]         = 60
    @opts[:default_requeue_unix_mdadm]        = 5.minutes
    # @opts[:default_requeue_unix_memory]       = 60
    @opts[:default_requeue_www_availability]  = 30

    # Requeue threads
    @requeue = []


    # =================
    # = Status thread =
    # =================
    @requeue_status_thread = Thread.new do
      Thread.current.abort_on_exception = true
      while STDIN.gets
        sync do
          puts "==========  STATUS  =========="
          puts "     Queue: #{@queue.length}"
          puts "   Requeue: #{@requeue.length}"
          puts "   Workers: #{@threads.select{|t| t[:working] }.length}/#{@threads.length} working (#{@threads.select(&:alive?).length} alive)"
          puts "   Threads: #{Thread.list.length}"
          # puts "            #{@threads.select(&:alive?).length} alive"
          # puts "            #{@threads.select{|t| t.status == "run" }.length} running"
          # puts "            #{@threads.select{|t| t.status == "sleep" }.length} sleeping"
          puts " Processed: #{@processed}"
          puts "========== //STATUS =========="
        end
      end
    end


    # =========
    # = Hooks =
    # =========
    hook :dequeue do |checker, args|
      opts = args.extract_options!
      retry_in = opts[:every] if opts[:every].is_a?(Fixnum)
      retry_in = @opts[:"default_requeue_#{checker.class.checker_name}"] if retry_in.nil?
      retry_in = @opts[:default_requeue] if retry_in.nil?
      if retry_in
        debug "Requeuing #{checker} in #{retry_in} seconds"
        requeue checker, args + [opts], retry_in
      end
    end

    hook :wm_shutdown do
      sync do
        debug "[ReQ] #{@requeue.length} items in requeue..."
        unless @requeue.empty?
          @requeue.each(&:kill).each(&:join).select!(&:alive?)
          debug "[ReQ] #{@requeue.length} items in requeue..."
        end
        @requeue_status_thread.try(:kill).try(:join)
      end
    end


    # ===========
    # = Methods =
    # ===========
    def requeue checker, args, delay = 10
      return if $wm_runtime_exiting
      sync do
        @requeue << Thread.new {
          Thread.current.abort_on_exception = true
          sleep(delay)
          checker.enqueue(*args)
          sync { @requeue.delete Thread.current }
        }
      end
    end
  end
end