URP.Stream (urp v0.10.2)

Copy Markdown

Bidirectional URP dispatch for XInputStream/XOutputStream/XSeekable.

When we pass a stream object reference to soffice (e.g. as an InputStream in a MediaDescriptor), soffice calls methods on our object over the same TCP connection. This module handles those incoming calls while we wait for the reply to our original request.

XInputStream funcIDs (depth-first interface traversal)

XInterface: queryInterface=0, acquire=1, release=2
XInputStream: readBytes=3, readSomeBytes=4, skipBytes=5, available=6, closeInput=7

XSeekable funcIDs

XInterface: queryInterface=0, acquire=1, release=2
XSeekable: seek=3, getPosition=4, getLength=5

Since XInputStream and XSeekable share funcIDs 3-5, we use the URP type cache to disambiguate: each request carries a type reference telling us which interface it targets.

XOutputStream funcIDs

XInterface: queryInterface=0, acquire=1, release=2
XOutputStream: writeBytes=3, flush=4, closeOutput=5

Summary

Types

connection()
input_source()
sink()

Functions

recv_handling_input(conn, data, stream_oid \\ nil)

Receive frames while handling XInputStream/XSeekable calls from soffice.

recv_handling_output(conn, sink \\ nil)

Receive frames while handling XOutputStream calls from soffice.

start_enum_reader(enumerable)

Spawn a demand-driven reader for enumerable.

try_handle_input(conn, payload)

Try to handle a stray request as an input stream call.

Types

connection()

@type connection() :: map()

input_source()

@type input_source() ::
  binary()
  | {:file, File.io_device(), non_neg_integer()}
  | {:enum, binary(), pid(), timeout()}

sink()

@type sink() :: nil | {:path, Path.t()} | (binary() -> any())

Functions

recv_handling_input(conn, data, stream_oid \\ nil)

@spec recv_handling_input(connection(), input_source() | source(), String.t() | nil) ::
  {binary(), connection()}

Receive frames while handling XInputStream/XSeekable calls from soffice.

source is either a binary (in-memory), {:file, pid, size} for file-backed streaming, or {:enum, buffer, reader} for enumerables.

When stream_oid is provided and the source is seekable (memory or file), queryInterface for XSeekable is supported — enabling ZIP-based formats like docx and xlsx.

Dispatches calls until we receive the reply to our pending request. Returns {reply_payload, updated_conn}.

recv_handling_output(conn, sink \\ nil)

@spec recv_handling_output(connection(), sink()) ::
  {binary(), binary() | :ok | {:error, String.t()}, connection()}

Receive frames while handling XOutputStream calls from soffice.

sink controls where output bytes go:

  • nil (default) — accumulate in memory, returns {reply, binary, conn}
  • {:path, path} — write chunks to file as they arrive, returns {reply, :ok, conn}
  • fun/1 — call with each chunk, returns {reply, :ok, conn}

start_enum_reader(enumerable)

@spec start_enum_reader(Enumerable.t()) :: pid()

Spawn a demand-driven reader for enumerable.

The enumerable advances only after a {:next, caller} request, preventing producer data from accumulating unboundedly in the caller's mailbox.

try_handle_input(conn, payload)

@spec try_handle_input(connection(), binary()) ::
  {:handled, connection()} | :not_input

Try to handle a stray request as an input stream call.

Called by Bridge.recv_reply/1 when it encounters a non-reply message outside the main stream recv loop. Returns :not_input if the request doesn't match the active input stream context.