Class: PgEventstore::AsyncRunner

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_eventstore/async_runner.rb,
sig/pg_eventstore/async_runner.rbs

Defined Under Namespace

Classes: Cancellation

Instance Method Summary collapse

Constructor Details

#initialize ⇒ AsyncRunner

Returns a new instance of AsyncRunner.



8
9
10
 
# File 'lib/pg_eventstore/async_runner.rb', line 8

def initialize
  @jobs = {}
end

Instance Method Details

#async { ... } ⇒ void

This method returns an undefined value.

Yields:

Yield Returns:

  • (void) 


59
60
61
 
# File 'lib/pg_eventstore/async_runner.rb', line 59

def async(&)
  @jobs[Fiber.new(&)] = false
end

#jobs_size ⇒ Integer

Returns:

  • (Integer) 


13
14
15
 
# File 'lib/pg_eventstore/async_runner.rb', line 13

def jobs_size
  @jobs.size
end

#run ⇒ void

This method returns an undefined value.



18
19
20
21
22
23
24
 
# File 'lib/pg_eventstore/async_runner.rb', line 18

def run
  loop do
    break if @jobs.empty?

    run_once
  end
end

#run_once ⇒ void

This method returns an undefined value.



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
 
# File 'lib/pg_eventstore/async_runner.rb', line 26

def run_once
  return if @jobs.empty?

  @jobs.keys.each do |job|
    @jobs[job] = true
    job.resume
    @jobs.delete(job) unless job.alive?
  end
rescue
  @jobs.each do |job, started|
    next unless job.alive?

    if started
      begin
        job.raise(Cancellation)
      rescue StandardError
        # Because the rescue mechanisms inside the terminating job can potentially raise - catch them here
      end
    end

    begin
      # Kill those jobs which survived our Cancellation exception
      job.kill if job.alive?
    rescue StandardError
      # Ensure we handle any errors inside an exception handler(e.g. ensure block) of the given job
    end
  end

  @jobs.clear
  raise
end