Class: Cosmo::API::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/cosmo/api/stream.rb,
sig/cosmo/api/stream.rbs

Constant Summary collapse

LIMIT =

Returns:

  • (::Integer) 
20

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name) ⇒ Stream

Returns a new instance of Stream.

Parameters:

  • name (::String) 


28
29
30
 
# File 'lib/cosmo/api/stream.rb', line 28

def initialize(name)
  @name = name
end

Instance Attribute Details

#name ⇒ ::String (readonly)

Returns the value of attribute name.

Returns:

  • (::String) 


26
27
28
 
# File 'lib/cosmo/api/stream.rb', line 26

def name
  @name
end

Class Method Details

.all ⇒ Array[Stream]

Returns:



12
13
14
 
# File 'lib/cosmo/api/stream.rb', line 12

def self.all
  client.list_streams.map { new(_1.dig("config", "name")) }
end

.client ⇒ Client

Returns:



22
23
24
 
# File 'lib/cosmo/api/stream.rb', line 22

def self.client
  @client ||= Client.instance
end

.jobs ⇒ Array[Stream]

Returns:



16
17
18
19
20
 
# File 'lib/cosmo/api/stream.rb', line 16

def self.jobs
  client.list_streams.select { _1.dig("config", "metadata", "_cosmo.type") == "jobs" }
                     .reject { %w[scheduled dead].include?(_1.dig("config", "name")) }
                     .map { new(_1.dig("config", "name")) }
end

Instance Method Details

#client ⇒ Client

Returns:



153
154
155
 
# File 'lib/cosmo/api/stream.rb', line 153

def client
  self.class.client
end

#delete(seq) ⇒ Object

Parameters:

  • seq (::Integer)

Returns:

  • (Object) 


108
109
110
 
# File 'lib/cosmo/api/stream.rb', line 108

def delete(seq)
  client.delete_message(name, seq)
end

#each {|arg0| ... } ⇒ void

This method returns an undefined value.

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/cosmo/api/stream.rb', line 52

def each
  return if total.zero?

  candidates = {}
  current, last, subjects = scan_range

  loop do
    break if current > last

    subject, job = next_candidate(subjects, candidates, current)
    break unless job

    candidates.delete(subject)
    current = job.seq.to_i + 1

    yield job
  end
end

#info ⇒ Hash[Symbol, untyped]

Returns:

  • (Hash[Symbol, untyped]) 


32
33
34
35
 
# File 'lib/cosmo/api/stream.rb', line 32

def info
  info = client.stream_info(name)
  { state: info.state, config: info.config }
end

#message(seq) ⇒ Job?

Parameters:

  • seq (::Integer)

Returns:



91
92
93
94
95
96
97
98
 
# File 'lib/cosmo/api/stream.rb', line 91

def message(seq)
  job = Job.new(name, client.get_message(name, seq: seq, direct: true))
  return if job.subject.to_s.start_with?(Cron::Entry::SUBJECT_PREFIX)

  job
rescue NATS::JetStream::Error::NotFound
  # nop, acked/nacked
end

#messages(page: nil, limit: nil) ⇒ Array[Job]

Parameters:

  • page: (::Integer, nil) (defaults to: nil)
  • limit: (::Integer, nil) (defaults to: nil)

Returns:



76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/cosmo/api/stream.rb', line 76

def messages(page: nil, limit: nil)
  jobs = []
  limit = (limit || LIMIT).to_i
  state = info[:state]
  start = state.first_seq.to_i
  start += (page.to_i - 1) * limit if page

  offset(start).each do |message|
    jobs << message
    break if jobs.size >= limit
  end

  jobs
end

#next_candidate(subjects, candidates, current) ⇒ [::String, Job]?

Lowest-seq message at or after current across all job subject filters, caching each subject's next candidate so it isn't re-queried every step.

Parameters:

  • subjects (Array[::String])
  • candidates (Hash[::String, Job?])
  • current (::Integer)

Returns:

  • ([::String, Job], nil) 


136
137
138
139
140
141
142
143
144
 
# File 'lib/cosmo/api/stream.rb', line 136

def next_candidate(subjects, candidates, current)
  subjects.each do |subject|
    next if candidates.key?(subject)

    candidates[subject] = next_message(subject, current)
  end

  candidates.compact.min_by { |_, msg| msg.seq.to_i }
end

#next_message(subject, seq) ⇒ Job?

Jump straight to the next message on the subject after seq, skipping any acked/deleted gaps

Parameters:

  • subject (::String)
  • seq (::Integer)

Returns:



147
148
149
150
151
 
# File 'lib/cosmo/api/stream.rb', line 147

def next_message(subject, seq)
  Job.new(name, client.get_message(name, next: true, seq: seq, subject: subject, direct: true))
rescue NATS::JetStream::Error::NotFound
  nil
end

#offset(value) ⇒ self

Parameters:

  • (::Integer)

Returns:

  • (self) 


71
72
73
74
 
# File 'lib/cosmo/api/stream.rb', line 71

def offset(value)
  @offset = value.to_i
  self
end

#pause! ⇒ Object

Returns:

  • (Object) 


112
113
114
 
# File 'lib/cosmo/api/stream.rb', line 112

def pause!
  client.pause_stream(name)
end

#paused? ⇒ Boolean

Returns:

  • (Boolean) 


120
121
122
 
# File 'lib/cosmo/api/stream.rb', line 120

def paused?
  client.stream_paused?(name)
end

#retries ⇒ ::Integer

Returns:

  • (::Integer) 


46
47
48
49
50
 
# File 'lib/cosmo/api/stream.rb', line 46

def retries
  client.list_consumers(name).sum { _1["num_redelivered"].to_i }
rescue NATS::Error
  0
end

#retry(seq) ⇒ void

This method returns an undefined value.

Parameters:

  • seq (::Integer) 


100
101
102
103
104
105
106
 
# File 'lib/cosmo/api/stream.rb', line 100

def retry(seq)
  job = message(seq)
  return unless job

  client.publish(job.x_subject, job.message.data)
  delete(seq)
end

#scan_range ⇒ [::Integer, ::Integer, Array[::String]]

Returns:

  • ([::Integer, ::Integer, Array[::String]]) 


126
127
128
129
130
131
132
 
# File 'lib/cosmo/api/stream.rb', line 126

def scan_range
  data = info
  state = data[:state]
  current = @offset || state.first_seq.to_i
  subjects = Array(data[:config].subjects).reject { _1.start_with?(Cron::Entry::SUBJECT_PREFIX) }
  [current, state.last_seq.to_i, subjects]
end

#total ⇒ ::Integer Also known as: size

Returns:

  • (::Integer) 


37
38
39
40
41
42
43
 
# File 'lib/cosmo/api/stream.rb', line 37

def total
  all_msgs = info[:state].messages.to_i
  cron_count = Client.instance.cron_subjects_in_stream(name, "#{Cron::Entry::SUBJECT_PREFIX}.#{name}.>").size
  [all_msgs - cron_count, 0].max
rescue NATS::Error
  0
end

#unpause! ⇒ Object

Returns:

  • (Object) 


116
117
118
 
# File 'lib/cosmo/api/stream.rb', line 116

def unpause!
  client.unpause_stream(name)
end