Priority keys

Author: jeffschoner

examples/bin/worker

@@ -54,6 +54,7 @@ worker.register_workflow(MetadataWorkflow)
 worker.register_workflow(ParentCloseWorkflow)
 worker.register_workflow(ParentIdReuseWorkflow)
 worker.register_workflow(ParentWorkflow)
+worker.register_workflow(PriorityWorkflow)
 worker.register_workflow(ProcessFileWorkflow)
 worker.register_workflow(QueryWorkflow)
 worker.register_workflow(QuickTimeoutWorkflow)

examples/spec/integration/priority_workflow_spec.rb

@@ -0,0 +1,21 @@
+require 'workflows/priority_workflow'
+
+describe PriorityWorkflow do
+  subject { described_class }
+
+  it 'executes with priority' do
+    workflow_id = SecureRandom.uuid
+    run_id = Temporal.start_workflow(
+      PriorityWorkflow,
+      true,
+      options: { workflow_id: workflow_id, priority_key: 4 },
+    )
+    Temporal.await_workflow_result(
+      PriorityWorkflow,
+      workflow_id: workflow_id,
+      run_id: run_id,
+    )
+    execution_info = Temporal.fetch_workflow_execution_info('ruby-samples', workflow_id, run_id)
+    expect(execution_info.priority_key).to eq(4)
+  end
+end

examples/workflows/priority_workflow.rb

@@ -0,0 +1,7 @@
+require 'activities/hello_world_activity'
+
+class PriorityWorkflow < Temporal::Workflow
+  def execute(important = false)
+    return HelloWorldActivity.execute!('somebody', options: { priority_key: important ? 1 : 5 })
+  end
+end

lib/temporal/client.rb

@@ -42,6 +42,7 @@ def initialize(config)
     # @option options [Hash] :headers
     # @option options [Hash] :search_attributes
     # @option options [Integer] :start_delay determines the amount of seconds to wait before initiating a Workflow
+    # @option options [Integer] :priority_key
     #
     # @return [String] workflow's run ID
     def start_workflow(workflow, *input, options: {}, **args)
@@ -68,7 +69,8 @@ def start_workflow(workflow, *input, options: {}, **args)
           headers: config.header_propagator_chain.inject(execution_options.headers),
           memo: execution_options.memo,
           search_attributes: Workflow::Context::Helpers.process_search_attributes(execution_options.search_attributes),
-          start_delay: execution_options.start_delay
+          start_delay: execution_options.start_delay,
+          priority_key: options[:priority_key]
         )
       else
         raise ArgumentError, 'If signal_input is provided, you must also provide signal_name' if signal_name.nil?
@@ -88,7 +90,8 @@ def start_workflow(workflow, *input, options: {}, **args)
           search_attributes: Workflow::Context::Helpers.process_search_attributes(execution_options.search_attributes),
           signal_name: signal_name,
           signal_input: signal_input,
-          start_delay: execution_options.start_delay
+          start_delay: execution_options.start_delay,
+          priority_key: options[:priority_key]
         )
       end
 
@@ -112,6 +115,7 @@ def start_workflow(workflow, *input, options: {}, **args)
     # @option options [Hash] :timeouts check Temporal::Configuration::DEFAULT_TIMEOUTS
     # @option options [Hash] :headers
     # @option options [Hash] :search_attributes
+    # @option options [Integer] :priority_key
     #
     # @return [String] workflow's run ID
     def schedule_workflow(workflow, cron_schedule, *input, options: {}, **args)
@@ -137,6 +141,7 @@ def schedule_workflow(workflow, cron_schedule, *input, options: {}, **args)
         cron_schedule: cron_schedule,
         memo: execution_options.memo,
         search_attributes: Workflow::Context::Helpers.process_search_attributes(execution_options.search_attributes),
+        priority_key: options[:priority_key]
       )
 
       response.run_id

lib/temporal/connection/grpc.rb

@@ -121,7 +121,8 @@ def start_workflow_execution(
         cron_schedule: nil,
         memo: nil,
         search_attributes: nil,
-        start_delay: nil
+        start_delay: nil,
+        priority_key: nil
       )
         request = Temporalio::Api::WorkflowService::V1::StartWorkflowExecutionRequest.new(
           identity: identity,
@@ -149,6 +150,9 @@ def start_workflow_execution(
           ),
           search_attributes: Temporalio::Api::Common::V1::SearchAttributes.new(
             indexed_fields: converter.to_payload_map_without_codec(search_attributes || {})
+          ),
+          priority: priority_key.nil? ? nil : Temporalio::Api::Common::V1::Priority.new(
+            priority_key: priority_key
           )
         )
 
@@ -382,7 +386,8 @@ def signal_with_start_workflow_execution(
         cron_schedule: nil,
         memo: nil,
         search_attributes: nil,
-        start_delay: nil
+        start_delay: nil,
+        priority_key: nil
       )
         proto_header_fields = if headers.nil?
                                 converter.to_payload_map({})
@@ -422,6 +427,9 @@ def signal_with_start_workflow_execution(
           ),
           search_attributes: Temporalio::Api::Common::V1::SearchAttributes.new(
             indexed_fields: converter.to_payload_map_without_codec(search_attributes || {})
+          ),
+          priority: priority_key.nil? ? nil : Temporalio::Api::Common::V1::Priority.new(
+            priority_key: priority_key
           )
         )
 

lib/temporal/connection/serializer/schedule_activity.rb

@@ -19,7 +19,10 @@ def to_proto
                 start_to_close_timeout: object.timeouts[:start_to_close],
                 heartbeat_timeout: object.timeouts[:heartbeat],
                 retry_policy: Temporal::Connection::Serializer::RetryPolicy.new(object.retry_policy, converter).to_proto,
-                header: serialize_headers(object.headers)
+                header: serialize_headers(object.headers),
+                priority: object.priority_key.nil? ? nil : Temporalio::Api::Common::V1::Priority.new(
+                  priority_key: object.priority_key
+                )
               )
           )
         end

lib/temporal/connection/serializer/start_child_workflow.rb

@@ -32,6 +32,9 @@ def to_proto
                 memo: serialize_memo(object.memo),
                 workflow_id_reuse_policy: Temporal::Connection::Serializer::WorkflowIdReusePolicy.new(object.workflow_id_reuse_policy, converter).to_proto,
                 search_attributes: serialize_search_attributes(object.search_attributes),
+                priority: object.priority_key.nil? ? nil : Temporalio::Api::Common::V1::Priority.new(
+                  priority_key: object.priority_key
+                )
               )
           )
         end

lib/temporal/workflow/command.rb

@@ -2,8 +2,8 @@ module Temporal
   class Workflow
     module Command
       # TODO: Move these classes into their own directories under workflow/command/*
-      ScheduleActivity = Struct.new(:activity_type, :activity_id, :input, :task_queue, :retry_policy, :timeouts, :headers, keyword_init: true)
-      StartChildWorkflow = Struct.new(:workflow_type, :workflow_id, :input, :namespace, :task_queue, :retry_policy, :parent_close_policy, :timeouts, :headers, :cron_schedule, :memo, :workflow_id_reuse_policy, :search_attributes, keyword_init: true)
+      ScheduleActivity = Struct.new(:activity_type, :activity_id, :input, :task_queue, :retry_policy, :timeouts, :headers, :priority_key, keyword_init: true)
+      StartChildWorkflow = Struct.new(:workflow_type, :workflow_id, :input, :namespace, :task_queue, :retry_policy, :parent_close_policy, :timeouts, :headers, :cron_schedule, :memo, :workflow_id_reuse_policy, :search_attributes, :priority_key, keyword_init: true)
       ContinueAsNew = Struct.new(:workflow_type, :task_queue, :input, :timeouts, :retry_policy, :headers, :memo, :search_attributes, keyword_init: true)
       RequestActivityCancellation = Struct.new(:activity_id, keyword_init: true)
       RecordMarker = Struct.new(:name, :details, keyword_init: true)

lib/temporal/workflow/context.rb

@@ -88,7 +88,8 @@ def execute_activity(activity_class, *input, **args)
           task_queue: execution_options.task_queue,
           retry_policy: execution_options.retry_policy,
           timeouts: execution_options.timeouts,
-          headers: config.header_propagator_chain.inject(execution_options.headers)
+          headers: config.header_propagator_chain.inject(execution_options.headers),
+          priority_key: options[:priority_key]
         )
 
         target, cancelation_id = schedule_command(command)
@@ -150,6 +151,7 @@ def execute_workflow(workflow_class, *input, **args)
           memo: execution_options.memo,
           workflow_id_reuse_policy: workflow_id_reuse_policy,
           search_attributes: Helpers.process_search_attributes(execution_options.search_attributes),
+          priority_key: options[:priority_key]
         )
 
         target, cancelation_id = schedule_command(command)

lib/temporal/workflow/execution_info.rb

@@ -3,7 +3,7 @@
 module Temporal
   class Workflow
     class ExecutionInfo < Struct.new(:workflow, :workflow_id, :run_id, :start_time, :close_time, :status,
-                                     :history_length, :memo, :search_attributes, keyword_init: true)
+                                     :history_length, :memo, :search_attributes, :priority_key, keyword_init: true)
       STATUSES = [
         Temporal::Workflow::Status::RUNNING,
         Temporal::Workflow::Status::COMPLETED,
@@ -25,7 +25,8 @@ def self.generate_from(response, converter)
           status: Temporal::Workflow::Status::API_STATUS_MAP.fetch(response.status),
           history_length: response.history_length,
           memo: converter.from_payload_map(response.memo.fields),
-          search_attributes: search_attributes
+          search_attributes: search_attributes,
+          priority_key: response.priority&.priority_key
         ).freeze
       end