diff --git a/README.md b/README.md index 34a0c07..f41a8dd 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ Prerequisites: * [saga](saga) - Using undo/compensation using a very simplistic Saga pattern. * [sorbet_generic](sorbet_generic) - Proof of concept of how to do _advanced_ Sorbet typing with the SDK. * [standalone_activity](standalone_activity) - Execute, start, list, and count Standalone Activities -- Activities run directly from a Client without a Workflow. +* [timer](timer) - Use a timer to implement a monthly subscription with cancellation handling. * [updatable_timer](updatable_timer) - Demonstrates a blocking sleep that can be updated. * [worker_specific_task_queues](worker_specific_task_queues) - Use a unique Task Queue for each Worker to run a sequence of Activities on the same Worker. * [worker_versioning](worker_versioning) - Use the Worker Versioning feature to more easily version your workflows & other code. diff --git a/test/timer/subscription_workflow_test.rb b/test/timer/subscription_workflow_test.rb new file mode 100644 index 0000000..8e4166b --- /dev/null +++ b/test/timer/subscription_workflow_test.rb @@ -0,0 +1,40 @@ +# frozen_string_literal: true + +require 'test' +require 'timer/subscription_workflow' +require 'securerandom' +require 'temporalio/testing' +require 'temporalio/worker' + +module Timer + class SubscriptionWorkflowTest < Test + def test_workflow_charges_and_cancels + Temporalio::Testing::WorkflowEnvironment.start_time_skipping do |env| + worker = Temporalio::Worker.new( + client: env.client, + task_queue: "tq-#{SecureRandom.uuid}", + activities: [MyActivities::Charge], + workflows: [SubscriptionWorkflow] + ) + worker.run do + handle = env.client.start_workflow( + SubscriptionWorkflow, + 'test-user', + id: "wf-#{SecureRandom.uuid}", + task_queue: worker.task_queue + ) + + # Wait a bit for the workflow to start and the timer to be set + env.sleep(31 * 24 * 60 * 60) # 31 days — past the first charge + + # Cancel the workflow + handle.cancel + + # Workflow should complete as cancelled + err = assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result } + assert_kind_of Temporalio::Error::CanceledError, err.cause + end + end + end + end +end diff --git a/timer/README.md b/timer/README.md new file mode 100644 index 0000000..5ed0a31 --- /dev/null +++ b/timer/README.md @@ -0,0 +1,20 @@ +# Timer + +Use a timer (`Temporalio::Workflow.sleep`) to implement a monthly subscription. Also, handle workflow cancellation. + +To run, first see [README.md](../README.md) for prerequisites. Then, in another terminal, start the Ruby worker +from this directory: + + bundle exec ruby worker.rb + +Then in another terminal, start the workflow from this directory: + + bundle exec ruby starter.rb + +The worker terminal will show logs from running the workflow. The workflow will sleep for 30 days then charge the +user, repeating until cancelled. To cancel the workflow, use the Temporal CLI: + + temporal workflow cancel --workflow-id timer-sample-workflow-id + +There is also a [test](../test/timer/subscription_workflow_test.rb) that demonstrates time-skipping to test the +timer behavior without waiting. diff --git a/timer/my_activities.rb b/timer/my_activities.rb new file mode 100644 index 0000000..956da6b --- /dev/null +++ b/timer/my_activities.rb @@ -0,0 +1,13 @@ +# frozen_string_literal: true + +require 'temporalio/activity' + +module Timer + module MyActivities + class Charge < Temporalio::Activity::Definition + def execute(user_id) + "charge successful for #{user_id}" + end + end + end +end diff --git a/timer/starter.rb b/timer/starter.rb new file mode 100644 index 0000000..3a3cdfd --- /dev/null +++ b/timer/starter.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +require 'temporalio/client' +require 'temporalio/env_config' +require_relative 'subscription_workflow' + +# Load config and apply defaults +args, kwargs = Temporalio::EnvConfig::ClientConfig.load_client_connect_options +args[0] ||= 'localhost:7233' # Default address +args[1] ||= 'default' # Default namespace + +# Create a client +client = Temporalio::Client.connect(*args, **kwargs) + +# Run workflow +puts 'Executing workflow' +client.start_workflow( + Timer::SubscriptionWorkflow, + 'user-id-123', + id: 'timer-sample-workflow-id', + task_queue: 'timer-sample' +) +puts 'Workflow started (cancel it from the UI or CLI to see cancellation handling)' diff --git a/timer/subscription_workflow.rb b/timer/subscription_workflow.rb new file mode 100644 index 0000000..fc4cce0 --- /dev/null +++ b/timer/subscription_workflow.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +require 'temporalio/workflow' +require_relative 'my_activities' + +module Timer + class SubscriptionWorkflow < Temporalio::Workflow::Definition + def execute(user_id) + loop do + Temporalio::Workflow.sleep(30 * 24 * 60 * 60) # 30 days + + result = Temporalio::Workflow.execute_activity( + MyActivities::Charge, + user_id, + start_to_close_timeout: 5 * 60 + ) + Temporalio::Workflow.logger.info("Activity result: #{result}") + end + rescue Temporalio::Error::CanceledError + Temporalio::Workflow.logger.info('Workflow cancelled, cleaning up...') + raise + end + end +end diff --git a/timer/worker.rb b/timer/worker.rb new file mode 100644 index 0000000..26ef3e3 --- /dev/null +++ b/timer/worker.rb @@ -0,0 +1,27 @@ +# frozen_string_literal: true + +require_relative 'subscription_workflow' +require 'logger' +require 'temporalio/client' +require 'temporalio/env_config' +require 'temporalio/worker' + +# Load config and apply defaults +args, kwargs = Temporalio::EnvConfig::ClientConfig.load_client_connect_options +args[0] ||= 'localhost:7233' # Default address +args[1] ||= 'default' # Default namespace + +# Create a Temporal client +client = Temporalio::Client.connect(*args, **kwargs, logger: Logger.new($stdout, level: Logger::INFO)) + +# Create worker with the activity and workflow +worker = Temporalio::Worker.new( + client:, + task_queue: 'timer-sample', + activities: [Timer::MyActivities::Charge], + workflows: [Timer::SubscriptionWorkflow] +) + +# Run the worker until SIGINT +puts 'Starting worker (ctrl+c to exit)' +worker.run(shutdown_signals: ['SIGINT'])