Class: KubeMQ::Transport::MessageBuffer Private

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/transport/message_buffer.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Note:

This class is thread-safe. All operations are mutex-protected.

Thread-safe bounded buffer for queuing messages during reconnection.

When the transport is disconnected, outbound messages are buffered here. If the buffer reaches capacity, the oldest message is dropped to make room -- consistent with other KubeMQ SDKs.

See Also:

Constant Summary collapse

DEFAULT_CAPACITY =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Default buffer capacity.

1000

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(capacity: DEFAULT_CAPACITY) ⇒ MessageBuffer

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of MessageBuffer.

Parameters:

  • capacity (Integer) (defaults to: DEFAULT_CAPACITY) —

    maximum buffer size (default: DEFAULT_CAPACITY)



24
25
26
27
28
 
# File 'lib/kubemq/transport/message_buffer.rb', line 24

def initialize(capacity: DEFAULT_CAPACITY)
  @capacity = capacity
  @queue = Thread::SizedQueue.new(capacity)
  @mutex = Mutex.new
end

Instance Attribute Details

#capacity ⇒ Integer (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns maximum number of messages this buffer can hold.

Returns:

  • (Integer) —

    maximum number of messages this buffer can hold



21
22
23
 
# File 'lib/kubemq/transport/message_buffer.rb', line 21

def capacity
  @capacity
end

Instance Method Details

#clear ⇒ nil

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Removes all messages from the buffer.

Returns:

  • (nil) 


99
100
101
102
 
# File 'lib/kubemq/transport/message_buffer.rb', line 99

def clear
  drain
  nil
end

#drain {|message| ... } ⇒ Array?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Drains all buffered messages. If a block is given, yields each message; otherwise returns them as an array via #to_a.

Yields:

  • (message) —

    each buffered message in FIFO order

Yield Parameters:

  • message (Object) —

    a buffered message

Returns:

  • (Array, nil) —

    array of messages if no block given; nil otherwise



55
56
57
58
59
60
61
62
63
64
 
# File 'lib/kubemq/transport/message_buffer.rb', line 55

def drain(&block)
  return to_a unless block

  loop do
    msg = @queue.pop(true)
    block.call(msg)
  rescue ThreadError
    break
  end
end

#empty? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns whether the buffer is empty.

Returns:

  • (Boolean) —

    true if no messages are buffered



85
86
87
 
# File 'lib/kubemq/transport/message_buffer.rb', line 85

def empty?
  @queue.empty?
end

#full? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns whether the buffer has reached capacity.

Returns:



92
93
94
 
# File 'lib/kubemq/transport/message_buffer.rb', line 92

def full?
  @queue.size >= @capacity
end

#push(message) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Adds a message to the buffer. If full, the oldest message is dropped.

Parameters:

  • message (Object) —

    the message to buffer



34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/kubemq/transport/message_buffer.rb', line 34

def push(message)
  @mutex.synchronize do
    begin
      @queue.pop(true) if @queue.size >= @capacity
    rescue ThreadError
      # queue was emptied concurrently
    end
    begin
      @queue.push(message, true)
    rescue ThreadError
      # queue full; discard
    end
  end
end

#size ⇒ Integer

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the current number of buffered messages.

Returns:

  • (Integer) —

    number of messages in the buffer



78
79
80
 
# File 'lib/kubemq/transport/message_buffer.rb', line 78

def size
  @queue.size
end

#to_a ⇒ Array<Object>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Drains all messages and returns them as an array.

Returns:

  • (Array<Object>) —

    all buffered messages in FIFO order



69
70
71
72
73
 
# File 'lib/kubemq/transport/message_buffer.rb', line 69

def to_a
  messages = []
  drain { |m| messages << m }
  messages
end