Class: Julewire::Core::Runtime

Inherits:
Object
  • Object
show all
Defined in:
lib/julewire/core/runtime.rb

Overview

rubocop:disable Metrics/ClassLength

Defined Under Namespace

Classes: CloseTransition, PipelineReplacement, ResetTransition

Constant Summary collapse

CONFIGURE_GUARD_KEY = 
:__julewire_core_configure_guard__
RUNTIME_COUNTER_KEYS = 
%i[
  close_attempts
  configure_attempts
  flush_attempts
  lifecycle_warnings
  post_close_emits_total
  reset_attempts
  runtime_callback_failures
  runtime_failures
].freeze

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Runtime

Returns a new instance of Runtime.



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/julewire/core/runtime.rb', line 30

def initialize
  @configure_mutex = Mutex.new
  @configure_generation = Concurrent::AtomicFixnum.new
  @post_close_emit_count = Concurrent::AtomicFixnum.new
  @runtime_health = build_runtime_health
  @integration_health = Diagnostics::IntegrationHealthStore.new
  @invalid_severity_reporter = Diagnostics::InvalidSeverityReporter.counter
  @state_ref = Concurrent::AtomicReference.new(RuntimeState.default)
  @execution_boundary = build_execution_boundary
end

Instance Method Details

#after_fork! ⇒ Object 



153
154
155
156
 
# File 'lib/julewire/core/runtime.rb', line 153

def after_fork!
  reject_runtime_call_during_configure!(:after_fork!)
  RuntimeRegistry.reset_after_fork(primary: self)
end

#attributes ⇒ Object 



45
 
# File 'lib/julewire/core/runtime.rb', line 45

def attributes = ContextStore.current.attributes_proxy

#carry ⇒ Object 



47
 
# File 'lib/julewire/core/runtime.rb', line 47

def carry = ContextStore.current.carry_proxy

#close(timeout: UNSET) ⇒ Object 



131
132
133
 
# File 'lib/julewire/core/runtime.rb', line 131

def close(timeout: UNSET)
  close_state_resources(close_state(timeout))
end

#config ⇒ Object 



41
 
# File 'lib/julewire/core/runtime.rb', line 41

def config = runtime_state.configuration

#configure ⇒ Object

Raises:

  • (ArgumentError) 


109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/julewire/core/runtime.rb', line 109

def configure(&)
  raise ArgumentError, "Julewire.configure requires a block" unless block_given?

  increment_runtime_count(:configure_attempts)
  replacement = build_configured_pipeline(&)
  deadline = Scheduling::Deadline.for(replacement.close_timeout)
  if replacement.close_pipeline
    report_pipeline_close_result(
      replacement.old_pipeline,
      timeout: Scheduling::Deadline.remaining(deadline),
      on_failure: replacement.old_on_failure,
      operation: :configure,
      skip_resource_identities: replacement.retained_resources
    )
  end
  config
end

#context ⇒ Object 



49
 
# File 'lib/julewire/core/runtime.rb', line 49

def context = ContextStore.current.context_proxy

#current_execution ⇒ Object 



53
54
55
56
 
# File 'lib/julewire/core/runtime.rb', line 53

def current_execution
  scope = ContextStore.current.current_scope
  scope && Execution::View.new(scope)
end

#current_execution? ⇒ Boolean

Returns:

  • (Boolean) 


58
59
60
 
# File 'lib/julewire/core/runtime.rb', line 58

def current_execution?
  ContextStore.current.current_scope?
end

#emit(record = UNSET, **fields) ⇒ Object 



66
67
68
 
# File 'lib/julewire/core/runtime.rb', line 66

def emit(record = UNSET, **fields, &)
  emit_with_level_check(record, true, fields, &)
end

#emit_envelope(input:, context:, scope:, carry:, attributes:, neutral:, enforce_level: true, owned: false) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/julewire/core/runtime.rb', line 80

def emit_envelope(input:, context:, scope:, carry:, attributes:, neutral:, enforce_level: true, owned: false)
  reject_runtime_call_during_configure!(:emit_envelope)
  state = runtime_state
  return record_post_close_emit(state) if state.pipeline_closed

  begin
    record = envelope_draft(input, context, attributes, neutral, carry, scope, state, owned: owned).to_record
    state.pipeline.emit_record(record, enforce_level: enforce_level)
  rescue StandardError => e
    notify_failure(e, state, action: :emit_envelope)
    nil
  end
end

#emit_integration(record, enforce_level: true) ⇒ Object 



74
75
76
77
78
 
# File 'lib/julewire/core/runtime.rb', line 74

def emit_integration(record, enforce_level: true)
  with_emit_guard(:emit_integration) do |state|
    state.pipeline.emit_integration(record, enforce_level: enforce_level)
  end
end

#emit_summary_record(scope) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/julewire/core/runtime.rb', line 94

def emit_summary_record(scope)
  reject_runtime_call_during_configure!(:emit_summary_record)
  state = runtime_state
  return record_post_close_emit(state) if state.pipeline_closed

  begin
    input = scope.owned_summary_record_input
    Serialization::DeepFreeze.validate_symbol_keys(input)
    state.pipeline.emit_isolated_input(input)
  rescue StandardError => e
    notify_failure(e, state, action: :emit_summary_record)
    nil
  end
end

#emit_without_level(record = UNSET, **fields) ⇒ Object 



70
71
72
 
# File 'lib/julewire/core/runtime.rb', line 70

def emit_without_level(record = UNSET, **fields, &)
  emit_with_level_check(record, false, fields, &)
end

#flush(timeout: UNSET) ⇒ Object 



127
128
129
 
# File 'lib/julewire/core/runtime.rb', line 127

def flush(timeout: UNSET)
  call_validated_lifecycle(:flush, timeout)
end

#health ⇒ Object 



171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/julewire/core/runtime.rb', line 171

def health
  state = runtime_state
  pipeline_health = state.pipeline.health
  integrations = @integration_health.health
  process_integrations = Diagnostics::ProcessIntegrationHealth.health
  {
    closed: state.pipeline_closed,
    counts: runtime_counts_snapshot,
    generation: state.pipeline_generation,
    integrations: integrations,
    last_callback_failure: @runtime_health.last_callback_failure,
    last_failure: @runtime_health.last_failure,
    pipeline: pipeline_health,
    process_integrations: process_integrations,
    status: runtime_status(state, pipeline_health, integrations, process_integrations)
  }
end

#labels ⇒ Object 



43
 
# File 'lib/julewire/core/runtime.rb', line 43

def labels = config.labels

#record_integration_failure(integration, error, **metadata) ⇒ Object 



163
164
165
 
# File 'lib/julewire/core/runtime.rb', line 163

def record_integration_failure(integration, error, **metadata)
  @integration_health.record_failure(integration, error, **metadata)
end

#record_integration_success(integration) ⇒ Object 



167
168
169
 
# File 'lib/julewire/core/runtime.rb', line 167

def record_integration_success(integration)
  @integration_health.record_success(integration)
end

#reset! ⇒ Object 



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/julewire/core/runtime.rb', line 135

def reset!
  increment_runtime_count(:reset_attempts)
  reset_result = reject_runtime_call_during_configure!(:reset!) do
    @configure_mutex.synchronize { reset_under_lock }
  end
  deadline = Scheduling::Deadline.for(reset_result.close_timeout)
  return unless reset_result.close_pipeline

  report_pipeline_close_result(
    reset_result.old_pipeline,
    timeout: Scheduling::Deadline.remaining(deadline),
    on_failure: reset_result.old_on_failure,
    operation: :reset
  )
end

#reset_after_fork_runtime! ⇒ Object 



158
159
160
161
 
# File 'lib/julewire/core/runtime.rb', line 158

def reset_after_fork_runtime!
  reset_after_fork_state!
  runtime_state.pipeline.after_fork!
end

#reset_facade! ⇒ Object 



151
 
# File 'lib/julewire/core/runtime.rb', line 151

def reset_facade! = RuntimeRegistry.reset(primary: self)

#start_execution ⇒ Object 



64
 
# File 'lib/julewire/core/runtime.rb', line 64

def start_execution(...) = @execution_boundary.start_execution(...)

#summary ⇒ Object 



51
 
# File 'lib/julewire/core/runtime.rb', line 51

def summary = ContextStore.current.summary_proxy

#with_execution ⇒ Object 



62
 
# File 'lib/julewire/core/runtime.rb', line 62

def with_execution(...) = @execution_boundary.with_execution(...)