Class: Temporalio::Worker::ActivityExecutor::Fiber

Inherits:
Object
  • Object
show all
Defined in:
lib/temporalio/worker/activity_executor/fiber.rb

Overview

Activity executor for scheduling activites as fibers.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.default ⇒ Fiber

Returns Default/shared Fiber executor instance.

Returns:

  • (Fiber) —

    Default/shared Fiber executor instance.



12
13
14
 
# File 'lib/temporalio/worker/activity_executor/fiber.rb', line 12

def self.default
  @default ||= new
end

Instance Method Details

#activity_context ⇒ Object

See Also:



31
32
33
 
# File 'lib/temporalio/worker/activity_executor/fiber.rb', line 31

def activity_context
  ::Fiber[:temporal_activity_context]
end

#execute_activity(defn) ⇒ Object

See Also:



26
27
28
 
# File 'lib/temporalio/worker/activity_executor/fiber.rb', line 26

def execute_activity(defn, &) # rubocop:disable Lint/UnusedMethodArgument
  ::Fiber.schedule(&)
end

#initialize_activity(defn) ⇒ Object

Raises:

  • (ArgumentError)

See Also:



17
18
19
20
21
22
23
 
# File 'lib/temporalio/worker/activity_executor/fiber.rb', line 17

def initialize_activity(defn)
  # If there is not a current scheduler, we're going to preemptively
  # fail the registration
  return unless ::Fiber.current_scheduler.nil?

  raise ArgumentError, "Activity '#{defn.name}' wants a fiber executor but no current fiber scheduler"
end

#set_activity_context(defn, context) ⇒ Object

See Also:



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/temporalio/worker/activity_executor/fiber.rb', line 36

def set_activity_context(defn, context)
  ::Fiber[:temporal_activity_context] = context
  # If they have opted in to raising on cancel, wire that up
  return unless defn.cancel_raise

  fiber = ::Fiber.current
  scheduler = ::Fiber.scheduler
  scheduler = nil unless scheduler.respond_to?(:fiber_interrupt)
  context&.cancellation&.add_cancel_callback do
    error = Error::CanceledError.new('Activity canceled')
    # Directly raising from another fiber can strand a `Fiber#transfer`
    # based scheduler's current fiber, so we defer to the scheduler to interrupt.
    # If on the same fiber, we can just raise directly.
    if scheduler.nil? || ::Fiber.current.equal?(fiber)
      fiber.raise(error)
    else
      scheduler.fiber_interrupt(fiber, error)
    end
  end
end