-module(temporal_sdk_activity). % elp:ignore W0012 W0040 -moduledoc """ Temporal activity task module. """. -export([ await_data/1, await_data/2, complete/1, cancel/1, fail/1, fail/3, heartbeat/0, heartbeat/1, last_heartbeat/0, cancel_requested/0, activity_paused/0, elapsed_time/0, elapsed_time/1, remaining_time/0, remaining_time/1, get_data/0, set_data/1 ]). -import(temporal_sdk_executor, [ call/1, cast/1 ]). -include("proto.hrl"). -type task() :: ?TEMPORAL_SPEC:'temporal.api.workflowservice.v1.PollActivityTaskQueueResponse'(). -export_type([task/0]). -type context() :: #{ cluster := temporal_sdk_cluster:cluster_name(), executor_pid := pid(), otel_ctx := otel_ctx:t(), task := task(), worker_opts := temporal_sdk_worker:opts(), started_at := SystemTime :: integer(), task_timeout := erlang:timeout(), header => temporal_sdk:term_from_mapstring_payload() }. -export_type([context/0]). -type handler_context() :: #{ data := data(), cancel_requested := boolean(), activity_paused := boolean(), last_heartbeat := heartbeat(), elapsed_time := non_neg_integer(), remaining_time := erlang:timeout() }. -export_type([handler_context/0]). -type data() :: term(). -export_type([data/0]). -type heartbeat() :: temporal_sdk:term_to_payloads(). -export_type([heartbeat/0]). %% ------------------------------------------------------------------------------------------------- %% Activity behaviour -type complete_action() :: {complete, Result :: temporal_sdk:term_to_payloads()}. -export_type([complete_action/0]). -type cancel_action() :: {cancel, CanceledDetails :: temporal_sdk:term_to_payloads()}. -export_type([cancel_action/0]). -type fail_action() :: {fail, { Source :: temporal_sdk:serializable(), Message :: temporal_sdk:serializable(), Stacktrace :: temporal_sdk:serializable() }}. -export_type([fail_action/0]). -type terminate_action() :: cancel_action() | complete_action() | fail_action(). -export_type([terminate_action/0]). -type heartbeat_action() :: heartbeat | {heartbeat, Heartbeat :: heartbeat()}. -export_type([heartbeat_action/0]). -type data_action() :: {data, NewData :: data()}. -export_type([data_action/0]). -callback execute(Context :: context(), Input :: temporal_sdk:term_from_payloads()) -> Result :: temporal_sdk:term_to_payloads(). -callback terminate(HandlerContext :: handler_context()) -> term(). -callback handle_heartbeat(HandlerContext :: handler_context()) -> terminate_action() | heartbeat_action(). -callback handle_cancel(HandlerContext :: handler_context()) -> terminate_action() | ignore. -callback handle_message(HandlerContext :: handler_context(), Message :: term()) -> terminate_action() | data_action() | ignore. -callback handle_failure( HandlerContext :: handler_context(), Class :: error | exit | throw | temporal_sdk:serializable(), Reason :: term() | temporal_sdk:serializable(), Stacktrace :: erlang:raise_stacktrace() | temporal_sdk:serializable() ) -> ApplicationFailure :: temporal_sdk:application_failure() | temporal_sdk:user_application_failure(). -optional_callbacks([ terminate/1, handle_heartbeat/1, handle_cancel/1, handle_message/2, handle_failure/4 ]). %% ------------------------------------------------------------------------------------------------- %% Commands -spec await_data(EtsPattern :: term()) -> {ok, data()} | timeout | invalid_pattern | no_return(). await_data(EtsPattern) -> case remaining_time(millisecond) of infinity -> await_data(EtsPattern, infinity); T -> await_data(EtsPattern, T) end. -spec await_data(EtsPattern :: term(), Timeout :: erlang:timeout()) -> {ok, data()} | timeout | invalid_pattern | no_return(). await_data(EtsPattern, Timeout) -> case temporal_sdk_utils_ets:compile_match_spec(EtsPattern) of {ok, Compiled} -> call({await_data, Compiled, Timeout}); error -> invalid_pattern end. -spec cancel(CanceledDetails :: temporal_sdk:term_to_payloads()) -> no_return(). cancel(CanceledDetails) -> case cancel_requested() of true -> call({cancel, CanceledDetails}); false -> erlang:error( "Unable to cancel activity without activity being request canceled first.", [CanceledDetails] ) end. -spec complete(Result :: temporal_sdk:term_to_payloads()) -> no_return(). complete(Result) -> call({complete, Result}). %% ApplicationFailure is sent immediately. No telemetry exception is generated. %% Can be compared to temporal_sdk_workflow:fail_workflow_execution/1. -spec fail( ApplicationFailure :: temporal_sdk:application_failure() | temporal_sdk:user_application_failure() ) -> no_return(). fail(ApplicationFailure) -> call({fail, ApplicationFailure}). %% {Class, Reason, Stacktrace} exception is translated to ApplicationFailure with %% handle_failure/4 callback or internal default translation and then dispatched to Temporal server. %% {Class, Reason, Stacktrace} telemetry exception is generated. -spec fail( Class :: error | exit | throw | temporal_sdk:serializable(), Reason :: term() | temporal_sdk:serializable(), Stacktrace :: erlang:raise_stacktrace() | temporal_sdk:serializable() ) -> no_return(). fail(Class, Reason, Stacktrace) -> call({fail, Class, Reason, Stacktrace}). -spec heartbeat() -> ok. heartbeat() -> cast({heartbeat}). -spec heartbeat(Heartbeat :: heartbeat()) -> ok. heartbeat(Heartbeat) -> cast({heartbeat, Heartbeat}). -spec cancel_requested() -> boolean() | no_return(). cancel_requested() -> call({cancel_requested}). -spec activity_paused() -> boolean() | no_return(). activity_paused() -> call({activity_paused}). -spec elapsed_time() -> NativeTime :: non_neg_integer() | no_return(). elapsed_time() -> call({elapsed_time}). -spec elapsed_time(Unit :: erlang:time_unit()) -> non_neg_integer() | no_return(). elapsed_time(Unit) -> erlang:convert_time_unit(elapsed_time(), native, Unit). -spec remaining_time() -> NativeTime :: non_neg_integer() | infinity | no_return(). remaining_time() -> call({remaining_time}). -spec remaining_time(Unit :: erlang:time_unit()) -> non_neg_integer() | infinity | no_return(). remaining_time(Unit) -> case remaining_time() of infinity -> infinity; T -> erlang:convert_time_unit(T, native, Unit) end. -spec last_heartbeat() -> LastHeartbeat :: heartbeat() | no_return(). last_heartbeat() -> call({last_heartbeat}). -spec get_data() -> Data :: data() | no_return(). get_data() -> call({get_data}). -spec set_data(TaskData :: term()) -> ok. set_data(TaskData) -> cast({set_data, TaskData}).