@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. Rebus cleans up in the background. When it cannot establish what the bus holds for a rule, it leaves the rule uncertain, keeps the connection up, and sends a match_uncertain_notice/0 to each holder of a reference on the rule and to the connection's :owner. The notice is a four-element tuple tagged :rebus_match_uncertain, so it never matches a {ref, %Rebus.Message{}} pattern. Closing the connection is the only way to make the bus forget such a rule. See Signal subscriptions and match rules for what else an uncertain rule changes.

Return values

Success is {:ok, reference}. Every failure is {:error, reason}, where reason is one of :timeout, :not_a_bus, :sender_routing_ambiguous, :match_rule_cleanup_pending, :match_rule_uncertain, :match_recovery_capacity_exhausted, {:bus_error, error_name}, :invalid_bus_reply, :not_connected, :disconnected, :encode_failed, :serial_exhausted, :not_started, {:reply_dropped, outcome} or :remote_connection_unsupported. The 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.

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 reply.

Only a method call that expects a reply is accepted, and timeout is in milliseconds. The call may carry descriptors, as Rebus.UnixFD handles, raw descriptor numbers or both; see Rebus.Message.new/2. Each result carries the complete message, including a Rebus.UnixFD handle in :unix_fds for each descriptor received with it, which the calling process owns. A D-Bus error reply from the peer is returned as {:error, {:dbus_error, %Rebus.Message{type: :error}}}, so it never shares a shape with a local or transport failure.

Return values

• {:ok, %Rebus.Message{type: :method_return}} - the peer replied.
• {:error, {:dbus_error, %Rebus.Message{type: :error}}} - the peer returned a D-Bus error. Its name is in header_fields.error_name, and handles for any descriptors it carried are in :unix_fds, owned by the caller.
• {:error, :timeout} - no reply arrived in time. The request may already have reached the peer, except when it was sent to a named PID whose connect/2 has not yet returned: that request was never written and is safe to retry. A reply that arrives later is dropped, and any descriptors it carried are closed when the notice of the timeout, which the calling process sends the connection as it gives up, reaches it.
• {:error, {:reply_dropped, :method_return}} or {:error, {:reply_dropped, {:error, error_name}}} - the peer replied, but the reply exceeded a local decoding cap and was discarded. Delivery is not in doubt, so decide whether to retry from what the operation does.
• {:error, :disconnected} - the connection stopped before the reply.
• Nothing was written for :encode_failed, :no_reply_expected, :not_connected, :serial_exhausted, :unix_fd_unsupported, :unix_fd_not_negotiated, :unix_fd_closed, :unix_fd_send_failed, {:invalid_message_type, type} and :remote_connection_unsupported. :unix_fd_closed means a handle in unix_fds was already closed, or belongs to another node, when the connection accepted the call.

A reply that exceeds a decoding cap inside its own header fields is a special case: Rebus recovers no trusted reply_serial, so the frame is dropped, the connection stays up, and the call reports {:error, :timeout} rather than :reply_dropped, because it cannot be matched to the reply.

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)

case Rebus.call(conn, message) do
  {:ok, reply} -> {:ok, reply.body}
  {:error, {:dbus_error, error}} -> {:error, error.header_fields.error_name}
  {:error, reason} -> {:error, reason}
end

@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, stay alive after the connecting process exits, so this is how a connection is released. An owned connection can be closed the same way. Once a connection has already stopped, for example because its owner exited, there is nothing left to close and this returns {:error, :not_found}.

A connection can be closed before its connect/2 returns, for example through its registered :name. One still connecting its socket or authenticating stops at once, and that connect/2 returns {:error, :shutdown}, or {:error, :noproc} when the close lands before connect/2 has started watching the connection. Either way it tries no further bus addresses.

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

Lifetime

The supervisor owns the connection until close/1 is called or, with :owner, until that process exits. An :owner is monitored, not linked, and need not be the process calling connect/2. When it exits for any reason, including :normal, the connection stops with {:shutdown, :owner_down}: its socket and any descriptors it still owns are closed, the bus releases the state it held, and inflight callers receive {:error, :disconnected}. Only a local PID is accepted; a remote one is :invalid_owner.

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.

Timeouts

Setup is one budget, not one per step. A slow identity lookup or socket connect leaves less time for authentication, and for an address list the budget is shared 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.

A write timeout that accepted no bytes fails only that caller. After a partial frame the connection terminates and inflight callers receive {:error, :disconnected}.

Message size

:max_message_size bounds the memory a peer can make one connection hold for inbound traffic. Rebus buffers one partial message at a time and reads at most 64 KiB at once, so a connection retains about the cap plus one read. The default is the protocol limit, which lets a peer make a connection buffer 128 MiB, so lower it for untrusted peer-to-peer or TCP endpoints. A message declared larger than the cap stops the connection with {:shutdown, :message_too_large} once its fixed header arrives, before the rest is buffered, and any descriptors received with it are closed. The cap also applies to the Hello reply; the 4096 floor leaves room for that reply and the NameAcquired signal that follows it.

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, :invalid_max_message_size, :invalid_max_pending_calls.
• 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}.
• Descriptors sent by a peer that was never offered descriptor passing, before the connection was established, on macOS: :unix_fd_not_negotiated. See Unix file descriptor passing.
• 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. The reference receives no further signals, and removing the last reference for a rule also removes the rule from the bus, unless Rebus has left the rule uncertain. A removal that times out or fails keeps the reference for a retry while Rebus clears the rule in the background. The reference and the bus rule are also removed when the owning process exits, and 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, :sender_routing_ambiguous and :match_recovery_capacity_exhausted.

@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. Descriptors are sent as for call/3.

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_closed, :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.

@spec unique_name(pid()) ::
  {:ok, binary() | nil} | {:error, :not_connected | :timeout | :disconnected}

Returns the unique name the message bus assigned to the connection.

A connection to a message bus is given its unique name, such as ":1.42", by the Hello call it makes during connect/2, and keeps it for its whole life. Peers address method calls and replies to that name, and the bus puts it in the sender field of everything the connection sends. A connection created with bus: false sends no Hello and has no unique name, so this returns {:ok, nil} for it.

Return values

• {:ok, name} - the unique name, a binary starting with ":".
• {:ok, nil} - the connection was created with bus: false.
• {: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, conn} = Rebus.connect(:session)
{:ok, ":" <> _} = Rebus.unique_name(conn)