Synchronous-proxy example in Ruby (#138)

* initial commit; broken and a bit messy with logging

* save current code while I go work on adding SignalExternalWorkflow support

* initial work to support SignalExternalWorkflow

* define the serializer and hook it up

* stub in what I think is the correct work for each event type

* some fixes per antstorm advice

* initial attempt at integration test

* docs on testing and an improvement to existing test

* encode the signal payload using correct helper

* return a Future and fulfill it correctly upon completion

* get the \*event_id from the right field in the command structure

* modify test to verify the signal is only received once

* test for failure to deliver a signal to external workflow

* do not discard the failure command otherwise non-deterministic

* simplify test workflow by eliminating unnecessary timer

* oops, had double call to #schedule_command so signals were sent twice

* edit description of example

* split to separate files and improve test coverage

* change method signature for consistency and a few other cleanups

* oops, fix EventType name to match correct constant

* cleanup to work with new #signal_external_workflow call

* improve readme

* add custom errors and set to nonretryable for upstream error handling

* finish error handling logic

* improve how errors are propagated, detected, and logged

* improve the README to explain the pattern and give another example

* small spelling fix

* make code more idiomatic for Ruby

* make Ruby a bit more idiomatic

* switch to CAP_SNAKECASE for constants

* remove task_queue setting since default is set in configuration.rb

* remove unused argument

* remove antiquated error handling and reraise exceptions instead

* loop using retry until no exceptions are received

Author: chuckremes2

examples/synchronous-proxy/README.md

@@ -0,0 +1,77 @@
+# Purpose
+
+This pattern is used when a non-workflow process needs to advance a workflow state
+machine from its initial state to its terminal state. It does this by adding input
+data to the workflow (via Signals) and receiving new information back from the
+workflow (when a secondary proxy workflow exits and returns a value).
+
+The only way to add information to a workflow is via a Signal.
+
+There are two ways to
+get information out of a workflow. One, the workflow has a Query handler and can respond
+to queries. However, this is limited in that Queries may not modify the state of the
+workflow itself. Two, the workflow can exit and return a result to its caller. This
+second approach is leveraged by the pattern to get information back from the primary
+workflow. This information could be used to determine branching behavior for the
+non-workflow caller.
+
+The flow of calls is outlined in the diagram below.
+
+![Flow Diagram](flow.png)
+
+# Explanation
+
+The primary use-case for this pattern is for a non-workflow process to *send and receive* data
+to and from a workflow. Note that a Temporal client may send a signal to a workflow but the
+information transfer is one-way (i.e. fire and forget). There is no mechanism for a workflow
+to send a signal to a non-workflow. A keen observer would note that a Query can be used to
+ask for information, however a Query is supposed to be idempotent and *should not cause any
+state change* in the workflow itself. Also, Queries imply polling for a result which is slow
+and inefficient. Therefore, it is not a mechanism for sending new information
+into a workflow and receiving a response.
+
+So, the non-workflow process can communicate to a workflow by:
+
+a) Starting that workflow, and
+
+b) Communicating with the workflow by creating proxy workflows to signal the main workflow and
+then block for a response. When these proxy workflows exit, they can return the response to the
+caller.
+
+In the real world, this pattern could be utilized for managing an interaction via a series of
+web pages. Imagine that a user lands on a home page and clicks a link to apply for a library
+card. The link hits the web application's controller and can now start the
+`ApplyForLibraryCardWorkflow`. The workflow ID could be returned back in a response to the caller
+as a session value, for example.
+
+On the next page, the user can fill out the application for the library card by providing their
+name, address, and phone number. Upon submission of this form (via POST), the web application
+controller can 1) lookup the associated workflow from the session, and 2) create the
+`SubmitPersonalDetailsWorkflow` workflow and pass in the form data. This workflow packages up
+the data and signals it to the `ApplyForLibraryCardWorkflow` and waits for a response via another
+signal. The main workflow applies the appropriate business logic to the payload and advances its
+state. It then signals back to the proxy workflow the result of its work and then blocks to
+await new data.
+
+Depending on the response from the `ApplyForLibraryCardWorkflow`, the controller can render a page
+to continue the application or ask for the user to correct some bad input.
+
+Continue and repeat this action via the web application controller(s) as it moves the user
+through the entire library card application journey. By its nature, web applications are all stateless
+and asynchronous, so the state and behavior are encapsulated by the workflow and its associated
+activity outcomes. The only state outside of the workflow that the web application cares about is the
+session information so it can match the user back to the correct workflow.
+
+# Execution
+
+Open two shells / terminal windows. In one, execute:
+```shell
+ruby worker/worker.rb
+```
+In the second, execute:
+```shell
+ruby ui/main.rb
+```
+In the shell running `ui` it will ask a series of questions. Answer the questions and the
+program will send the appropriate signals around to complete the process. Upon completion it
+prints a success message.

examples/synchronous-proxy/activities.rb

@@ -0,0 +1,57 @@
+module SynchronousProxy
+  class RegisterEmailActivity < Temporal::Activity
+    def execute(email)
+      logger.info "activity: registered email #{email}"
+      nil
+    end
+  end
+
+  class ValidateSizeActivity < Temporal::Activity
+    InvalidSize = Class.new(StandardError)
+
+    retry_policy(
+      interval: 1,
+      backoff: 1,
+      max_attempts: 3,
+      non_retriable_errors: [InvalidSize])
+
+    def execute(size)
+      logger.info "activity: validate size #{size}"
+      return nil if TShirtSizes.include?(size)
+
+      raise InvalidSize.new("#{size} is not a valid size choice.")
+    end
+  end
+
+  class ValidateColorActivity < Temporal::Activity
+    InvalidColor = Class.new(StandardError)
+
+    retry_policy(
+      interval: 1,
+      backoff: 1,
+      max_attempts: 3,
+      non_retriable_errors: [InvalidColor])
+
+    def execute(color)
+      logger.info "activity: validate color #{color}"
+      return nil if TShirtColors.include?(color)
+
+      raise InvalidColor.new("#{color} is not a valid color choice.")
+    end
+  end
+
+  class ScheduleDeliveryActivity < Temporal::Activity
+    def execute(order)
+      delivery_date = Time.now + (2 * 60 * 60 * 24)
+      logger.info "activity: scheduled delivery for order #{order} at #{delivery_date}"
+      delivery_date
+    end
+  end
+
+  class SendDeliveryEmailActivity < Temporal::Activity
+    def execute(order, order_id, delivery_date)
+      logger.info "email to: #{order.email}, order: #{order}, scheduled delivery: #{delivery_date}"
+      nil
+    end
+  end
+end

examples/synchronous-proxy/configuration.rb

@@ -0,0 +1,17 @@
+require 'bundler'
+Bundler.require :default
+
+require 'temporal'
+
+Temporal.configure do |config|
+  config.host = 'localhost'
+  config.port = 7233
+  config.namespace = 'ruby-samples'
+  config.task_queue = 'ui-driven'
+end
+
+begin
+  Temporal.register_namespace('ruby-samples', 'A safe space for playing with Temporal Ruby')
+rescue Temporal::NamespaceAlreadyExistsFailure
+  nil # service was already registered
+end

examples/synchronous-proxy/flow.png

@@ -0,0 +1,112 @@
+module SynchronousProxy
+  module Proxy
+    # We support talking between two workflows using these helper methods. Each workflow
+    # that wants to communicate will include this module. Unlike the Go examples which use
+    # channels, the Ruby support is via #on_signal (to receive), Temporal.signal_workflow
+    # (to send), and #wait_for (to block and wait for incoming signal).
+    #
+    # The basic trick is that we register a single #on_signal signal handler per workflow
+    # via the #setup_signal_handler method. Each incoming signal is parsed to determine if
+    # it's a request or response and then the appropriate ivar is set. After the signal handler
+    # runs, this client executes the block attached to the #wait_for method to see if it
+    # returns true. If the block evaluates that it has received a value into the ivar, it
+    # returns true and unblocks.
+    #
+    module Communications
+      REQUEST_SIGNAL_NAME = "proxy-request-signal".freeze
+      RESPONSE_SIGNAL_NAME = "proxy-response-signal".freeze
+
+      SignalDetails = Struct.new(
+        :name, :key, :value, :error, :calling_workflow_id,
+        keyword_init: true
+      ) do
+        def error?
+          key == "error"
+        end
+
+        def to_input
+          [calling_workflow_id, name, key, value]
+        end
+
+        def self.from_input(input)
+          new(name: input[1], key: input[2], value: input[3], calling_workflow_id: input[0])
+        end
+      end
+
+      def setup_signal_handler
+        w_id = workflow.metadata.id
+        logger.info("#{self.class.name}#setup_signal_handler, Setup signal handler for workflow #{w_id}")
+
+        workflow.on_signal do |signal, input|
+          logger.info("#{self.class.name}#setup_signal_handler, Received signal for workflow #{w_id}, signal #{signal}, input #{input.inspect}")
+          details = SignalDetails.from_input(input)
+
+          case signal
+          when REQUEST_SIGNAL_NAME
+            @request_signal = details
+
+          when RESPONSE_SIGNAL_NAME
+            @response_signal = details
+
+          else
+            logger.warn "#{self.class.name}#setup_signal_handler, Unknown signal received"
+          end
+        end
+      end
+
+      def wait_for_response
+        # #workflow is defined as part of the Temporal::Workflow class and is therefore available to
+        # any methods inside the class plus methods that are included from a Module like this one
+        workflow.wait_for { !!@response_signal }
+      end
+
+      def wait_for_request
+        workflow.wait_for { !!@request_signal }
+      end
+
+      def send_error_response(target_workflow_id, err)
+        w_id = workflow.metadata.id
+
+        logger.info("#{self.class.name}#send_error_response, Sending error response from #{w_id} to #{target_workflow_id}")
+        logger.info("#{self.class.name}#send_error_response, err is #{err.inspect}")
+        details = SignalDetails.new(key: "error", value: err, calling_workflow_id: w_id)
+        workflow.signal_external_workflow(workflow, RESPONSE_SIGNAL_NAME, target_workflow_id, "", details.to_input)
+        nil
+      end
+
+      def send_response(target_workflow_id, key, value)
+        w_id = workflow.metadata.id
+
+        logger.info("#{self.class.name}#send_response, Sending response from #{w_id} to #{target_workflow_id}")
+        details = SignalDetails.new(key: key, value: value, calling_workflow_id: w_id)
+        workflow.signal_external_workflow(workflow, RESPONSE_SIGNAL_NAME, target_workflow_id, "", details.to_input)
+        nil
+      end
+
+      def send_request(target_workflow_id, key, value)
+        w_id = workflow.metadata.id
+
+        logger.info("#{self.class.name}#send_request, Sending request from #{w_id} to #{target_workflow_id}, key #{key}, value #{value}, calling workflow #{w_id}")
+        details = SignalDetails.new(key: key, value: value, calling_workflow_id: w_id)
+        workflow.signal_external_workflow(workflow, REQUEST_SIGNAL_NAME, target_workflow_id, "", details.to_input)
+        nil
+      end
+
+      def receive_response(description="unknown")
+        @response_signal = nil
+        w_id = workflow.metadata.id
+        Temporal.logger.info("#{self.class.name}#receive_response, Waiting for response on [#{description}] in workflow #{w_id}")
+        wait_for_response
+        @response_signal
+      end
+
+      def receive_request(description="unknown")
+        @request_signal = nil
+        w_id = workflow.metadata.id
+        Temporal.logger.info("#{self.class.name}#receive_request, Waiting for request on [#{description}] in workflow #{w_id}")
+        wait_for_request
+        @request_signal
+      end
+    end
+  end
+end

examples/synchronous-proxy/proxy/communications.rb

@@ -0,0 +1,68 @@
+require_relative "../configuration"
+require_relative "../workflows"
+
+module SynchronousProxy
+  module UI
+    class Main
+      def run
+        random_id = rand(999_999_999)
+        sequence_no = 0
+        status = create_order(random_id, sequence_no)
+
+        sequence_no += 1
+          email = prompt_and_read_input("Please enter you email address:")
+          status = update_order(random_id: random_id, sequence_no: sequence_no, order_id: status.order_id, stage: SynchronousProxy::RegisterStage, value: email)
+          puts "status #{status.inspect}"
+
+        sequence_no += 1
+        begin
+          size = prompt_and_read_input("Please enter your requested size:")
+          status = update_order(random_id: random_id, sequence_no: sequence_no, order_id: status.order_id, stage: SynchronousProxy::SizeStage, value: size)
+          puts "status #{status.inspect}"
+        rescue SynchronousProxy::ValidateSizeActivity::InvalidSize => e
+          STDERR.puts e.message
+          retry
+        end
+
+        sequence_no += 1
+        begin
+          color = prompt_and_read_input("Please enter your required tshirt color:")
+          status = update_order(random_id: random_id, sequence_no: sequence_no, order_id: status.order_id, stage: SynchronousProxy::ColorStage, value: color)
+          puts "status #{status.inspect}"
+        rescue SynchronousProxy::ValidateColorActivity::InvalidColor => e
+          STDERR.puts e.message
+          retry
+        end
+
+        puts "Thanks for your order!"
+        puts "You will receive an email with shipping details shortly"
+        puts "Exiting at #{Time.now}"
+      end
+
+      def create_order(random_id, sequence_no)
+        w_id = "new-tshirt-order-#{random_id}-#{sequence_no}"
+        workflow_options = {workflow_id: w_id}
+        Temporal.start_workflow(SynchronousProxy::OrderWorkflow, options: workflow_options)
+        status = SynchronousProxy::OrderStatus.new
+        status.order_id = w_id
+        status
+      end
+
+      def update_order(random_id:, sequence_no:, order_id:, stage:, value:)
+        w_id = "update_#{stage}_#{random_id}-#{sequence_no}"
+        workflow_options = {workflow_id: w_id}
+        run_id = Temporal.start_workflow(SynchronousProxy::UpdateOrderWorkflow, order_id, stage, value, options: workflow_options)
+        Temporal.await_workflow_result(SynchronousProxy::UpdateOrderWorkflow, workflow_id: w_id, run_id: run_id)
+      end
+
+      def prompt_and_read_input(prompt)
+        print(prompt + " ")
+        gets.chomp
+      end
+    end
+  end
+end
+
+if $0 == __FILE__
+  SynchronousProxy::UI::Main.new.run
+end

examples/synchronous-proxy/ui/main.rb

@@ -0,0 +1,15 @@
+require_relative "../configuration"
+require_relative "../workflows"
+require_relative "../activities"
+require 'temporal/worker'
+
+worker = Temporal::Worker.new
+worker.register_workflow(SynchronousProxy::OrderWorkflow)
+worker.register_workflow(SynchronousProxy::UpdateOrderWorkflow)
+worker.register_workflow(SynchronousProxy::ShippingWorkflow)
+worker.register_activity(SynchronousProxy::RegisterEmailActivity)
+worker.register_activity(SynchronousProxy::ValidateSizeActivity)
+worker.register_activity(SynchronousProxy::ValidateColorActivity)
+worker.register_activity(SynchronousProxy::ScheduleDeliveryActivity)
+worker.register_activity(SynchronousProxy::SendDeliveryEmailActivity)
+worker.start