Class: GBDispatch::Queue

Inherits:
Object
  • Object
show all
Includes:
Concurrent::Async
Defined in:
lib/gb_dispatch/queue.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name) ⇒ Queue

Returns a new instance of Queue.

Parameters:

  • name (String) —

    queue name, should be the same as is register in Celluloid



10
11
12
13
 
# File 'lib/gb_dispatch/queue.rb', line 10

def initialize(name)
  super()
  @name = name
end

Instance Attribute Details

#name ⇒ String (readonly)

Returns queue name.

Returns:

  • (String) —

    queue name



7
8
9
 
# File 'lib/gb_dispatch/queue.rb', line 7

def name
  @name
end

Instance Method Details

#perform_after(time, block = nil) { ... } ⇒ Concurrent::ScheduledTask

Perform block after given period

Parameters:

  • time (Fixnum)
  • block (Proc) (defaults to: nil)

Yields:

  • if there is no block given it yield without param.

Returns:

  • (Concurrent::ScheduledTask) 


42
43
44
45
46
47
48
49
50
 
# File 'lib/gb_dispatch/queue.rb', line 42

def perform_after(time, block=nil)
  task = Concurrent::ScheduledTask.new(time) do
    self.async.perform_now do
      block ? block.call : yield
    end
  end
  task.execute
  task
end

#perform_now(block = nil) { ... } ⇒ Object, Exception

Perform given block

If used with rails it will wrap block with connection pool.

Parameters:

  • block (Proc) (defaults to: nil)

Yields:

  • if there is no block given it yield without param.

Returns:

  • (Object, Exception) —

    returns value of executed block or exception if block execution failed.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/gb_dispatch/queue.rb', line 21

def perform_now(block=nil)
  Thread.current[:name] ||= name
  thread_block = ->() do
    with_rails_executor do
      with_connection_pool do
        block ? block.call : yield
      end
    end
  end
  begin
    Runner.execute thread_block, name: name
  rescue Exception => e
    return e
  end
end

#to_s ⇒ Object 



52
53
54
 
# File 'lib/gb_dispatch/queue.rb', line 52

def to_s
  self.name.to_s
end