Class: Temporalio::Client::ActivityHandle

Inherits:
Object
  • Object
show all
Defined in:
lib/temporalio/client/activity_handle.rb

Overview

Handle for interacting with a standalone activity. Usually created via #activity_handle or #start_activity.

WARNING: Standalone Activities are experimental.

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#id ⇒ String (readonly)

Returns ID for the activity.

Returns:

  • (String) —

    ID for the activity.



21
22
23
 
# File 'lib/temporalio/client/activity_handle.rb', line 21

def id
  @id
end

#result_hint ⇒ Object? (readonly)

Returns Result hint used when deserializing the activity's result. May be overridden per #result call.

Returns:

  • (Object, nil) —

    Result hint used when deserializing the activity's result. May be overridden per #result call.



28
29
30
 
# File 'lib/temporalio/client/activity_handle.rb', line 28

def result_hint
  @result_hint
end

#run_id ⇒ String? (readonly)

Returns Run ID for this activity execution. When nil, this handle targets the latest run.

Returns:

  • (String, nil) —

    Run ID for this activity execution. When nil, this handle targets the latest run.



24
25
26
 
# File 'lib/temporalio/client/activity_handle.rb', line 24

def run_id
  @run_id
end

Instance Method Details

#cancel(reason = nil, rpc_options: nil) ⇒ Object

Request cancellation of the activity.

Parameters:

  • reason (String, nil) (defaults to: nil) —

    Optional cancellation reason recorded on the server.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Raises:



102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/temporalio/client/activity_handle.rb', line 102

def cancel(reason = nil, rpc_options: nil)
  @client._impl.cancel_activity(
    Interceptor::CancelActivityInput.new(
      activity_id: id,
      activity_run_id: run_id,
      reason:,
      rpc_options:
    )
  )
  nil
end

#describe(include_input: false, include_outcome: false, include_heartbeat_details: false, include_last_failure: false, rpc_options: nil) ⇒ ActivityExecution::Description

Describe the activity.

The payload-bearing fields are opt-in because they can be arbitrarily large; request them only when needed. Each has a corresponding predicate on the returned description that reports whether the server supplied it.

Parameters:

  • include_input (Boolean) (defaults to: false) —

    If true and the activity received input, include the input.

  • include_outcome (Boolean) (defaults to: false) —

    If true and the activity is closed, include the outcome.

  • include_heartbeat_details (Boolean) (defaults to: false) —

    If true and the activity recorded heartbeat details, include them.

  • include_last_failure (Boolean) (defaults to: false) —

    If true and the activity has a failed attempt, include the last failure.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Returns:

Raises:



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/temporalio/client/activity_handle.rb', line 77

def describe(
  include_input: false,
  include_outcome: false,
  include_heartbeat_details: false,
  include_last_failure: false,
  rpc_options: nil
)
  @client._impl.describe_activity(
    Interceptor::DescribeActivityInput.new(
      activity_id: id,
      activity_run_id: run_id,
      include_input:,
      include_outcome:,
      include_heartbeat_details:,
      include_last_failure:,
      rpc_options:
    )
  )
end

#pause(reason = nil, rpc_options: nil) ⇒ Object

Pause the activity. A paused activity is not scheduled or retried until it is unpaused via #unpause.

WARNING: Standalone Activities are experimental.

Parameters:

  • reason (String, nil) (defaults to: nil) —

    Optional reason recorded on the server.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Raises:



139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/temporalio/client/activity_handle.rb', line 139

def pause(reason = nil, rpc_options: nil)
  @client._impl.pause_activity(
    Interceptor::PauseActivityInput.new(
      activity_id: id,
      activity_run_id: run_id,
      reason:,
      rpc_options:
    )
  )
  nil
end

#result(result_hint: nil, rpc_options: nil) ⇒ Object?

Wait for the activity's outcome (result or failure). Internally long-polls PollActivityExecution and reissues until the activity reaches a terminal state, so this can block indefinitely for long-running activities.

Parameters:

  • result_hint (Object, nil) (defaults to: nil) —

    Override the result hint. If nil, uses #result_hint.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Returns:

  • (Object, nil) —

    Deserialized activity result.

Raises:



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/temporalio/client/activity_handle.rb', line 49

def result(result_hint: nil, rpc_options: nil)
  hint = result_hint || @result_hint
  outcome = @client._impl.fetch_activity_outcome(
    Interceptor::FetchActivityOutcomeInput.new(
      activity_id: id,
      activity_run_id: run_id,
      rpc_options:
    )
  )
  _process_outcome(outcome, hint)
end

#terminate(reason = nil, rpc_options: nil) ⇒ Object

Terminate the activity (force-close).

Parameters:

  • reason (String, nil) (defaults to: nil) —

    Optional termination reason recorded on the activity's failure outcome.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Raises:



119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/temporalio/client/activity_handle.rb', line 119

def terminate(reason = nil, rpc_options: nil)
  @client._impl.terminate_activity(
    Interceptor::TerminateActivityInput.new(
      activity_id: id,
      activity_run_id: run_id,
      reason:,
      rpc_options:
    )
  )
  nil
end

#unpause(reason: nil, jitter: nil, rpc_options: nil) ⇒ Object

Unpause the activity, allowing it to be scheduled or retried again.

WARNING: Standalone Activities are experimental.

Parameters:

  • reason (String, nil) (defaults to: nil) —

    Optional reason recorded on the server.

  • jitter (Float, nil) (defaults to: nil) —

    If set, the activity will start at a random time within this duration (in seconds).

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Raises:



160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/temporalio/client/activity_handle.rb', line 160

def unpause(reason: nil, jitter: nil, rpc_options: nil)
  @client._impl.unpause_activity(
    Interceptor::UnpauseActivityInput.new(
      activity_id: id,
      activity_run_id: run_id,
      reason:,
      jitter:,
      rpc_options:
    )
  )
  nil
end

#update_options(*updates, restore_original: false, rpc_options: nil) ⇒ ActivityExecutionOptions

Update the activity's options. Only the options named by updates are changed; anything not named is left as-is.

Updates are created from the keys on Temporalio::Client::ActivityOptions, via Temporalio::Client::ActivityOptions::Key#value_set to set an option or Temporalio::Client::ActivityOptions::Key#value_unset to clear it.

WARNING: Standalone Activities are experimental.

Parameters:

  • updates (Array<ActivityOptions::Update>) —

    The option updates to apply. At least one is required unless restore_original is true.

  • restore_original (Boolean) (defaults to: false) —

    If true, restore the options to the originals the activity was created with. Mutually exclusive with any update.

  • rpc_options (RPCOptions, nil) (defaults to: nil) —

    Advanced RPC options.

Returns:

Raises:

  • (ArgumentError) —

    If a non-update is given, if restore_original is combined with any update, or if no update is provided and restore_original is false.

  • (Error::RPCError) —

    RPC error from call.



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/temporalio/client/activity_handle.rb', line 192

def update_options(*updates, restore_original: false, rpc_options: nil)
  unless updates.all?(ActivityOptions::Update)
    raise ArgumentError,
          'Updates must be created via ActivityOptions::Key#value_set or #value_unset'
  end

  if restore_original && !updates.empty?
    raise ArgumentError, 'restore_original cannot be combined with any option update'
  elsif !restore_original && updates.empty?
    raise ArgumentError,
          'At least one option update must be given, or restore_original must be used'
  end

  # For repeated keys, later values override previous ones.
  by_path = updates.to_h { |update| [update.key.name, update] }

  proto = Api::Activity::V1::ActivityOptions.new
  by_path.each_value do |update|
    # An unset update names its path but leaves the field absent, which is how the server is
    # told to clear the option rather than set it to a value.
    update.key._apply(proto, update.value) unless update.value.nil?
  end

  @client._impl.update_activity_options(
    Interceptor::UpdateActivityOptionsInput.new(
      activity_id: id,
      activity_run_id: run_id,
      activity_options: proto,
      update_mask: Google::Protobuf::FieldMask.new(paths: by_path.keys),
      restore_original:,
      rpc_options:
    )
  )
end