class Thrift::HeaderTransport

HeaderTransport implements the THeader framing protocol.

THeader is a transport that adds headers and supports multiple protocols and transforms. It can auto-detect and communicate with legacy protocols (framed/unframed binary/compact) for backward compatibility.

Wire format:

+----------------------------------------------------------------+
| LENGTH (4 bytes, big-endian, excludes itself)                  |
+----------------------------------------------------------------+
| HEADER MAGIC (2 bytes: 0x0FFF) | FLAGS (2 bytes)               |
+----------------------------------------------------------------+
| SEQUENCE NUMBER (4 bytes)                                      |
+----------------------------------------------------------------+
| HEADER SIZE/4 (2 bytes)        | HEADER DATA (variable)...    |
+----------------------------------------------------------------+
| PAYLOAD (variable)                                             |
+----------------------------------------------------------------+

Constants

BINARY_VERSION_1
BINARY_VERSION_MASK: Binary protocol version mask and version 1
COMPACT_PROTOCOL_ID: Compact protocol ID
COMPACT_VERSION
COMPACT_VERSION_MASK
DEFAULT_MAX_DECOMPRESSED_SIZE: Default decompressed-size cap for ZLIB transform (~15.6 MB, matches other Thrift bindings)
DEFAULT_MAX_FRAME_SIZE: The largest frame accepted by default (~15.6 MB, matches other Thrift bindings). Defaulting to MAX_FRAME_SIZE let a four-byte header ask for a gigabyte, which is the ceiling rather than a limit.
HEADER_MAGIC: Header magic value (first 2 bytes of header)
MAX_FRAME_SIZE: The largest frame the length field can express (~1GB). This is the protocol’s ceiling, not a sensible default – see DEFAULT_MAX_FRAME_SIZE.
MAX_VARINT32_BYTES
MAX_VARINT32_LAST_BYTE

Attributes

flags [R]

max_frame_size [R]

The maximum frame size currently in force

protocol_id [R]

sequence_id [R]

Public Class Methods

new (transport, allowed_client_types = nil, default_protocol = HeaderSubprotocolID::COMPACT)

Source

    # File lib/thrift/transport/header_transport.rb
106 def initialize(transport, allowed_client_types = nil, default_protocol = HeaderSubprotocolID::COMPACT)
107   @transport = transport
108   @client_type = HeaderClientType::HEADERS
109   @protocol_id = default_protocol
110   @allowed_client_types = allowed_client_types || [
111     HeaderClientType::HEADERS,
112     HeaderClientType::FRAMED_BINARY,
113     HeaderClientType::UNFRAMED_BINARY,
114     HeaderClientType::FRAMED_COMPACT,
115     HeaderClientType::UNFRAMED_COMPACT,
116   ]
117 
118   @read_buffer = StringIO.new(Bytes.empty_byte_buffer)
119   @write_buffer = StringIO.new(Bytes.empty_byte_buffer)
120 
121   @read_headers = {}
122   @write_headers = {}
123   @write_transforms = []
124 
125   @sequence_id = 0
126   @flags = 0
127   @max_frame_size = DEFAULT_MAX_FRAME_SIZE
128   @max_decompressed_size = DEFAULT_MAX_DECOMPRESSED_SIZE
129   @unframed_bytes_read = 0
130 end

Creates a new HeaderTransport wrapping the given transport.

@param transport [BaseTransport] The underlying transport to wrap @param allowed_client_types [Array<Integer>] Allowed client types for auto-detection.

Defaults to all types for backward compatibility.

@param default_protocol [Integer] Default protocol ID (BINARY or COMPACT)

Public Instance Methods

add_transform (transform_id)

Source

    # File lib/thrift/transport/header_transport.rb
183 def add_transform(transform_id)
184   unless transform_id == HeaderTransformID::ZLIB
185     raise TransportException.new(TransportException::UNKNOWN, "Unknown transform: #{transform_id}")
186   end
187   @write_transforms << transform_id unless @write_transforms.include?(transform_id)
188 end

Adds a transform to apply when writing

@param transform_id [Integer] Transform ID (e.g., HeaderTransformID::ZLIB)

clear_headers ()

Source

    # File lib/thrift/transport/header_transport.rb
176 def clear_headers
177   @write_headers.clear
178 end

Clears all write headers

close ()

Source

    # File lib/thrift/transport/header_transport.rb
156 def close
157   @transport.close
158 end

flush ()

Source

    # File lib/thrift/transport/header_transport.rb
233 def flush
234   payload = @write_buffer.string
235   @write_buffer = StringIO.new(Bytes.empty_byte_buffer)
236 
237   return if payload.empty?
238 
239   case @client_type
240   when HeaderClientType::HEADERS
241     flush_header_format(payload)
242   when HeaderClientType::FRAMED_BINARY, HeaderClientType::FRAMED_COMPACT
243     validate_frame_size!(payload.bytesize)
244     flush_framed(payload)
245   when HeaderClientType::UNFRAMED_BINARY, HeaderClientType::UNFRAMED_COMPACT
246     validate_frame_size!(payload.bytesize)
247     @transport.write(payload)
248     @transport.flush
249   else
250     flush_header_format(payload)
251   end
252 end

get_headers ()

Source

    # File lib/thrift/transport/header_transport.rb
161 def get_headers
162   @read_headers
163 end

Returns the headers read from the last frame

message_boundaries? ()

Source

    # File lib/thrift/transport/header_transport.rb
265 def message_boundaries?
266   true
267 end

open ()

Source

    # File lib/thrift/transport/header_transport.rb
152 def open
153   @transport.open
154 end

open? ()

Source

    # File lib/thrift/transport/header_transport.rb
148 def open?
149   @transport.open?
150 end

protocol_id= (protocol_id)

Source

    # File lib/thrift/transport/header_transport.rb
140 def protocol_id=(protocol_id)
141   unless [HeaderSubprotocolID::BINARY, HeaderSubprotocolID::COMPACT].include?(protocol_id)
142     raise ArgumentError, "Unknown protocol ID: #{protocol_id}"
143   end
144 
145   @protocol_id = protocol_id
146 end

read (sz)

Source

    # File lib/thrift/transport/header_transport.rb
209 def read(sz)
210   # Try reading from existing buffer
211   data = @read_buffer.read(sz)
212   data = Bytes.empty_byte_buffer if data.nil?
213 
214   bytes_left = sz - data.bytesize
215   return data if bytes_left == 0
216 
217   # Handle unframed passthrough - read directly from underlying transport
218   if @client_type == HeaderClientType::UNFRAMED_BINARY ||
219      @client_type == HeaderClientType::UNFRAMED_COMPACT
220     return data + read_unframed(bytes_left)
221   end
222 
223   # Need to read the next frame
224   read_frame(bytes_left)
225   additional = @read_buffer.read(bytes_left)
226   data + (additional || Bytes.empty_byte_buffer)
227 end

reset_message_size ()

Source

    # File lib/thrift/transport/header_transport.rb
270 def reset_message_size
271   return unless @read_buffer.nil? || @read_buffer.eof?
272 
273   @unframed_bytes_read = 0
274 end

Starts a new protocol message without forcing client-type detection.

reset_protocol ()

Source

    # File lib/thrift/transport/header_transport.rb
259 def reset_protocol
260   return unless @read_buffer.nil? || @read_buffer.eof?
261 
262   read_frame(0)
263 end

Reads the next frame to detect protocol/client type before decoding.

sequence_id= (sequence_id)

Source

    # File lib/thrift/transport/header_transport.rb
132 def sequence_id=(sequence_id)
133   if sequence_id < -(2**31) || sequence_id > (2**31) - 1
134     raise RangeError, "sequence_id must be a signed int32"
135   end
136 
137   @sequence_id = sequence_id
138 end

set_header (key, value)

Source

    # File lib/thrift/transport/header_transport.rb
169 def set_header(key, value)
170   key = Bytes.force_binary_encoding(key.to_s)
171   value = Bytes.force_binary_encoding(value.to_s)
172   @write_headers[key] = value
173 end

Sets a header to be written with the next flush

@param key [String] Header key (must be binary string) @param value [String] Header value (must be binary string)

set_max_decompressed_size (size)

Source

    # File lib/thrift/transport/header_transport.rb
202 def set_max_decompressed_size(size)
203   if size <= 0 || size > MAX_FRAME_SIZE
204     raise ArgumentError, "max_decompressed_size must be > 0 and <= #{MAX_FRAME_SIZE}"
205   end
206   @max_decompressed_size = size
207 end

Sets the maximum allowed decompressed size for ZLIB transforms

set_max_frame_size (size)

Source

    # File lib/thrift/transport/header_transport.rb
194 def set_max_frame_size(size)
195   if size <= 0 || size > MAX_FRAME_SIZE
196     raise ArgumentError, "max_frame_size must be > 0 and <= #{MAX_FRAME_SIZE}"
197   end
198   @max_frame_size = size
199 end

Sets the maximum allowed frame size

to_s ()

Source

    # File lib/thrift/transport/header_transport.rb
254 def to_s
255   "header(#{@transport})"
256 end

write (buf)

Source

    # File lib/thrift/transport/header_transport.rb
229 def write(buf)
230   @write_buffer.write(Bytes.force_binary_encoding(buf))
231 end

Private Instance Methods

bounded_inflate (compressed)

Source

    # File lib/thrift/transport/header_transport.rb
457 def bounded_inflate(compressed)
458   inflater = Zlib::Inflate.new
459   buffer = Bytes.empty_byte_buffer
460   append_chunk = lambda do |chunk|
461     if buffer.bytesize + chunk.bytesize > @max_decompressed_size
462       raise TransportException.new(
463         TransportException::SIZE_LIMIT,
464         "Decompressed size exceeds limit of #{@max_decompressed_size}",
465       )
466     end
467 
468     buffer << chunk
469   end
470   begin
471     inflater.inflate(compressed, &append_chunk)
472     inflater.finish(&append_chunk)
473     buffer
474   rescue Zlib::DataError, Zlib::BufError
475     raise TransportException.new(TransportException::UNKNOWN, "Invalid ZLIB payload")
476   ensure
477     inflater.close rescue nil
478   end
479 end

Inflates compressed with a running byte-count check against @max_decompressed_size. Raises TransportException::SIZE_LIMIT if the decompressed output would exceed the limit.

flush_framed (payload)

Source

    # File lib/thrift/transport/header_transport.rb
545 def flush_framed(payload)
546   frame = [payload.bytesize].pack("N") + payload
547   @transport.write(frame)
548   @transport.flush
549 end

Flushes data in simple framed format (for legacy compatibility)

flush_header_format (payload)

Source

    # File lib/thrift/transport/header_transport.rb
482 def flush_header_format(payload)
483   # Apply transforms
484   @write_transforms.each do |transform_id|
485     if transform_id == HeaderTransformID::ZLIB
486       payload = Zlib::Deflate.deflate(payload)
487     end
488   end
489 
490   # Build header data
491   header_buf = StringIO.new(Bytes.empty_byte_buffer)
492 
493   # Protocol ID
494   write_varint32(header_buf, @protocol_id)
495 
496   # Transforms
497   write_varint32(header_buf, @write_transforms.size)
498   @write_transforms.each { |t| write_varint32(header_buf, t) }
499 
500   # Info headers (key-value pairs)
501   unless @write_headers.empty?
502     write_varint32(header_buf, HeaderInfoType::KEY_VALUE)
503     write_varint32(header_buf, @write_headers.size)
504     @write_headers.each do |key, value|
505       write_varstring(header_buf, key)
506       write_varstring(header_buf, value)
507     end
508   end
509 
510   # Pad header to 4-byte boundary
511   header_data = header_buf.string
512   padding = (4 - (header_data.bytesize % 4)) % 4
513   header_data += "\x00" * padding
514 
515   # Calculate total frame size (excludes the 4-byte length field itself)
516   # Frame = magic(2) + flags(2) + seqid(4) + header_len(2) + header_data + payload
517   frame_size = 2 + 2 + 4 + 2 + header_data.bytesize + payload.bytesize
518   validate_frame_size!(frame_size)
519   @write_headers = {}
520 
521   # Write complete frame
522   frame = Bytes.empty_byte_buffer
523   frame << [frame_size].pack("N")               # Length
524   frame << [HEADER_MAGIC].pack("n")             # Magic
525   frame << [@flags].pack("n")                   # Flags
526   frame << [unsigned_int32(@sequence_id)].pack("N") # Sequence ID
527   frame << [header_data.bytesize / 4].pack("n") # Header length (in 32-bit words)
528   frame << header_data                          # Header data
529   frame << payload                              # Payload
530 
531   @transport.write(frame)
532   @transport.flush
533 end

Flushes data in Header format

handle_unframed (first_word, req_sz)

Source

    # File lib/thrift/transport/header_transport.rb
359 def handle_unframed(first_word, req_sz)
360   @unframed_bytes_read = first_word.bytesize
361   raise_unframed_size_limit if @unframed_bytes_read > @max_frame_size
362 
363   bytes_left = req_sz - 4
364   if bytes_left > 0 && @unframed_bytes_read < @max_frame_size
365     rest = read_unframed(bytes_left)
366     @read_buffer = StringIO.new(first_word + rest)
367   else
368     @read_buffer = StringIO.new(first_word)
369   end
370 end

Handles unframed protocol - puts first_word back in buffer

parse_header_format (buf)

Source

    # File lib/thrift/transport/header_transport.rb
389 def parse_header_format(buf)
390   # Skip magic (already identified)
391   buf.read(2)
392 
393   # Read flags and sequence ID
394   @flags = buf.read(2).unpack1("n")
395   @sequence_id = signed_int32(buf.read(4).unpack1("N"))
396 
397   # Read header length (in 32-bit words)
398   header_words = buf.read(2).unpack1("n")
399   if header_words >= 16_384
400     raise TransportException.new(TransportException::UNKNOWN, "Header size is unreasonable")
401   end
402   header_length = header_words * 4
403   end_of_headers = buf.pos + header_length
404 
405   if end_of_headers > buf.string.bytesize
406     raise TransportException.new(TransportException::UNKNOWN, "Header size exceeds frame size")
407   end
408 
409   # Read protocol ID
410   @protocol_id = read_varint32(buf, end_of_headers)
411 
412   # Read transforms
413   transforms = []
414   transform_count = read_varint32(buf, end_of_headers)
415   transform_count.times do
416     transform_id = read_varint32(buf, end_of_headers)
417     unless transform_id == HeaderTransformID::ZLIB
418       raise TransportException.new(TransportException::UNKNOWN, "Unknown transform: #{transform_id}")
419     end
420     transforms << transform_id
421   end
422   # Read info headers
423   while buf.pos < end_of_headers
424     info_type = read_varint32(buf, end_of_headers)
425     if info_type == 0
426       # header padding
427       break
428     elsif info_type == HeaderInfoType::KEY_VALUE
429       count = read_varint32(buf, end_of_headers)
430       count.times do
431         key = read_varstring(buf, end_of_headers)
432         value = read_varstring(buf, end_of_headers)
433         @read_headers[key] = value
434       end
435     else
436       # Unknown info type, skip to end of headers
437       break
438     end
439   end
440 
441   # Skip any remaining header padding
442   buf.pos = end_of_headers
443 
444   # Read payload and apply transforms
445   payload = buf.read
446   transforms.each do |transform_id|
447     if transform_id == HeaderTransformID::ZLIB
448       payload = bounded_inflate(payload)
449     end
450   end
451 
452   StringIO.new(payload)
453 end

Parses a Header format frame

raise_unframed_size_limit ()

Source

    # File lib/thrift/transport/header_transport.rb
381 def raise_unframed_size_limit
382   raise TransportException.new(
383     TransportException::SIZE_LIMIT,
384     "Unframed message size exceeds maximum #{@max_frame_size}",
385   )
386 end

read_frame (req_sz)

Source

    # File lib/thrift/transport/header_transport.rb
287 def read_frame(req_sz)
288   @read_headers = {}
289   @unframed_bytes_read = 0
290 
291   # Read first 4 bytes - could be frame length or protocol magic
292   begin
293     first_word = @transport.read_all(4)
294   rescue EOFError
295     raise TransportException.new(TransportException::END_OF_FILE, "Unexpected EOF reading frame size")
296   end
297   frame_size = first_word.unpack1("N")
298 
299   # Check for unframed binary protocol
300   if (frame_size & BINARY_VERSION_MASK) == BINARY_VERSION_1
301     set_client_type(HeaderClientType::UNFRAMED_BINARY)
302     @protocol_id = HeaderSubprotocolID::BINARY
303     handle_unframed(first_word, req_sz)
304     return
305   end
306 
307   # Check for unframed compact protocol
308   if Bytes.get_string_byte(first_word, 0) == COMPACT_PROTOCOL_ID &&
309      (Bytes.get_string_byte(first_word, 1) & COMPACT_VERSION_MASK) == COMPACT_VERSION
310     set_client_type(HeaderClientType::UNFRAMED_COMPACT)
311     @protocol_id = HeaderSubprotocolID::COMPACT
312     handle_unframed(first_word, req_sz)
313     return
314   end
315 
316   # It's a framed protocol - validate frame size
317   if frame_size > @max_frame_size
318     raise TransportException.new(TransportException::UNKNOWN, "Frame size #{frame_size} exceeds maximum #{@max_frame_size}")
319   end
320   if frame_size < 4
321     raise TransportException.new(TransportException::UNKNOWN, "Frame size #{frame_size} is too small")
322   end
323 
324   # Read the complete frame
325   begin
326     frame_data = @transport.read_all(frame_size)
327   rescue EOFError
328     raise TransportException.new(TransportException::END_OF_FILE, "Unexpected EOF reading frame")
329   end
330   frame_buf = StringIO.new(frame_data)
331 
332   # Check the second word for protocol type
333   second_word = frame_buf.read(4)
334   frame_buf.rewind
335 
336   magic = second_word.unpack1("n")
337 
338   if magic == HEADER_MAGIC
339     if frame_size < 10
340       raise TransportException.new(TransportException::UNKNOWN, "Header transport frame is too small")
341     end
342     set_client_type(HeaderClientType::HEADERS)
343     @read_buffer = parse_header_format(frame_buf)
344   elsif (second_word.unpack1("N") & BINARY_VERSION_MASK) == BINARY_VERSION_1
345     set_client_type(HeaderClientType::FRAMED_BINARY)
346     @protocol_id = HeaderSubprotocolID::BINARY
347     @read_buffer = frame_buf
348   elsif Bytes.get_string_byte(second_word, 0) == COMPACT_PROTOCOL_ID &&
349         (Bytes.get_string_byte(second_word, 1) & COMPACT_VERSION_MASK) == COMPACT_VERSION
350     set_client_type(HeaderClientType::FRAMED_COMPACT)
351     @protocol_id = HeaderSubprotocolID::COMPACT
352     @read_buffer = frame_buf
353   else
354     raise TransportException.new(TransportException::UNKNOWN, "Could not detect client transport type")
355   end
356 end

Reads the next frame, detecting client type on first read

read_unframed (size)

Source

    # File lib/thrift/transport/header_transport.rb
372 def read_unframed(size)
373   remaining = @max_frame_size - @unframed_bytes_read
374   raise_unframed_size_limit if remaining <= 0
375 
376   data = @transport.read([size, remaining].min)
377   @unframed_bytes_read += data.bytesize
378   data
379 end

read_varint32 (io, boundary_pos = nil)

Source

    # File lib/thrift/transport/header_transport.rb
552 def read_varint32(io, boundary_pos = nil)
553   shift = 0
554   result = 0
555   loop do
556     if boundary_pos && io.pos >= boundary_pos
557       raise TransportException.new(TransportException::UNKNOWN, "Trying to read past header boundary")
558     end
559     byte = io.getbyte
560     raise TransportException.new(TransportException::END_OF_FILE, "Unexpected EOF reading varint") if byte.nil?
561 
562     if shift == (MAX_VARINT32_BYTES - 1) * 7
563       if (byte & 0x80) != 0
564         raise TransportException.new(TransportException::UNKNOWN, "Variable-length int over 5 bytes.")
565       end
566       if (byte & ~MAX_VARINT32_LAST_BYTE) != 0
567         raise TransportException.new(TransportException::UNKNOWN, "Variable-length int overflows uint32.")
568       end
569     end
570 
571     result |= (byte & 0x7f) << shift
572     return result if (byte & 0x80) == 0
573     shift += 7
574   end
575 end

Reads a varint32 from the given IO

read_varstring (io, boundary_pos = nil)

Source

    # File lib/thrift/transport/header_transport.rb
596 def read_varstring(io, boundary_pos = nil)
597   size = read_varint32(io, boundary_pos)
598   if size < 0
599     raise TransportException.new(TransportException::UNKNOWN, "Negative string size: #{size}")
600   end
601   if boundary_pos && size > (boundary_pos - io.pos)
602     raise TransportException.new(TransportException::UNKNOWN, "Info header length exceeds header size")
603   end
604   data = io.read(size)
605   if data.nil? || data.bytesize < size
606     raise TransportException.new(TransportException::END_OF_FILE, "Unexpected EOF reading string")
607   end
608   data
609 end

Reads a varint-prefixed string from the given IO

set_client_type (client_type)

Source

    # File lib/thrift/transport/header_transport.rb
279 def set_client_type(client_type)
280   unless @allowed_client_types.include?(client_type)
281     raise TransportException.new(TransportException::UNKNOWN, "Client type #{client_type} not allowed by server")
282   end
283   @client_type = client_type
284 end

Sets the client type after validation

signed_int32 (value)

Source

    # File lib/thrift/transport/header_transport.rb
618 def signed_int32(value)
619   value > 0x7fffffff ? value - (2**32) : value
620 end

unsigned_int32 (value)

Source

    # File lib/thrift/transport/header_transport.rb
622 def unsigned_int32(value)
623   value < 0 ? value + (2**32) : value
624 end

validate_frame_size! (frame_size)

Source

    # File lib/thrift/transport/header_transport.rb
535 def validate_frame_size!(frame_size)
536   return if frame_size <= @max_frame_size
537 
538   raise TransportException.new(
539     TransportException::UNKNOWN,
540     "Frame size #{frame_size} exceeds maximum #{@max_frame_size}",
541   )
542 end

write_varint32 (io, n)

Source

    # File lib/thrift/transport/header_transport.rb
578 def write_varint32(io, n)
579   # Encode the low 32 bits: Ruby's >> is arithmetic, so a negative value
580   # would keep its sign bit set and never satisfy the (n & ~0x7F) == 0 exit
581   # condition below. Masking yields the same unsigned varint read_varint32
582   # produces and accepts.
583   n &= 0xFFFFFFFF
584   loop do
585     if (n & ~0x7F) == 0
586       io.write([n].pack("C"))
587       break
588     else
589       io.write([(n & 0x7F) | 0x80].pack("C"))
590       n >>= 7
591     end
592   end
593 end

Writes a varint32 to the given IO

write_varstring (io, value)

Source

    # File lib/thrift/transport/header_transport.rb
612 def write_varstring(io, value)
613   value = Bytes.force_binary_encoding(value)
614   write_varint32(io, value.bytesize)
615   io.write(value)
616 end

Writes a varint-prefixed string to the given IO