Class: A2A::Utils::MessageBuffer

Inherits:
Object
  • Object
show all
Defined in:
lib/a2a/utils/message_buffer.rb

Constant Summary collapse

DEFAULT_BUFFER_SIZE =

Default buffer size (64KB)

64 * 1024
COMPRESSION_THRESHOLD =

Compression threshold (1MB)

1024 * 1024

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(initial_capacity: DEFAULT_BUFFER_SIZE, compress: true, encoding: Encoding::UTF_8) ⇒ MessageBuffer

Initialize a new message buffer

Parameters:

  • initial_capacity (Integer) (defaults to: DEFAULT_BUFFER_SIZE) —

    Initial buffer capacity

  • compress (Boolean) (defaults to: true) —

    Whether to use compression for large messages

  • encoding (Encoding) (defaults to: Encoding::UTF_8) —

    Text encoding to use



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/a2a/utils/message_buffer.rb', line 30

def initialize(initial_capacity: DEFAULT_BUFFER_SIZE, compress: true, encoding: Encoding::UTF_8)
  @buffer = StringIO.new
  @buffer.set_encoding(encoding)
  @initial_capacity = initial_capacity
  @compress = compress
  @compressed = false
  @encoding = encoding
  @chunks = []
  @total_size = 0
end

Instance Attribute Details

#compressed ⇒ Object (readonly)

Returns the value of attribute compressed.



22
23
24
 
# File 'lib/a2a/utils/message_buffer.rb', line 22

def compressed
  @compressed
end

#encoding ⇒ Object (readonly)

Returns the value of attribute encoding.



22
23
24
 
# File 'lib/a2a/utils/message_buffer.rb', line 22

def encoding
  @encoding
end

#size ⇒ Integer (readonly)

Get the current size of the buffer

Returns:

  • (Integer) —

    Size in bytes



90
91
92
 
# File 'lib/a2a/utils/message_buffer.rb', line 90

def size
  @size
end

Class Method Details

.create_memory_mapped(file_path) ⇒ MessageBuffer

Create a memory-mapped buffer for very large data

Parameters:

  • file_path (String) —

    Path to temporary file

Returns:



248
249
250
251
252
253
254
255
 
# File 'lib/a2a/utils/message_buffer.rb', line 248

def self.create_memory_mapped(file_path)
  # This would require additional gems like 'mmap' for true memory mapping
  # For now, we'll use a file-backed buffer
  buffer = new
  buffer.instance_variable_set(:@file_backed, true)
  buffer.instance_variable_set(:@file_path, file_path)
  buffer
end

.from_json(json_string) ⇒ MessageBuffer

Create buffer from JSON string

Parameters:

  • json_string (String) —

    JSON string

Returns:



135
136
137
138
139
140
 
# File 'lib/a2a/utils/message_buffer.rb', line 135

def self.from_json(json_string)
  data = A2A::Utils::Performance.optimized_json_parse(json_string)
  buffer = new
  buffer.write(data.to_s)
  buffer
end

Instance Method Details

#<<(data) ⇒ Integer

Append data to the buffer (alias for write)

Parameters:

  • data (String) —

    Data to append

Returns:

  • (Integer) —

    Number of bytes written



68
69
70
 
# File 'lib/a2a/utils/message_buffer.rb', line 68

def <<(data)
  write(data)
end

#clear! ⇒ Object

Clear the buffer



105
106
107
108
109
110
111
112
 
# File 'lib/a2a/utils/message_buffer.rb', line 105

def clear!
  @buffer = StringIO.new
  @buffer.set_encoding(@encoding)
  @compressed = false
  @compressed_size = 0
  @total_size = 0
  @chunks.clear
end

#compress! ⇒ Object

Compress buffer contents



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/a2a/utils/message_buffer.rb', line 164

def compress!
  return if @compressed || @buffer.size < COMPRESSION_THRESHOLD

  @buffer.rewind
  original_data = @buffer.read
  @buffer.rewind

  compressed_data = Zlib::Deflate.deflate(original_data)

  # Only use compression if it actually saves space
  return unless compressed_data.size < original_data.size

  @buffer = StringIO.new(compressed_data)
  @compressed = true
  @compressed_size = compressed_data.size
  @original_size = original_data.size
end

#compression_ratio ⇒ Float

Get compression ratio

Returns:

  • (Float) —

    Compression ratio (0.0 to 1.0)



207
208
209
210
211
 
# File 'lib/a2a/utils/message_buffer.rb', line 207

def compression_ratio
  return 0.0 unless @compressed && @original_size&.positive?

  1.0 - (@compressed_size.to_f / @original_size)
end

#decompress! ⇒ Object

Decompress buffer contents



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/a2a/utils/message_buffer.rb', line 185

def decompress!
  return unless @compressed

  @buffer.rewind
  compressed_data = @buffer.read
  @buffer.rewind

  original_data = Zlib::Inflate.inflate(compressed_data)

  @buffer = StringIO.new
  @buffer.set_encoding(@encoding)
  @buffer.write(original_data)
  @buffer.rewind

  @compressed = false
  @total_size = original_data.size
end

#each_chunk(chunk_size = DEFAULT_BUFFER_SIZE) {|String| ... } ⇒ Object

Stream data in chunks

Parameters:

  • chunk_size (Integer) (defaults to: DEFAULT_BUFFER_SIZE) —

    Size of each chunk

Yields:

  • (String) —

    Each chunk of data



147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/a2a/utils/message_buffer.rb', line 147

def each_chunk(chunk_size = DEFAULT_BUFFER_SIZE)
  return enum_for(:each_chunk, chunk_size) unless block_given?

  decompress! if @compressed

  @buffer.rewind

  while (chunk = @buffer.read(chunk_size))
    yield chunk
  end

  @buffer.rewind
end

#empty? ⇒ Boolean

Check if buffer is empty

Returns:

  • (Boolean) —

    True if buffer is empty



98
99
100
 
# File 'lib/a2a/utils/message_buffer.rb', line 98

def empty?
  size.zero?
end

#ensure_capacity(capacity) ⇒ Object (private)

Ensure buffer has minimum capacity

Parameters:

  • capacity (Integer) —

    Required capacity



263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/a2a/utils/message_buffer.rb', line 263

def ensure_capacity(capacity)
  return if @buffer.size >= capacity

  # Expand buffer if needed
  current_pos = @buffer.pos
  @buffer.seek(0, IO::SEEK_END)

  # Write null bytes to expand
  padding_needed = capacity - @buffer.size
  @buffer.write("\0" * padding_needed) if padding_needed.positive?

  @buffer.pos = current_pos
end

#optimize! ⇒ Object

Optimize buffer for memory usage



230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/a2a/utils/message_buffer.rb', line 230

def optimize!
  # Force garbage collection on buffer
  GC.start if defined?(GC)

  # Compress if beneficial
  compress! if @compress && !@compressed && size > COMPRESSION_THRESHOLD

  # Compact string if possible (Ruby 2.7+)
  return unless @buffer.string.respond_to?(:squeeze!)

  @buffer.string.squeeze!
end

#read(length = nil) ⇒ String

Read data from the buffer

Parameters:

  • length (Integer, nil) (defaults to: nil) —

    Number of bytes to read (nil for all)

Returns:

  • (String) —

    Data read from buffer



77
78
79
80
81
82
83
84
 
# File 'lib/a2a/utils/message_buffer.rb', line 77

def read(length = nil)
  decompress! if @compressed

  @buffer.rewind
  data = @buffer.read(length)
  @buffer.rewind
  data
end

#stats ⇒ Hash

Get buffer statistics

Returns:

  • (Hash) —

    Buffer statistics



217
218
219
220
221
222
223
224
225
 
# File 'lib/a2a/utils/message_buffer.rb', line 217

def stats
  {
    size: size,
    compressed: @compressed,
    compression_ratio: compression_ratio,
    encoding: @encoding.name,
    chunks: @chunks.size
  }
end

#to_json(*_args) ⇒ String

Get buffer contents as JSON

Returns:

  • (String) —

    JSON representation



126
127
128
 
# File 'lib/a2a/utils/message_buffer.rb', line 126

def to_json(*_args)
  A2A::Utils::Performance.optimized_json_generate(to_s)
end

#to_s ⇒ String

Get buffer contents as string

Returns:

  • (String) —

    Buffer contents



118
119
120
 
# File 'lib/a2a/utils/message_buffer.rb', line 118

def to_s
  read
end

#write(data) ⇒ Integer

Write data to the buffer

Parameters:

  • data (String) —

    Data to write

Returns:

  • (Integer) —

    Number of bytes written



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/a2a/utils/message_buffer.rb', line 46

def write(data)
  data = data.to_s.encode(@encoding) unless data.encoding == @encoding

  if @compressed
    # If already compressed, decompress first
    decompress!
  end

  bytes_written = @buffer.write(data)
  @total_size += bytes_written

  # Compress if buffer gets too large
  compress! if @compress && @total_size > COMPRESSION_THRESHOLD && !@compressed

  bytes_written
end