Class: LogStash::Util::WrappedAckedQueue::ReadBatch
- Inherits:
-
Object
- Object
- LogStash::Util::WrappedAckedQueue::ReadBatch
- Defined in:
- lib/logstash/util/wrapped_acked_queue.rb
Instance Method Summary collapse
- #cancel(event) ⇒ Object
- #cancelled_size ⇒ Object
- #close ⇒ Object
- #each(&blk) ⇒ Object
- #filtered_size ⇒ Object
- #flush_signal_received? ⇒ Boolean
-
#initialize(queue, size, wait) ⇒ ReadBatch
constructor
A new instance of ReadBatch.
- #merge(event) ⇒ Object
- #shutdown_signal_received? ⇒ Boolean
- #size ⇒ Object
- #starting_size ⇒ Object
Constructor Details
#initialize(queue, size, wait) ⇒ ReadBatch
Returns a new instance of ReadBatch.
204 205 206 207 208 209 210 211 212 213 214 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 204 def initialize(queue, size, wait) @originals = Hash.new # TODO: disabled for https://github.com/elastic/logstash/issues/6055 - will have to properly refactor # @cancelled = Hash.new @generated = Hash.new @iterating_temp = Hash.new @iterating = false # Atomic Boolean maybe? Although batches are not shared across threads take_originals_from_queue(queue, size, wait) # this sets a reference to @acked_batch end |
Instance Method Details
#cancel(event) ⇒ Object
235 236 237 238 239 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 235 def cancel(event) # TODO: disabled for https://github.com/elastic/logstash/issues/6055 - will have to properly refactor raise("cancel is unsupported") # @cancelled[event] = true end |
#cancelled_size ⇒ Object
269 270 271 272 273 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 269 def cancelled_size # TODO: disabled for https://github.com/elastic/logstash/issues/6055 = will have to properly refactor raise("cancelled_size is unsupported ") # @cancelled.size end |
#close ⇒ Object
216 217 218 219 220 221 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 216 def close # this will ack the whole batch, regardless of whether some # events were cancelled or failed return if @acked_batch.nil? @acked_batch.close end |
#each(&blk) ⇒ Object
241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 241 def each(&blk) # take care not to cause @originals or @generated to change during iteration # below the checks for @cancelled.include?(e) have been replaced by e.cancelled? # TODO: for https://github.com/elastic/logstash/issues/6055 = will have to properly refactor @iterating = true @originals.each do |e, _| blk.call(e) unless e.cancelled? end @generated.each do |e, _| blk.call(e) unless e.cancelled? end @iterating = false update_generated end |
#filtered_size ⇒ Object
265 266 267 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 265 def filtered_size @originals.size + @generated.size end |
#flush_signal_received? ⇒ Boolean
279 280 281 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 279 def flush_signal_received? false end |
#merge(event) ⇒ Object
223 224 225 226 227 228 229 230 231 232 233 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 223 def merge(event) return if event.nil? || @originals.key?(event) # take care not to cause @generated to change during iteration # @iterating_temp is merged after the iteration if @iterating_temp[event] = true else # the periodic flush could generate events outside of an each iteration @generated[event] = true end end |
#shutdown_signal_received? ⇒ Boolean
275 276 277 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 275 def shutdown_signal_received? false end |
#size ⇒ Object
257 258 259 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 257 def size filtered_size end |
#starting_size ⇒ Object
261 262 263 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 261 def starting_size @originals.size end |