Class: Julewire::Core::Destinations::Collection

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

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(destinations, on_drop:, on_failure:) ⇒ Collection

Returns a new instance of Collection.



8
9
10
11
12
13
 
# File 'lib/julewire/core/destinations/collection.rb', line 8

def initialize(destinations, on_drop:, on_failure:)
  @destinations = destinations.dup.freeze
  @on_drop = on_drop
  @on_failure = on_failure
  @prepared_destinations = []
end

Class Method Details

.build(configuration:, defaults:, on_drop:, on_failure:) ⇒ Object 



16
17
18
19
20
21
22
 
# File 'lib/julewire/core/destinations/collection.rb', line 16

def build(configuration:, defaults:, on_drop:, on_failure:)
  new(
    validate_destinations(configuration.destinations.build(defaults: defaults)),
    on_drop: on_drop,
    on_failure: on_failure
  )
end

Instance Method Details

#after_fork! ⇒ Object 



41
42
43
44
45
46
47
 
# File 'lib/julewire/core/destinations/collection.rb', line 41

def after_fork!
  @destinations.each do |destination|
    call_destination_after_fork(destination)
  end
  @prepared_destinations = []
  self
end

#before_fork!(timeout:) ⇒ Object 



49
50
51
52
53
54
55
 
# File 'lib/julewire/core/destinations/collection.rb', line 49

def before_fork!(timeout:)
  Validation.validate_timeout!(timeout, name: :timeout)
  return self unless @prepared_destinations.empty?

  prepare_destinations_before_fork(timeout)
  self
end

#cancel_before_fork! ⇒ Object 



57
58
59
60
61
 
# File 'lib/julewire/core/destinations/collection.rb', line 57

def cancel_before_fork!
  @prepared_destinations.reverse_each { call_destination_after_fork(it) }
  @prepared_destinations = []
  self
end

#close(timeout: nil, skip_resource_identities: nil) ⇒ Object 



67
68
69
 
# File 'lib/julewire/core/destinations/collection.rb', line 67

def close(timeout: nil, skip_resource_identities: nil)
  call_lifecycle(:close, timeout: timeout, skip_resource_identities: skip_resource_identities)
end

#emit(record) ⇒ Object 



35
36
37
38
39
 
# File 'lib/julewire/core/destinations/collection.rb', line 35

def emit(record)
  @destinations.each do |destination|
    emit_to_destination(destination, record)
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


33
 
# File 'lib/julewire/core/destinations/collection.rb', line 33

def empty? = @destinations.empty?

#flush(timeout: nil) ⇒ Object 



63
64
65
 
# File 'lib/julewire/core/destinations/collection.rb', line 63

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

#health ⇒ Object 



77
78
79
 
# File 'lib/julewire/core/destinations/collection.rb', line 77

def health
  @destinations.to_h { [destination_name(it), destination_health(it)] }
end

#lifecycle_resource_identities ⇒ Object 



71
72
73
74
75
 
# File 'lib/julewire/core/destinations/collection.rb', line 71

def lifecycle_resource_identities
  @destinations.each_with_object({}.compare_by_identity) do |destination, identities|
    identities[resource_identity(destination)] = true
  end
end