Merge pull request coinbase#106 from stripe-private-oss-forks/jeffschoner/downstream

Downstream latest temporal-ruby changes

Author: jeffschoner-stripe

examples/activities/terminate_workflow_activity.rb

@@ -0,0 +1,5 @@
+class TerminateWorkflowActivity < Temporal::Activity
+  def execute(namespace, workflow_id, run_id)
+    Temporal.terminate_workflow(workflow_id, namespace: namespace, run_id: run_id)
+  end
+end

examples/bin/worker

@@ -25,6 +25,8 @@ worker.register_workflow(BranchingWorkflow)
 worker.register_workflow(CallFailingActivityWorkflow)
 worker.register_workflow(CancellingTimerWorkflow)
 worker.register_workflow(CheckWorkflow)
+worker.register_workflow(ChildWorkflowTimeoutWorkflow)
+worker.register_workflow(ChildWorkflowTerminatedWorkflow)
 worker.register_workflow(FailingActivitiesWorkflow)
 worker.register_workflow(FailingWorkflow)
 worker.register_workflow(HelloWorldWorkflow)
@@ -42,6 +44,7 @@ worker.register_workflow(QuickTimeoutWorkflow)
 worker.register_workflow(RandomlyFailingWorkflow)
 worker.register_workflow(ReleaseWorkflow)
 worker.register_workflow(ResultWorkflow)
+worker.register_workflow(ScheduleChildWorkflow)
 worker.register_workflow(SendSignalToExternalWorkflow)
 worker.register_workflow(SerialHelloWorldWorkflow)
 worker.register_workflow(SideEffectWorkflow)
@@ -66,6 +69,7 @@ worker.register_activity(LongRunningActivity)
 worker.register_activity(ProcessFileActivity)
 worker.register_activity(RandomlyFailingActivity)
 worker.register_activity(RandomNumberActivity)
+worker.register_activity(TerminateWorkflowActivity)
 worker.register_activity(SleepActivity)
 worker.register_activity(UploadFileActivity)
 worker.register_activity(Trip::BookFlightActivity)

examples/spec/integration/child_workflow_terminated_workflow_spec.rb

@@ -0,0 +1,22 @@
+require 'workflows/child_workflow_terminated_workflow.rb'
+
+describe ChildWorkflowTerminatedWorkflow do
+  subject { described_class }
+
+  it 'successfully can catch if a child workflow times out' do
+    workflow_id = SecureRandom.uuid
+
+    Temporal.start_workflow(
+      subject,
+      options: { workflow_id: workflow_id }
+    )
+
+    result = Temporal.await_workflow_result(
+      subject,
+      workflow_id: workflow_id
+    )
+
+    expect(result[:child_workflow_terminated]).to eq(true)
+    expect(result[:error]).to be_a(Temporal::ChildWorkflowTerminatedError)
+  end
+end

examples/spec/integration/child_workflow_timeout_workflow_spec.rb

@@ -0,0 +1,22 @@
+require 'workflows/child_workflow_timeout_workflow.rb'
+
+describe ChildWorkflowTimeoutWorkflow do
+  subject { described_class }
+
+  it 'successfully can catch if a child workflow times out' do
+    workflow_id = SecureRandom.uuid
+
+    Temporal.start_workflow(
+      subject,
+      options: { workflow_id: workflow_id }
+    )
+
+    result = Temporal.await_workflow_result(
+      subject,
+      workflow_id: workflow_id
+    )
+    puts result
+    expect(result[:child_workflow_failed]).to eq(true)
+    expect(result[:error]).to be_a(Temporal::ChildWorkflowTimeoutError)
+  end
+end

examples/spec/integration/schedule_child_workflow_spec.rb

@@ -0,0 +1,37 @@
+require 'workflows/schedule_child_workflow'
+require 'workflows/hello_world_workflow'
+
+describe ScheduleChildWorkflow, :integration do
+  let(:cron_schedule) { "*/6 * * * *" }
+
+  it 'schedules a child workflow with a given cron schedule' do
+    child_workflow_id = 'schedule_child_test_wf-' + SecureRandom.uuid
+    workflow_id, run_id = run_workflow(
+      described_class,
+      child_workflow_id,
+      cron_schedule,
+      options: {
+        timeouts: { execution: 10 }
+      }
+    )
+
+    wait_for_workflow_completion(workflow_id, run_id)
+    parent_history = fetch_history(workflow_id, run_id)
+
+    child_workflow_event = parent_history.history.events.detect do |event|
+      event.event_type == :EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED
+    end
+    expect(
+      child_workflow_event.start_child_workflow_execution_initiated_event_attributes.cron_schedule
+    ).to eq(cron_schedule)
+
+    # Expecting the child workflow to terminate as a result of the parent close policy
+    expect do
+      Temporal.await_workflow_result(
+        HelloWorldWorkflow,
+        workflow_id: child_workflow_id
+      )
+    end.to raise_error(Temporal::WorkflowTerminated)
+
+  end
+end

examples/workflows/child_workflow_terminated_workflow.rb

@@ -0,0 +1,21 @@
+require 'workflows/simple_timer_workflow'
+require 'activities/terminate_workflow_activity'
+
+class ChildWorkflowTerminatedWorkflow < Temporal::Workflow
+  def execute
+    # start a child workflow that executes for 60 seconds, then attempts to try and terminate that workflow
+    result = SimpleTimerWorkflow.execute(60)
+    child_workflow_execution = result.child_workflow_execution_future.get
+    TerminateWorkflowActivity.execute!(
+      'ruby-samples',
+      child_workflow_execution.workflow_id,
+      child_workflow_execution.run_id
+    )
+
+    # check that the result is now 'failed'
+    {
+      child_workflow_terminated: result.failed?, # terminated is represented as failed? with the Terminated Error
+      error: result.get
+    }
+  end
+end

examples/workflows/child_workflow_timeout_workflow.rb

@@ -0,0 +1,15 @@
+require 'workflows/quick_timeout_workflow'
+
+class ChildWorkflowTimeoutWorkflow < Temporal::Workflow
+  def execute
+    # workflow timesout before it can finish running, we should be able to detect that with .failed?
+    result = QuickTimeoutWorkflow.execute
+
+    result.get # wait for the workflow to finish so we can detect if it failed or not
+
+    {
+      child_workflow_failed: result.failed?,
+      error: result.get
+    }
+  end
+end

examples/workflows/schedule_child_workflow.rb

@@ -0,0 +1,6 @@
+class ScheduleChildWorkflow < Temporal::Workflow
+  def execute(child_workflow_id, cron_schedule)
+    HelloWorldWorkflow.schedule(cron_schedule, options: { workflow_id: child_workflow_id })
+    workflow.sleep(1)
+  end
+end

lib/temporal/configuration.rb

@@ -3,7 +3,7 @@
 require 'temporal/connection/converter/payload/nil'
 require 'temporal/connection/converter/payload/bytes'
 require 'temporal/connection/converter/payload/json'
-require 'temporal/connection/converter/payload/json_protobuf'
+require 'temporal/connection/converter/payload/proto_json'
 require 'temporal/connection/converter/composite'
 
 module Temporal
@@ -12,8 +12,7 @@ class Configuration
     Execution = Struct.new(:namespace, :task_queue, :timeouts, :headers, :search_attributes, keyword_init: true)
 
     attr_reader :timeouts, :error_handlers
-    attr_writer :converter
-    attr_accessor :connection_type, :host, :port, :credentials, :identity, :logger, :metrics_adapter, :namespace, :task_queue, :headers, :max_page_size, :connection_options, :search_attributes
+    attr_accessor :connection_type, :converter, :host, :port, :credentials, :identity, :logger, :metrics_adapter, :namespace, :task_queue, :headers, :max_page_size, :connection_options, :search_attributes
 
     # See https://docs.temporal.io/blog/activity-timeouts/ for general docs.
     # We want an infinite execution timeout for cron schedules and other perpetual workflows.
@@ -40,8 +39,8 @@ class Configuration
       payload_converters: [
         Temporal::Connection::Converter::Payload::Nil.new,
         Temporal::Connection::Converter::Payload::Bytes.new,
-        Temporal::Connection::Converter::Payload::JSON.new,
-        Temporal::Connection::Converter::Payload::JSONProtobuf.new,
+        Temporal::Connection::Converter::Payload::ProtoJSON.new,
+        Temporal::Connection::Converter::Payload::JSON.new
       ]
     ).freeze
 
@@ -77,10 +76,6 @@ def timeouts=(new_timeouts)
       @timeouts = DEFAULT_TIMEOUTS.merge(new_timeouts)
     end
 
-    def converter
-      @converter
-    end
-
     def for_connection
       Connection.new(
         type: connection_type,
@@ -90,25 +85,27 @@ def for_connection
         identity: identity || default_identity,
         options: {
           max_page_size: max_page_size
-        }.merge(connection_options),
+        }.merge(connection_options)
       ).freeze
     end
 
-    def default_identity
-      hostname = `hostname`
-      pid = Process.pid
-
-      "#{pid}@#{hostname}"
-    end
-
     def default_execution_options
       Execution.new(
         namespace: namespace,
         task_queue: task_list,
         timeouts: timeouts,
         headers: headers,
-        search_attributes: search_attributes,
+        search_attributes: search_attributes
       ).freeze
     end
+
+    private
+
+    def default_identity
+      hostname = `hostname`
+      pid = Process.pid
+
+      "#{pid}@#{hostname}".freeze
+    end
   end
 end