Class: ConditionalWait

Inherits:

Object

• Object
• ConditionalWait

Includes:

ThreadImmutable

Defined in:

lib/wakame/util.rb

Defined Under Namespace

Classes: TimeoutError

Class Method Summary

• .wait(timeout = 60*30, &blk) ⇒ Object

Instance Method Summary

• #initialize(&blk) ⇒ ConditionalWait constructor

  A new instance of ConditionalWait.

• #poll(period = 5, max_retry = 10, &blk) ⇒ Object
• #wait(tout = nil) ⇒ Object
• #wait_event(event_class, &blk) ⇒ Object

Methods included from ThreadImmutable

#bind_thread, included, #target_thread, #target_thread?, #thread_check

Constructor Details

#initialize(&blk) ⇒ ConditionalWait

Returns a new instance of ConditionalWait.

# File 'lib/wakame/util.rb', line 459

def initialize(&blk)
  bind_thread
  @wait_queue = ::Queue.new
  @wait_tickets = []
  @poll_threads = []
  @event_tickets = []

  blk.call(self) if blk
end

Class Method Details

.wait(timeout = 60*30, &blk) ⇒ Object

# File 'lib/wakame/util.rb', line 559

def self.wait(timeout=60*30, &blk)
  cond = ConditionalWait.new
  cond.bind_thread(Thread.current)
  
  #cond.instance_eval(&blk)
  blk.call(cond)
  
  cond.wait(timeout)
end

Instance Method Details

#poll(period = 5, max_retry = 10, &blk) ⇒ Object

# File 'lib/wakame/util.rb', line 469

def poll( period=5, max_retry=10, &blk)
  wticket = Wakame::Util.gen_id
  @poll_threads << Thread.new {
    retry_count = 0

    begin
      catch(:finish) {
        while retry_count < max_retry
          start_at = Time.now
          if blk.call == true
            throw :finish
          end
          Thread.pass
          if period > 0
            t = Time.now - start_at
            sleep (period - t) if period > t
          end
          retry_count += 1
        end
      }
      
      if retry_count >= max_retry
        Wakame.log.error('Over retry count')
        raise 'Over retry count'
      end

    rescue => e
      Wakame.log.error(e)
      @wait_queue << [false, wticket, e]
    else
      @wait_queue << [true, wticket]
    end
  }
  @poll_threads.last[:name]="#{self.class} poll"

  @wait_tickets << wticket
end

#wait(tout = nil) ⇒ Object

# File 'lib/wakame/util.rb', line 529

def wait(tout=nil)

  unless @wait_tickets.empty?
    Wakame.log.debug("#{self.class} waits for #{@wait_tickets.size} num of event(s)/polling(s).")

    timeout(tout, TimeoutError) {
      while @wait_tickets.size > 0 && q = @wait_queue.shift
        @wait_tickets.delete(q[1])
        
        unless q[0]
          Wakame.log.debug("#{q[1]} failed with #{q[2]}")
          raise q[2]
        end
      end
    }
  end
  
ensure
  # Cleanup generated threads/event tickets
  @poll_threads.each { |t|
    begin
      t.kill
    rescue => e
      Wakame.log.error(e)
    end
  }
  @event_tickets.each { |t| Wakame::EventDispatcher.unsubscribe(t) }
end

#wait_event(event_class, &blk) ⇒ Object

# File 'lib/wakame/util.rb', line 508

def wait_event(event_class, &blk)
  wticket = Wakame::Util.gen_id
  Wakame.log.debug("#{self.class} called wait_event(#{event_class}) on thread #{Thread.current} (target_thread=#{self.target_thread?}). has_blk=#{blk}")
  ticket = Wakame::EventDispatcher.subscribe(event_class) { |event|
    begin
      if blk.call(event) == true
        Wakame::EventDispatcher.unsubscribe(ticket)
        @wait_queue << [true, wticket]
      end
    rescue => e
      Wakame.log.error(e)
      Wakame::EventDispatcher.unsubscribe(ticket)
      @wait_queue << [false, wticket, e]
    end
  }
  @event_tickets << ticket

  @wait_tickets << wticket
end