An Elixir atproto Jetstream V2 subscriber library
atproto jetstream elixir
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566defmodule Kite.Dictionary do @moduledoc false
@path "/xrpc/network.bsky.jetstream.getZstdDictionary" @timeout :timer.seconds(3)
def fetch(jetstream) do %{transport: transport, host: host, port: port} = jetstream
with {:ok, conn} <- Mint.HTTP.connect(transport, host, port, protocols: [:http1], mode: :passive), {:ok, conn, ref} <- request(conn), {:ok, body} <- receive_response(conn, ref, nil, []) do Mint.HTTP.close(conn) {:ok, IO.iodata_to_binary(body)} end end
defp request(conn) do with {:error, conn, reason} <- Mint.HTTP.request(conn, "GET", @path, [], nil) do Mint.HTTP.close(conn) {:error, reason} end end
defp receive_response(conn, ref, status, body) do case Mint.HTTP.recv(conn, 0, @timeout) do {:ok, conn, responses} -> case collect(responses, ref, status, body) do {:done, 200, body} -> {:ok, body}
{:done, status, _body} -> Mint.HTTP.close(conn) {:error, {:http_status, status}}
{:more, status, body} -> receive_response(conn, ref, status, body)
{:error, _reason} = error -> Mint.HTTP.close(conn) error end
{:error, conn, reason, _responses} -> Mint.HTTP.close(conn) {:error, reason} end end
defp collect([], _ref, status, body), do: {:more, status, body}
defp collect([{:status, ref, status} | rest], ref, _status, body), do: collect(rest, ref, status, body)
defp collect([{:headers, ref, _headers} | rest], ref, status, body), do: collect(rest, ref, status, body)
defp collect([{:data, ref, data} | rest], ref, status, body), do: collect(rest, ref, status, [body, data])
defp collect([{:done, ref} | _rest], ref, status, body), do: {:done, status, body}
defp collect([{:error, ref, reason} | _rest], ref, _status, _body), do: {:error, reason}end