cleanup Filter building.

Author: apotonick

lib/trailblazer/activity/variable_mapping/runtime/filter.rb

@@ -5,29 +5,25 @@ module Runtime
         class Filter < Struct.new(:read_name, :write_name, keyword_init: true) # We *could* allow more options here.
           # This Node represents one step in the input/output pipe,
           # one filter.
-          def self.build_node(args_for_provider:, id:, **options)
-            provider_with_step_interface = args_for_provider[0]
-            options_for_provider_node = args_for_provider[2] || {} # FIXME: change public API of build_node.
+          def self.build_node(id:, **options)
+            circuit = build_circuit(**options)
 
             options_for_filter = options.slice(*members) # extract :read_name, :write_name.
 
+            create_node_for(circuit, id: id, **options_for_filter)
+          end
+
+          def self.build_circuit(args_for_step_build:, **options)
+            provider_arg, provider_options = args_for_step_build
+
             provider_node = Activity::Step.build(
-              provider_with_step_interface,
-              **options_for_provider_node, # DISCUSS: what's this?
+              provider_arg,
+              **provider_options, # e.g. {exec_context: Filter.new}
               binary: false
             )
 
-            steps = [ # FIXME: better defaulting, please, not very obvious.
-              [:invoke_provider, node: provider_node],
-              [:add_value_to_aggregate, :add_value_to_aggregate, Circuit::Task::Adapter::LibInterface::InstanceMethod],
-            ]
-
-            pipe = build_circuit(steps, **options)
+            steps = rows_for_build(provider_node)
 
-            create_node_for(pipe, id: id, **options_for_filter)
-          end
-
-          def self.build_circuit(steps, **)
             Circuit::Builder.Circuit(*steps)
           end
 
@@ -38,17 +34,26 @@ def self.create_node_for(circuit, id:, **options_for_filter)
             return Circuit::Node::MergeToCircuitOptions[id, circuit, Circuit::Processor, exec_context: filter_exec_context]
           end
 
+          # FIXME: can we reuse nodes?
+          def self.rows_for_build(provider_node)
+            [
+              [:invoke_provider, node: provider_node],
+              [:add_value_to_aggregate, :add_value_to_aggregate, Circuit::Task::Adapter::LibInterface::InstanceMethod],
+            ]
+          end
+
           module Out
 
           end
 
           class Conditioned < Filter
-            def self.build_circuit(*)
-              provider_with_step_interface = :read_variable_from_application_ctx
-
-              provider_node = Activity::Step.build(provider_with_step_interface, binary: false)
+            def self.build_circuit(**options)
+              super(**options, args_for_step_build: [:read_variable_from_application_ctx, {}]) # the provider is {#read_variable_from_application_ctx}.
+            end
 
-              circuit_steps = [
+            # FIXME: we can reuse most nodes?
+            def self.rows_for_build(provider_node)
+              [
                 [:variable_present_in_application_ctx?, :variable_present_in_application_ctx?, Circuit::Task::Adapter::LibInterface::InstanceMethod,
                   connections: Circuit::Resolver::Conditional.new([Left], nil, :invoke_provider)], # Left means terminate.
                 [:invoke_provider, node: provider_node, # extract a value
@@ -59,15 +64,16 @@ def self.build_circuit(*)
                   connections: Circuit::Resolver::Fixed.new(nil) # terminus.
                 ]
               ]
-
-              Circuit::Builder.Circuit(*circuit_steps)
             end
+
+            # TODO: save memory by not creating identical circuits!
+            # CIRCUIT = build_node(id: :bla)
           end
 
           class Defaulted < Filter
-            def self.build_circuit(*, default_provider:, **options)
+            def self.build_node(default_provider:, id:, **options)
               # FIXME: playing with "inheritance" here
-              conditioned_circuit = Conditioned.build_circuit
+              conditioned_circuit = Conditioned.build_circuit(**options)
 
               default_provider_node = Activity::Step.build(default_provider, binary: false)
 
@@ -79,7 +85,12 @@ def self.build_circuit(*, default_provider:, **options)
                 resolver: Circuit::Resolver::Fixed.new(:wrap_value_with_hash),
               ]
 
-              Circuit::Adds.(conditioned_circuit, adds_instruction)
+              circuit = Circuit::Adds.(conditioned_circuit, adds_instruction)
+
+              # FIXME: redundant.
+              options_for_filter = options.slice(*members) # extract :read_name, :write_name.
+
+              create_node_for(circuit, id: id, **options_for_filter)
             end
           end
 

test/filter_test.rb

@@ -10,7 +10,7 @@ class FilterTest < Minitest::Spec
   it "read a variable from the {application_ctx}, like In() => {:slug => :my_slug}" do
     my_node = Filter.build_node(
       id: nil,
-      args_for_provider: [:read_variable_from_application_ctx],
+      args_for_step_build: [:read_variable_from_application_ctx, {}],
       read_name: :slug,
       write_name: :my_slug,
     )
@@ -37,7 +37,7 @@ class FilterTest < Minitest::Spec
 
     my_node = Filter.build_node(
       id: nil,
-      args_for_provider: [my_input_provider],
+      args_for_step_build: [my_input_provider, {}],
       write_name: :my_slug,
     )
 
@@ -58,7 +58,7 @@ class FilterTest < Minitest::Spec
     assert_equal flow_options, {}
   end
 
-  it "invoke an {:instance_method}, wrap the value, FIXME: we're using the {args_for_provider} option!!!!!!!!!" do
+  it "invoke an {:instance_method}, wrap the value" do
     my_exec_context = Class.new do
       def downcase_slug(ctx, slug:, **)
         slug.upcase
@@ -68,7 +68,7 @@ def downcase_slug(ctx, slug:, **)
 # FIXME: remove write_name where it's nil
     my_node = Filter.build_node(
       id: nil,
-      args_for_provider: [:downcase_slug, StepInterface::InstanceMethod, exec_context: my_exec_context],
+      args_for_step_build: [:downcase_slug, exec_context: my_exec_context],
       read_name: nil,
       write_name: :my_slug,
     )
@@ -95,7 +95,7 @@ def downcase_slug(ctx, slug:, **)
 
     my_node = Filter.build_node(
       id: nil,
-      args_for_provider: [my_input_provider, StepInterface],
+      args_for_step_build: [my_input_provider, {}],
     )
 
     lib_ctx, flow_options = assert_run my_node, seq: nil, node: true, target_ctx: original_target_ctx = {slug: "generator-1"}.freeze,
@@ -113,7 +113,7 @@ def downcase_slug(ctx, slug:, **)
 
       my_node = Filter.build_node(
         id: nil,
-        args_for_provider: [my_input_provider],
+        args_for_step_build: [my_input_provider, {}],
         write_name: :my_slug,
         # adds: [Filter::Build::WRAP_VALUE_WITH_HASH], # FIXME: this is for Filter level, then we also have step_block on the Step level.
       )
@@ -177,7 +177,7 @@ def downcase_slug(ctx, slug:, **)
     it "writes value to aggregate if it's present (Conditioned)" do
       my_node = Filter::Conditioned.build_node(
         id: nil,
-        args_for_provider: [nil], # FIXME: we don't need this here.
+        args_for_step_build: [nil], # FIXME: we don't need this here.
         write_name: :slug,
         read_name: :slug,
       )
@@ -195,7 +195,6 @@ def downcase_slug(ctx, slug:, **)
     it "works even if the incoming signal is something other than {nil}" do
       my_node = Filter::Conditioned.build_node(
         id: nil,
-        args_for_provider: [nil], # FIXME: we don't need this here.
         write_name: :slug,
         read_name: :slug,
       )
@@ -209,14 +208,41 @@ def downcase_slug(ctx, slug:, **)
       assert_equal flow_options, {}
     end
 
+    it "Conditioned always uses the same circuit" do
+      skip
+      require "benchmark/memory"
+
+      Benchmark.memory do |x|
+        x.report("rebuild") do
+          my_node = Filter::Conditioned.build_node(
+            id: nil,
+            write_name: :slug,
+            read_name: :slug,
+          )
+
+          my_node_2 = Filter::Conditioned.build_node(
+            id: nil,
+            write_name: :slug,
+            read_name: :slug,
+          )
+        end
+
+        x.compare!
+      end
+
+      raise
+    end
   end
 
   it "defaults value if absent, and reads value otherwise (Defaulted)" do
     my_provider_for_default = ->(ctx, params:, **) { params[:id] }
 
     my_node = Filter::Defaulted.build_node(
-      id: nil,default_provider: my_provider_for_default, read_name: :global_id, write_name: :my_global_id,
-      args_for_provider: [nil] # FIXME: remove!
+      id: nil,
+      default_provider: my_provider_for_default,
+      read_name: :global_id,
+      write_name: :my_global_id,
+      # args_for_step_build: [nil] # FIXME: remove!
     )