Class: Julewire::Core::Destinations::Destination

Inherits:
Object
  • Object
show all
Includes:
ProcessorHandling
Defined in:
lib/julewire/core/destinations/destination.rb

Constant Summary collapse

COUNTER_KEYS = 
%i[
  callback_error
  encode_error
  formatter_error
  formatted
  output_accepted
  output_error
  output_exception
  output_rejected
  processor_dropped
  processor_error
  processor_invalid
  received
  record_too_large
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, close_output:, encoder:, formatter:, max_record_bytes:, on_drop:, on_failure:, output:, processors: nil) ⇒ Destination

rubocop:disable Metrics/ParameterLists -- Destination definitions pass normalized settings.

Raises:

  • (ArgumentError) 


27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/julewire/core/destinations/destination.rb', line 27

def initialize( # rubocop:disable Metrics/ParameterLists -- Destination definitions pass normalized settings.
  name:,
  close_output:,
  encoder:,
  formatter:,
  max_record_bytes:,
  on_drop:,
  on_failure:,
  output:,
  processors: nil
)
  @name = Destinations.normalize_name(name)
  @formatter = validate_callable(formatter, name: :formatter)
  @encoder = validate_callable(encoder, name: :encoder)
  Validation.validate_byte_limit!(max_record_bytes, name: :max_record_bytes)
  @max_record_bytes = max_record_bytes
  @on_drop = validate_optional_callback(on_drop, name: :on_drop)
  @on_failure = validate_optional_callback(on_failure, name: :on_failure)
  raise ArgumentError, "destination #{@name.inspect} output is required" if output.nil?

  @output = Sink.wrap(output, close_output: close_output)
  @processor_chain = processor_chain(processors)
  initialize_tracking
  @write_step = build_write_step
end

Instance Attribute Details

#name ⇒ Object (readonly) 



25
26
27
 
# File 'lib/julewire/core/destinations/destination.rb', line 25

def name
  @name
end

Instance Method Details

#after_fork! ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/julewire/core/destinations/destination.rb', line 73

def after_fork!
  initialize_tracking
  @output.after_fork!
  self
rescue StandardError => e
  notify_failure(
    e,
    action: :after_fork,
    output_class: output_class_name,
    phase: :output_lifecycle
  )
  self
end

#close(timeout: nil) ⇒ Object 



69
70
71
 
# File 'lib/julewire/core/destinations/destination.rb', line 69

def close(timeout: nil)
  call_output_lifecycle(:close, timeout: timeout)
end

#emit(record) ⇒ Object 



53
54
55
56
57
58
59
 
# File 'lib/julewire/core/destinations/destination.rb', line 53

def emit(record)
  degradation_marker = @health.degradation_marker
  record = process_record(record)
  return unless record

  emit_processed_record(record, degradation_marker: degradation_marker)
end

#emit_processed_record(record, degradation_marker:) ⇒ Object 



61
62
63
 
# File 'lib/julewire/core/destinations/destination.rb', line 61

def emit_processed_record(record, degradation_marker:)
  clear_degradation_if_unchanged(degradation_marker) if @write_step.call(record)
end

#flush(timeout: nil) ⇒ Object 



65
66
67
 
# File 'lib/julewire/core/destinations/destination.rb', line 65

def flush(timeout: nil)
  call_output_lifecycle(:flush, timeout: timeout)
end

#health ⇒ Object 



91
92
93
94
95
96
97
98
99
 
# File 'lib/julewire/core/destinations/destination.rb', line 91

def health
  {
    counts: counts_health,
    last_callback_failure: @health.last_callback_failure,
    last_failure: @health.last_failure,
    last_loss: @health.last_loss,
    status: degraded? ? :degraded : :ok
  }
end

#resource_identity ⇒ Object 



87
88
89
 
# File 'lib/julewire/core/destinations/destination.rb', line 87

def resource_identity
  @output.resource_identity
end