Class: FlowChat::Processor

Inherits:
Object
  • Object
show all
Includes:
Instrumentation
Defined in:
lib/flow_chat/processor.rb

Constant Summary

Constants included from Instrumentation

Instrumentation::DELIVERED_MESSAGE_ID_KEY, Instrumentation::DELIVERY_DURATION_KEY

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Instrumentation

#elapsed_ms_since, #inbound_message?, #instrument, instrument, #platform_message_id_from, report_api_error, #report_delivery_failure, #report_delivery_success, #report_to_app, #report_to_subscribers

Constructor Details

#initialize(controller, enable_simulator: nil) {|_self| ... } ⇒ Processor

Returns a new instance of Processor.

Yields:

  • (_self)

Yield Parameters:



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/flow_chat/processor.rb', line 9

def initialize(controller, enable_simulator: nil)
  FlowChat.logger.debug { "Processor: Initializing processor for controller #{controller.class.name}" }

  @context = FlowChat::Context.new
  @context["controller"] = controller
  @context["enable_simulator"] = enable_simulator.nil? ? (defined?(Rails) && Rails.env.local?) : enable_simulator
  @custom_middleware_builder = ::Middleware::Builder.new(name: "processor.custom_middleware_builder")
  @session_options = FlowChat::Config.session
  @async_job_class = nil
  @async_job_params = {}

  FlowChat.logger.debug { "Processor: Simulator mode #{@context["enable_simulator"] ? "enabled" : "disabled"}" }

  yield self if block_given?

  FlowChat.logger.debug { "Processor: Initialized #{self.class.name} successfully" }
end

Instance Attribute Details

#async_job_class ⇒ Object (readonly)

Returns the value of attribute async_job_class.



7
8
9
 
# File 'lib/flow_chat/processor.rb', line 7

def async_job_class
  @async_job_class
end

#async_job_params ⇒ Object (readonly)

Returns the value of attribute async_job_params.



7
8
9
 
# File 'lib/flow_chat/processor.rb', line 7

def async_job_params
  @async_job_params
end

#context ⇒ Object (readonly)

Returns the value of attribute context.



7
8
9
 
# File 'lib/flow_chat/processor.rb', line 7

def context
  @context
end

#custom_middleware_builder ⇒ Object (readonly)

Returns the value of attribute custom_middleware_builder.



7
8
9
 
# File 'lib/flow_chat/processor.rb', line 7

def custom_middleware_builder
  @custom_middleware_builder
end

Instance Method Details

#async_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


110
111
112
 
# File 'lib/flow_chat/processor.rb', line 110

def async_enabled?
  !@async_job_class.nil?
end

#run(flow_class, action, **options) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/flow_chat/processor.rb', line 114

def run(flow_class, action, **options)
  # Instrument flow execution (this will log via LogSubscriber)
  instrument(Events::FLOW_EXECUTION_START, {
    flow_name: flow_class.name.underscore,
    action: action.to_s,
    processor_type: self.class.name
  })

  @context["processor"] = self
  @context["flow.name"] = flow_class.name.underscore
  @context["flow.class"] = flow_class
  @context["flow.action"] = action
  @context["flow.options"] = options

  FlowChat.logger.debug { "Processor: Context prepared for flow #{flow_class.name}" }

  stack = create_middleware_stack
  yield stack if block_given?

  FlowChat.logger.debug { "Processor: Executing middleware stack for #{flow_class.name}##{action}" }

  # Instrument flow execution with timing (this will log completion via LogSubscriber)
  instrument(Events::FLOW_EXECUTION_END, {
    flow_name: flow_class.name.underscore,
    action: action.to_s
  }) do
    stack.call(@context)
  end
rescue => error
  FlowChat.logger.error { "Processor: Flow execution failed - #{flow_class.name}##{action}, Error: #{error.class.name}: #{error.message}" }
  FlowChat.logger.debug { "Processor: Stack trace: #{error.backtrace.join("\n")}" }

  # Instrument flow execution error (this will log error via LogSubscriber)
  instrument(Events::FLOW_EXECUTION_ERROR, {
    flow_name: flow_class.name.underscore,
    action: action.to_s,
    error_class: error.class.name,
    error_message: error.message,
    backtrace: error.backtrace&.first(10)
  })

  raise
end

#use_async(job_class = nil, **job_params) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/flow_chat/processor.rb', line 92

def use_async(job_class = nil, **job_params)
  # If no job class provided, use GenericAsyncJob with factory param
  if job_class.nil?
    unless job_params.key?(:factory)
      raise ArgumentError, "When use_async is called without a job class, factory: parameter is required"
    end

    FlowChat.logger.debug { "Processor: Configuring async processing with GenericAsyncJob for factory '#{job_params[:factory]}'" }
    @async_job_class = FlowChat::GenericAsyncJob
  else
    FlowChat.logger.debug { "Processor: Configuring async processing with job class #{job_class.name} and params: #{job_params.inspect}" }
    @async_job_class = job_class
  end
  @async_job_params = job_params

  self
end

#use_cross_platform_sessions ⇒ Object 



71
72
73
74
75
76
 
# File 'lib/flow_chat/processor.rb', line 71

def use_cross_platform_sessions
  FlowChat.logger.debug { "Processor: Enabling cross-platform sessions via session configuration" }
  use_session_config(
    boundaries: [:flow]
  )
end

#use_durable_sessions(cross_gateway: false) ⇒ Object 



85
86
87
88
89
90
 
# File 'lib/flow_chat/processor.rb', line 85

def use_durable_sessions(cross_gateway: false)
  FlowChat.logger.debug { "Processor: Enabling durable sessions via session configuration" }
  use_session_config(
    identifier: :user_id
  )
end

#use_gateway(gateway_class, *args) ⇒ Object 



27
28
29
30
31
32
 
# File 'lib/flow_chat/processor.rb', line 27

def use_gateway(gateway_class, *args)
  FlowChat.logger.debug { "Processor: Configuring gateway #{gateway_class.name} with args: #{args.inspect}" }
  @gateway_class = gateway_class
  @gateway_args = args
  self
end

#use_middleware(middleware) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/flow_chat/processor.rb', line 59

def use_middleware(middleware)
  if block_given?
    yield custom_middleware_builder
    return self
  end

  raise "Middleware must be a class" unless middleware.is_a?(Class)
  FlowChat.logger.debug { "Processor: Adding custom middleware: #{middleware.name}" }
  custom_middleware_builder.use middleware
  self
end

#use_session_config(boundaries: nil, hash_identifiers: nil, identifier: nil, &block) ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/flow_chat/processor.rb', line 41

def use_session_config(boundaries: nil, hash_identifiers: nil, identifier: nil, &block)
  if block_given?
    FlowChat.logger.debug { "Processor: Configuring session config with custom proc" }
    @session_options = @session_options.dup
    @session_options.session_id_proc = block
  else
    FlowChat.logger.debug { "Processor: Configuring session config: boundaries=#{boundaries.inspect}, hash_identifiers=#{hash_identifiers}, identifier=#{identifier}" }

    # Update the session options directly
    @session_options = @session_options.dup
    @session_options.boundaries = Array(boundaries) unless boundaries.nil?
    @session_options.hash_identifiers = hash_identifiers unless hash_identifiers.nil?
    @session_options.identifier = identifier unless identifier.nil?
  end

  self
end

#use_session_store(session_store) ⇒ Object 



34
35
36
37
38
39
 
# File 'lib/flow_chat/processor.rb', line 34

def use_session_store(session_store)
  raise "Session store must be a class" unless session_store.is_a?(Class)
  FlowChat.logger.debug { "Processor: Configuring session store #{session_store.name}" }
  @context["session.store"] = session_store
  self
end

#use_url_isolation ⇒ Object 



78
79
80
81
82
83
 
# File 'lib/flow_chat/processor.rb', line 78

def use_url_isolation
  FlowChat.logger.debug { "Processor: Enabling URL-based session isolation" }
  current_boundaries = @session_options.boundaries.dup
  current_boundaries << :url unless current_boundaries.include?(:url)
  use_session_config(boundaries: current_boundaries)
end