Class: Julewire::Ractor::Destination

Inherits:
Object
  • Object
show all
Defined in:
lib/julewire/ractor/destination.rb

Overview

rubocop:disable Metrics/ClassLength -- Owns parent queue, worker lifecycle, and health.

Constant Summary collapse

DEFAULT_MAX_QUEUE = 
1024
DEFAULT_REQUEST_TIMEOUT = 
1

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(output:, name: :ractor, formatter: Julewire::RecordFormatter.new, encoder: Julewire::JsonEncoder.new, max_record_bytes: Core::DEFAULT_MAX_RECORD_BYTES, max_queue: DEFAULT_MAX_QUEUE, close_output: false, request_timeout: DEFAULT_REQUEST_TIMEOUT, on_drop: nil, on_failure: nil) ⇒ Destination

rubocop:disable Metrics/ParameterLists -- Destination setup mirrors core destination knobs.

Raises:

  • (ArgumentError) 


57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/julewire/ractor/destination.rb', line 57

def initialize( # rubocop:disable Metrics/ParameterLists -- Destination setup mirrors core destination knobs.
  output:,
  name: :ractor,
  formatter: Julewire::RecordFormatter.new,
  encoder: Julewire::JsonEncoder.new,
  max_record_bytes: Core::DEFAULT_MAX_RECORD_BYTES,
  max_queue: DEFAULT_MAX_QUEUE,
  close_output: false,
  request_timeout: DEFAULT_REQUEST_TIMEOUT,
  on_drop: nil,
  on_failure: nil
)
  @name = Core::Destinations.normalize_name(name)
  @formatter = validate_callable(formatter, name: :formatter)
  @encoder = validate_callable(encoder, name: :encoder)
  Core::Destinations::Sink.validate_writeable!(output)
  Core::Validation.validate_byte_limit!(max_record_bytes, name: :max_record_bytes)
  Core::Validation.validate_non_negative_integer!(max_queue, name: :max_queue)
  Core::Validation.validate_timeout!(request_timeout, name: :request_timeout)
  raise ArgumentError, "request_timeout must be a non-negative finite Numeric" if request_timeout.nil?

  Core::Validation.validate_callable!(on_drop, name: :on_drop, allow_nil: true)
  Core::Validation.validate_callable!(on_failure, name: :on_failure, allow_nil: true)
  @output = output
  @max_record_bytes = max_record_bytes
  @max_queue = max_queue
  @close_output = close_output
  @request_timeout = request_timeout
  @on_drop = on_drop
  @on_failure = on_failure
  @fork_lifecycle_mutex = Mutex.new
  initialize_tracking(closed: false)
  start_worker
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



55
56
57
 
# File 'lib/julewire/ractor/destination.rb', line 55

def name
  @name
end

Instance Method Details

#after_fork! ⇒ Object 



161
162
163
164
165
166
167
168
169
170
 
# File 'lib/julewire/ractor/destination.rb', line 161

def after_fork!
  @fork_lifecycle_mutex.synchronize { resume_after_fork! }
  self
rescue UnsafeForkError => e
  record_failure(e, phase: :after_fork)
  raise
rescue StandardError => e
  record_failure(e, phase: :after_fork)
  self
end

#before_fork!(timeout: nil) ⇒ Object 



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/julewire/ractor/destination.rb', line 140

def before_fork!(timeout: nil)
  timeout = lifecycle_timeout(timeout)
  @fork_lifecycle_mutex.synchronize do
    validate_before_fork_process!
    deadline = Core::Scheduling::Deadline.for(timeout)
    remaining_timeout = -> { Core::Scheduling::Deadline.remaining(deadline) }
    if closed?
      close_ports(timeout: remaining_timeout.call) if @worker
      return self
    end

    @closed.set(true)
    flush_before_fork!(remaining_timeout.call)
    stop_before_fork!(remaining_timeout.call)
  end
  self
rescue StandardError => e
  record_failure(e, phase: :before_fork)
  raise
end

#close(timeout: nil) ⇒ Object 



110
111
112
113
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
 
# File 'lib/julewire/ractor/destination.rb', line 110

def close(timeout: nil)
  timeout = lifecycle_timeout(timeout)
  @fork_lifecycle_mutex.synchronize do
    @closed.set(true)
    prepared_for_fork = @prepared_for_fork
    @prepared_for_fork = false
    unless @worker
      @owned_output_close_pending ||= prepared_for_fork && @close_output
      return true unless @owned_output_close_pending

      begin
        start_worker
      rescue StandardError => e
        record_failure(e, phase: :worker_start)
        return false
      end
    end

    result = request(:close, timeout: timeout, allow_closed: true)
    begin
      close_ports(timeout: timeout)
      @owned_output_close_pending = false if result
      result
    rescue Core::Error => e
      record_failure(e, phase: :worker_stop)
      false
    end
  end
end

#emit(record) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/julewire/ractor/destination.rb', line 92

def emit(record)
  increment(:received)
  outcome = enqueue(record)
  case outcome
  when :closed
    drop(:closed_dropped, record)
  when :queue_full
    drop(:queue_full_dropped, record)
  when StandardError
    record_failure(outcome, phase: :ractor_send)
    drop(:send_error, record)
  end
end

#flush(timeout: nil) ⇒ Object 



106
107
108
 
# File 'lib/julewire/ractor/destination.rb', line 106

def flush(timeout: nil)
  @health.recover_if_successful { request(:flush, timeout: lifecycle_timeout(timeout)) }
end

#health ⇒ Object 



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/julewire/ractor/destination.rb', line 174

def health
  worker = request(:health, timeout: @request_timeout)
  if worker.equal?(false)
    worker = @worker_health.get
  else
    raise TypeError, "ractor destination health must be a Hash" unless worker.instance_of?(Hash)

    Core::Integration::Protocol.validate_symbol_keys(worker)
    @worker_health.set(worker)
  end

  @health.snapshot(
    in_flight: @queue_slots.value,
    max_queue: @max_queue,
    status: status_for(worker),
    worker: worker
  )
end

#resource_identity ⇒ Object 



172
 
# File 'lib/julewire/ractor/destination.rb', line 172

def resource_identity = self