Class: Parse::Embeddings::StreamingBody

Inherits:
Object
  • Object
show all
Defined in:
lib/parse/embeddings/streaming_body.rb

Overview

An IO-shaped request body that splices base64-encoded files into a JSON envelope without ever holding a whole file in memory.

Multimodal endpoints want one JSON document with the media inlined as a data: URI. Building that with to_json costs roughly 2.4x the media size resident (raw bytes + the 1.33x base64 copy + the serialized JSON String), which is enough to OOM a small dyno on a single moderate video. This class instead emits the body as a stream of segments — literal JSON fragments interleaved with files that are read and encoded READ_CHUNK bytes at a time — so peak memory is bounded by the chunk size regardless of how large the media is, and nothing is spilled to disk either.

Faraday's net_http adapter assigns any body responding to #read to Net::HTTP::Request#body_stream, which pulls it incrementally. #size is exact, so callers can set Content-Length and avoid chunked transfer encoding (which some API gateways reject).

Base64 is spliced directly into the JSON string literal with no escaping: the alphabet (A-Za-z0-9+/=) contains no character that JSON requires escaping, so this is safe by construction.

Defined Under Namespace

Classes: SizeMismatch

Constant Summary collapse

READ_CHUNK =

Bytes read from a source file per fill. MUST stay a multiple of 3 so each chunk encodes to a padding-free base64 block and the concatenation is byte-identical to encoding the whole file at once. Only the final (short) chunk may carry = padding.

57 * 1024

Instance Method Summary collapse

Constructor Details

#initialize(segments) ⇒ StreamingBody

Returns a new instance of StreamingBody.

Parameters:

  • segments (Array<String, Hash>) —

    literal Strings are emitted verbatim; Hashes of the form { path: String, size: Integer } are base64-streamed.



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/parse/embeddings/streaming_body.rb', line 44

def initialize(segments)
  @segments = segments.map do |seg|
    case seg
    when String then seg.dup.force_encoding(Encoding::BINARY)
    when Hash
      unless seg[:path].is_a?(String) && seg[:size].is_a?(Integer)
        raise ArgumentError,
              "Parse::Embeddings::StreamingBody: file segment needs :path and :size."
      end
      seg
    else
      raise ArgumentError,
            "Parse::Embeddings::StreamingBody: segment must be a String or " \
            "{path:, size:} Hash (got #{seg.class})."
    end
  end
  rewind
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.



116
117
118
119
120
 
# File 'lib/parse/embeddings/streaming_body.rb', line 116

def close
  @io&.close
  @io = nil
  nil
end

#read(len = nil, out = nil) ⇒ String?

Returns nil once exhausted, per IO#read semantics.

Parameters:

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

    bytes wanted; nil reads to the end (which defeats the memory bound — Net::HTTP always passes a length, so this is only for completeness).

  • out (String, nil) (defaults to: nil) —

    optional output buffer to fill.

Returns:

  • (String, nil) —

    nil once exhausted, per IO#read semantics.



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/parse/embeddings/streaming_body.rb', line 79

def read(len = nil, out = nil)
  fill(len)
  if @buffer.empty?
    out&.clear
    return len.nil? ? "" : nil
  end

  chunk =
    if len.nil?
      b = @buffer
      @buffer = +""
      b
    else
      @buffer.slice!(0, len)
    end

  if out
    out.replace(chunk)
    out
  else
    chunk
  end
end

#rewind ⇒ void

This method returns an undefined value.

Reset to the start so the body can be replayed (Net::HTTP rewinds body_stream when it retries a request).



107
108
109
110
111
112
113
 
# File 'lib/parse/embeddings/streaming_body.rb', line 107

def rewind
  close
  @buffer = +""
  @index = 0
  @io = nil
  nil
end

#size ⇒ Integer Also known as: length

Exact byte length of the fully-emitted body. Base64 expands every 3 input bytes to 4 output bytes, padded up.

Returns:

  • (Integer) 


67
68
69
70
71
 
# File 'lib/parse/embeddings/streaming_body.rb', line 67

def size
  @size ||= @segments.sum do |seg|
    seg.is_a?(String) ? seg.bytesize : 4 * ((seg[:size] + 2) / 3)
  end
end