@spec add_match(pid(), Rebus.MatchRule.t(), non_neg_integer()) ::
  {:ok, reference()} | {:error, match_error_reason()}

Subscribes the calling process to signals selected by a match rule.

Signals matching the rule arrive as {ref, %Rebus.Message{}}. Build the rule with Rebus.MatchRule.new/1; raw strings are not accepted. Equivalent rules share one bus registration, and each call gets its own reference.

{:error, :timeout} is ambiguous: you get no reference, but the bus may already hold the rule. A reply whose header fields exceed a local decoding limit has the same result because its reply serial cannot be trusted; see call/3 for the fuller contract. {:error, :sender_routing_ambiguous} means the rule overlaps an existing one with a different sender. {:error, :not_a_bus} means the connection was opened with bus: false, and nothing was sent.

Rebus closes the connection when too many ambiguous cleanups accumulate, or when one remains unresolved after its retry budget. It also closes the connection after {:error, :match_subscription_state_lost}, because closing makes the bus discard the connection's rules.

Return values

Success is {:ok, reference}. Every failure is {:error, reason}: :timeout, :not_a_bus, :sender_routing_ambiguous, :match_rule_cleanup_pending, :match_subscription_state_lost, {:bus_error, error_name}, :invalid_bus_reply, :not_connected, :disconnected, :encode_failed, :serial_exhausted, :fd_claim_expired, :not_started, {:reply_dropped, outcome} and :remote_connection_unsupported. The match rules guide lists what each one means.

Example

rule = Rebus.MatchRule.new!(interface: "org.example.Status", member: "Changed")
{:ok, ref} = Rebus.add_match(conn, rule)
receive do
  {^ref, %Rebus.Message{type: :signal}} -> :ok
end

@spec add_signal_handler(pid()) ::
  {:ok, reference()} | {:error, :not_connected | :timeout | :disconnected}

Registers the calling process to receive every signal on the connection.

This asks the bus for nothing; use add_match/3 to have the bus route more signals here. A connection can carry several handlers, each receiving every signal that arrives on it. Rebus removes a handler when its process exits or the connection closes. Call delete_signal_handler/2 as soon as you stop wanting signals, so a busy bus does not fill the process mailbox.

Return values

• {:ok, ref} - the handler is registered.
• {:error, :not_connected} - connection setup has not completed.
• {:error, :timeout} - the connection did not service the request in time.
• {:error, :disconnected} - the connection has stopped.

Examples

{:ok, ref} = Rebus.add_signal_handler(conn)

Each signal then arrives in the calling process as:

{^ref, %Rebus.Message{
  type: :signal,
  header_fields: %{path: path, interface: interface, member: member, sender: sender},
  body: [signal_args]
}}

@spec call(pid(), Rebus.Message.t(), non_neg_integer()) ::
  {:ok, Rebus.Message.t()} | {:error, call_error()}

Sends a method call and waits for its correlated reply.

Only a method call that expects a reply is accepted, and timeout is in milliseconds. Each result carries the complete message, including any received descriptors in :unix_fds. A D-Bus error reply from the peer is returned as {:error, %Rebus.Message{type: :error}}.

Return values

• {:ok, %Rebus.Message{type: :method_return}} - the peer replied successfully.
• {:error, %Rebus.Message{type: :error}} - the peer returned a D-Bus error reply.
• {:error, :timeout} - no reply arrived in time, and the request may already have reached the peer. A request sent to a named PID whose connect/2 has not yet returned was never written and is safe to retry. A reply that exhausted a local decoding cap inside its own header fields also lands here: Rebus recovered no trusted reply_serial, so the frame is dropped and the connection stays up, but the call cannot be matched to it.
• {:error, {:reply_dropped, :method_return}} or {:error, {:reply_dropped, {:error, error_name}}} - the peer definitely replied, but the reply exhausted a local decoding cap and was discarded. Neither is delivery-ambiguous, so decide whether to retry from what the operation does. A cap exhausted inside the reply's own header fields cannot produce these shapes; it returns {:error, :timeout} instead.
• {:error, :fd_claim_expired} - Rebus closed the reply's descriptors instead of handing them over.
• {:error, :disconnected} - the connection stopped before the reply, or before its descriptors transferred. A call carrying descriptors can return after timeout; see Unix file descriptor passing.
• Nothing was written for :encode_failed, :no_reply_expected, :not_connected, :serial_exhausted, :unix_fd_unsupported, :unix_fd_not_negotiated, :unix_fd_send_failed, {:invalid_message_type, type} and :remote_connection_unsupported.

Examples

message = Rebus.Message.new!(:method_call,
  path: "/org/freedesktop/DBus",
  interface: "org.freedesktop.DBus",
  destination: "org.freedesktop.DBus",
  member: "ListNames"
)

{:ok, %Rebus.Message{type: :method_return, body: [names]}} = Rebus.call(conn, message)
{:error, :timeout} = Rebus.call(conn, message, 1_000)

@spec close(pid()) :: :ok | {:error, :not_found | :remote_connection_unsupported}

Stops a local connection process created by connect/2.

Connections are supervised and, unless connect/2 was given an :owner, remain alive after the connecting process exits. Use this function to release a named or otherwise no-longer-needed connection. It accepts only local connection PIDs; remote PIDs are not supported.

An owned connection is stopped the same way. Once an owner's exit has stopped its connection there is nothing left to close, and this function answers {:error, :not_found} as it does for any other PID that is no longer a connection.

Return values

• :ok - The supervised connection was stopped.
• {:error, :not_found} - The PID is not a current Rebus connection.
• {:error, :remote_connection_unsupported} - The PID belongs to another node.

@spec connect(
  address(),
  keyword()
) :: {:ok, pid()} | {:error, term()}

Connects to a D-Bus endpoint and returns its connection process.

The call blocks until the connection is usable: authenticated and, on a message bus, holding its unique name. The returned PID is supervised and outlives the process that connected it, so release it with close/1, or bind it to a process with :owner.

Addresses

• :system - the system bus.
• :session - the session bus.
• %{family: :local, path: "/tmp/my-dbus"} - a Unix domain socket.
• %{family: :inet, addr: {127, 0, 0, 1}, port: 12345} - a TCP endpoint, with :inet6 and an eight-element address for IPv6.

:system reads DBUS_SYSTEM_BUS_ADDRESS, then the :system_bus_address config key, then the default unix:path=/run/dbus/system_bus_socket. :session reads DBUS_SESSION_BUS_ADDRESS, then falls back to unix:path=$XDG_RUNTIME_DIR/bus when XDG_RUNTIME_DIR is set. An empty variable counts as unset. Both hold a D-Bus address list whose supported entries are tried in the order they are listed, and a guid= on the entry that answers must match the server's identity. See Authentication for the rest.

Options

Notes

• A PID found by name before its connect/2 returns is still being established. Operations sent to it may time out, and are safe to retry once connect/2 succeeds.
• An :owner is monitored, not linked, and need not be the process calling connect/2. The connection stops with {:shutdown, :owner_down} when the owner exits for any reason, including :normal. The stop is an ordinary one, so the connection's socket and any retained descriptors are closed and the bus releases the state that connection held. Inflight callers receive {:error, :disconnected}. Only a PID on the local node is accepted; a remote PID is :invalid_owner.
• The supervisor owns the connection until close/1 or, with :owner, until that process exits.
• A write timeout that accepted no bytes fails only that caller. After a partial frame the connection terminates and inflight callers receive {:error, :disconnected}.
• For an address list, setup shares one budget across every candidate, so a slow candidate leaves less time for the entries after it.
• Failed address-list attempts are logged at debug level, without any address, host, path or GUID.

Return values

Success is {:ok, pid}. Every failure is {:error, reason}:

• Invalid option: :invalid_timeout, :invalid_read_timeout, :invalid_write_timeout, :invalid_allow_anonymous, :invalid_bus_option, :invalid_owner, :invalid_name.
• Owner that exited before the connection was established: :owner_down.
• Unusable address: {:invalid_bus_address, reason}, :unsupported_bus_transport, {:tcp_resolution_failed, reason}, :no_system_bus_address, :no_session_bus_address.
• Refused authentication: :auth_id_unavailable, :auth_cookie_unavailable, :auth_failed, {:auth_rejected, mechanisms}, :guid_mismatch. See Authentication.
• Expired setup budget: :read_timeout, or {:read_timeout, reason} once an address-list attempt has already failed.
• Refused or unusable Hello reply: {:hello_failed, reason}.
• Taken :name: {:name_taken, pid} for another Rebus connection, or {:name_registered, pid} for any other process.
• Any other socket or setup failure: the failure's own atom.

Examples

# A custom Unix socket
{:ok, conn} = Rebus.connect(%{family: :local, path: "/tmp/my-dbus"})

# A TCP endpoint
address = %{family: :inet, addr: {127, 0, 0, 1}, port: 12345}
{:ok, conn} = Rebus.connect(address)

# Release a named connection when its lifecycle is complete
{:ok, conn} = Rebus.connect(address, name: :local_bus)
:ok = Rebus.close(conn)

# Let the connection end with the process whose bus-side state it holds
{:ok, conn} = Rebus.connect(address, owner: self())

@spec connect!(
  address(),
  keyword()
) :: pid()

Same as connect/2, but raises ArgumentError on failure.

The exception message carries the reason connect/2 would have returned.

@spec delete_signal_handler(pid(), reference()) ::
  :ok | {:error, :not_connected | :timeout | :disconnected}

Stops a signal handler registered by add_signal_handler/1.

The handler receives no further signals; others on the same connection carry on. Deleting a reference is idempotent while the connection is available.

Return values

• :ok - the handler is gone.
• {:error, :not_connected} - connection setup has not completed.
• {:error, :timeout} - the connection did not service the request in time.
• {:error, :disconnected} - the connection has stopped.

Examples

{:ok, ref} = Rebus.add_signal_handler(conn)
:ok = Rebus.delete_signal_handler(conn, ref)

@spec remove_match(pid(), reference(), non_neg_integer()) ::
  :ok | {:error, match_error_reason()}

Removes a match-rule subscription reference.

Removing a reference is idempotent and scoped to its own connection. A removed reference receives no further signals, and the last reference for a rule also removes the rule from the bus. A removal that times out or fails keeps the reference for a retry, while Rebus clears the rule in the background. Rebus removes the reference and the bus rule when the owning process exits. Closing the connection discards both.

Return values

Success is :ok. Failures are the {:error, reason} shapes listed for add_match/3, apart from :not_a_bus and :sender_routing_ambiguous.

@spec send(pid(), Rebus.Message.t()) :: :ok | {:error, error_reason()}

Sends a message without waiting for a reply.

Use this for signals and for method calls flagged :no_reply_expected. send/2 allows five seconds for the connection to accept the message; send/3 takes that timeout as an argument.

Return values

• :ok - the frame was handed to the socket.
• {:error, :timeout} - the message may already have reached the peer. A message sent to a named PID whose connect/2 has not yet returned was never written and is safe to retry.
• {:error, :disconnected} - the connection stopped.
• Nothing was written for :encode_failed, :reply_expected, {:invalid_message_type, type}, :not_connected, :serial_exhausted, :unix_fd_unsupported, :unix_fd_not_negotiated, :unix_fd_send_failed and :remote_connection_unsupported.

@spec send(pid(), Rebus.Message.t(), non_neg_integer()) ::
  :ok | {:error, error_reason()}

Sends a message with a custom dispatch timeout in milliseconds.

Accepts the same messages as send/2 and returns the same values. The timeout bounds how long the connection has to accept the message. {:error, :timeout} is delivery-ambiguous.