Module: MisarMail::Core::SSE

Defined in:
lib/misar_mail/core/sse.rb

Overview

Server-Sent Events client for the MisarMail streaming endpoints.

Both streams frame events as "data: " and close with the sentinel "data: [DONE]". One of the two is a POST, so this reads the response body incrementally rather than using an EventSource-style helper.

Constant Summary collapse

DONE = 
"[DONE]"

Class Method Summary collapse

Class Method Details

.stream(url, api_key, method: "GET", body: nil) ⇒ Object

Yields one decoded payload per event until the stream terminates.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/misar_mail/core/sse.rb', line 20

def stream(url, api_key, method: "GET", body: nil)
  return enum_for(:stream, url, api_key, method: method, body: body) unless block_given?

  uri = URI.parse(url)
  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = uri.scheme == "https"
  http.read_timeout = 300

  headers = {
    "Authorization" => "Bearer #{api_key}",
    "Accept" => "text/event-stream"
  }
  headers["Content-Type"] = "application/json" unless body.nil?

  request = Net::HTTP.const_get(method.capitalize).new(uri.request_uri, headers)
  request.body = JSON.generate(body) unless body.nil?

  http.request(request) do |response|
    # Errors arrive as a normal JSON body, not as an SSE frame.
    if response.code.to_i >= 400
      data = begin
        JSON.parse(response.read_body)
      rescue StandardError
        {}
      end
      raise MisarMail::Error.new(response.code.to_i, data["error"].to_s, "api_error", data)
    end

    buffer = +""
    response.read_body do |chunk|
      buffer << chunk
      while (index = buffer.index("\n"))
        line = buffer.slice!(0..index).chomp
        next unless line.start_with?("data:")

        payload = line[5..].strip
        return if payload == DONE
        next if payload.empty?

        begin
          yield JSON.parse(payload)
        rescue JSON::ParserError
          # One malformed frame should not discard everything already
          # streamed.
          yield({ "raw" => payload })
        end
      end
    end
  end
end