Package org.acemq.rabbitmq.admin

Class RabbitAdmin

java.lang.Object
org.acemq.rabbitmq.admin.RabbitAdmin
All Implemented Interfaces:
AutoCloseable
public final class RabbitAdmin extends Object implements AutoCloseable
RabbitMQ's HTTP management API. 

 try (RabbitAdmin admin = RabbitAdmin.connect("http://localhost:15672", "guest", "guest")) {
     QueueInfo queue = admin.queue("orders.new").orElseThrow();
     System.out.println(queue.argument("x-message-ttl"));
 }

Port 15672, not 5672, and HTTP rather than AMQP. This is a different protocol to a different endpoint with different credentials, and it is available only where the rabbitmq_management plugin is enabled — which is why nothing in the message path may depend on it.

Every call has a timeout

Not configurable away, only adjustable. A management call is usually made while something else is already wrong — during a deployment, or while somebody is diagnosing an incident — and a request that hangs forever at that moment turns a slow broker into a stuck process.

Absent is not an error

A queue that does not exist comes back as Optional.empty() rather than an exception, because "is it there?" is the question most callers are asking. A 401 or a missing plugin is an exception, and says which of the two it was: those are configuration mistakes, and reporting them as "no queue" would send somebody looking in the wrong place entirely.

  • Field Details

    • DEFAULT_VHOST

      public static final String DEFAULT_VHOST
      The default virtual host, which is literally a forward slash.
      See Also:

  • Method Details

    • connect

      public static RabbitAdmin connect(String url, String user, String password)
      Parameters:
      url - the management base URL, for example http://localhost:15672
      user - a user with the monitoring or administrator tag
      password - its password
      Returns:
      a client for the default virtual host

    • connect

      public static RabbitAdmin connect(String url, String user, String password, Duration timeout)
      Parameters:
      url - the management base URL
      user - a user with the monitoring or administrator tag
      password - its password
      timeout - how long any one call may take
      Returns:
      a client for the default virtual host

    • forVhost

      public RabbitAdmin forVhost(String name)
      Parameters:
      name - a virtual host
      Returns:
      a client reading that virtual host instead of /, sharing this one's endpoint, credentials and timeout

    • queue

      public Optional<QueueInfo> queue(String name)
      Looks a queue up, with the arguments the broker actually holds.
      Parameters:
      name - the queue's name
      Returns:
      the queue, or empty when there is no such queue in this virtual host

    • queues

      public List<QueueInfo> queues()
      Returns:
      every queue in this virtual host

    • version

      public String version()
      Checks the broker is reachable and the credentials work.

      Worth calling at start-up rather than discovering at the first real call, because the three ways this fails — wrong password, plugin absent, wrong port — all produce the same confusion later.

      Returns:
      the broker's reported version

    • exchange

      public Optional<ExchangeInfo> exchange(String name)
      Parameters:
      name - the exchange's name; the empty string is the default exchange
      Returns:
      the exchange, or empty when there is no such exchange in this virtual host

    • exchanges

      public List<ExchangeInfo> exchanges()
      Returns:
      every exchange in this virtual host, including the broker's own

    • bindings

      public List<BindingInfo> bindings()
      Returns:
      every binding in this virtual host

    • bindingsForQueue

      public List<BindingInfo> bindingsForQueue(String queue)
      What is bound to a queue.

      The answer AMQP cannot give, and the one that identifies an unroutable message: a publish that nothing receives is a routing key with no binding, and until now the only way to find that out was to publish and be told.

      Parameters:
      queue - the queue
      Returns:
      its bindings, including the implicit default-exchange one

    • vhosts

      public List<VhostInfo> vhosts()
      Returns:
      every virtual host on the broker. Not scoped to this client's vhost

    • createVhost

      public void createVhost(String name)
      Creates a virtual host, or does nothing when it is already there.
      Parameters:
      name - what to call it

    • deleteVhost

      public void deleteVhost(String name)
      Deletes a virtual host and everything in it.

      Every queue, every message in them, every exchange, every binding and every permission, with no confirmation and nothing to undo it. There is no soft delete in this API and this method does not invent one.

      Parameters:
      name - the virtual host to remove

    • users

      public List<UserInfo> users()
      Returns:
      every user on the broker

    • createUser

      public void createUser(String name, String password, String... tags)
      Creates or updates a user.

      The password does not leave this process. It is hashed here, with RabbitMQ's own salted SHA-256, and the broker is sent password_hash and hashing_algorithm. Up to 0.1.x the password itself was sent and the broker hashed it, which put it in a request body over whatever transport the base URL named — often plain HTTP. HTTPS is still right for this API: the hash is not the password, and is still enough to stand up a broker the password works against.

      Parameters:
      name - the user
      password - its password
      tags - administrator, monitoring, management, policymaker, or none for a user that can connect and nothing else

    • createUser

      public void createUser(String name, PasswordHash hash, String... tags)
      Creates or updates a user from a hash computed elsewhere. 
      
 admin.createUser("orders-service", PasswordHash.of(hashFromVault, PasswordHash.SHA256));
      Parameters:
      name - the user
      hash - its password hash
      tags - as for createUser(String, String, String...)

    • deleteUser

      public void deleteUser(String name)
      Parameters:
      name - the user to remove. Connections it already has are closed by the broker

    • deleteUsers

      public void deleteUsers(String... names)
      Deletes several users in one request.

      A name that is not a user is skipped rather than refused, so this is safe to repeat. POST /api/users/bulk-delete.

      Parameters:
      names - the users to remove

    • usersWithoutPermissions

      public List<UserInfo> usersWithoutPermissions()
      Users that cannot reach any virtual host.

      The broker's own answer to "which accounts are left over?". A user here can still log in to the management API if it has a tag, so it is not necessarily harmless — but it is not used by any application, because no application can connect as it.

      Returns:
      every user with no permissions in any virtual host

    • permissions

      public List<PermissionInfo> permissions()
      Returns:
      every permission on the broker, across all virtual hosts

    • grant

      public void grant(String user, String configure, String write, String read)
      Grants a user access to this client's virtual host.

      Three regular expressions, matched against resource names. ".*" for all three is unrestricted access — what most tutorials show, and rarely what anybody means.

      Parameters:
      user - the user
      configure - what it may declare and delete
      write - what it may publish to
      read - what it may consume from

    • revoke

      public void revoke(String user)
      Parameters:
      user - the user losing access to this client's virtual host

    • policies

      public List<PolicyInfo> policies()
      Returns:
      the policies in this virtual host

    • putPolicy

      public void putPolicy(String name, String pattern, Map<String,Object> definition, int priority)
      Creates or replaces a policy.

      Unlike a queue argument, this can be changed later — which is the reason to prefer a policy for anything operational. Note that where several policies match, the highest priority wins outright; the definitions are not merged.

      Parameters:
      name - what to call it
      pattern - a regular expression matched against names
      definition - what to set
      priority - higher wins

    • deletePolicy

      public void deletePolicy(String name)
      Parameters:
      name - the policy to remove. What it was applying reverts immediately

    • shovels

      public List<ShovelInfo> shovels()
      Returns:
      the shovels the broker is running, or an empty list when the shovel plugin is not enabled — which is a configuration fact rather than a failure

    • federationUpstreams

      public List<FederationUpstream> federationUpstreams()
      Returns:
      the upstreams defined in this virtual host, or an empty list when the federation plugin is not enabled

    • putFederationUpstream

      public void putFederationUpstream(String name, String uri, Map<String,Object> settings)
      Declares an upstream, or replaces one of the same name.

      This alone federates nothing. An upstream names a broker; a policy whose definition carries federation-upstream is what links anything to it. That two-step is the usual reason federation appears not to work, and nothing reports it because nothing is wrong. 

      
 admin.putFederationUpstream("other-dc", "amqp://user:pass@other-dc:5672",
         Map.of("expires", 3_600_000));
 admin.putPolicy("federate", "^events\\.", Map.of("federation-upstream", "other-dc"), 1);

      Set expires unless there is a reason not to. Without it the queue federation creates on the upstream survives forever with nothing consuming from it, and a decommissioned downstream leaves a queue growing on somebody else's broker.

      Parameters:
      name - what to call it, and what a policy will refer to
      uri - where the upstream broker is, credentials included
      settings - expires, message-ttl, ack-mode, max-hops, prefetch-count; may be empty

    • deleteFederationUpstream

      public void deleteFederationUpstream(String name)
      Parameters:
      name - the upstream to remove. Any link using it stops; a policy still naming it stays and quietly matches nothing

    • federationLinks

      public List<FederationLinkInfo> federationLinks()
      The links that actually exist.

      The only proof federation is working. An upstream can be declared and a policy can match and there can still be no link, because the upstream is unreachable or its credentials are wrong — and none of that surfaces anywhere else.

      Returns:
      the links in this virtual host, or empty when the federation management plugin is not enabled

    • declareShovel

      public void declareShovel(String name, Map<String,Object> definition)
      Declares a dynamic shovel, or replaces one of the same name.

      A shovel consumes. A message it moves is gone from the source, which is what makes it right for draining a queue during a migration and wrong for copying a stream of events — federation does that. Pointing one at a queue somebody is still reading produces two consumers racing, and the shovel usually wins. 

      
 admin.declareShovel("drain-orders", Map.of(
         "src-protocol", "amqp091", "src-uri", "amqp://old-broker", "src-queue", "orders",
         "dest-protocol", "amqp091", "dest-uri", "amqp://new-broker", "dest-queue", "orders"));
      Parameters:
      name - what to call it
      definition - the shovel's source and destination

    • deleteShovel

      public void deleteShovel(String name)
      Parameters:
      name - the shovel to remove. It stops moving messages immediately

    • parameters

      public List<ParameterInfo> parameters(String component)
      Runtime parameters of one component.

      Federation upstreams and dynamic shovels are both stored this way, which is why they are declared and deleted identically and why neither appears in a topology export that only looks at queues and exchanges.

      Parameters:
      component - federation-upstream, shovel, or another plugin's
      Returns:
      the parameters, or empty when the plugin providing that component is absent

    • health

      public Health health()
      The broker's own health checks. 
      
 if (!admin.health().isHealthy()) {
     admin.health().failures().forEach(f -> log.error("{}", f));
 }
      Returns:
      a view over /api/health/checks/

    • exportDefinitions

      public Definitions exportDefinitions()
      Exports everything the broker is configured to be.

      Queues, exchanges, bindings, vhosts, users, permissions, topic permissions, policies and runtime parameters, in one request. This is the backup: assembling one by walking the topology misses federation upstreams and shovels, which are parameters, and the restored broker looks complete while federating nothing.

      No messages. This is configuration only.

      Returns:
      the definitions, with the broker's own JSON kept verbatim

    • exportVhostDefinitions

      public Definitions exportVhostDefinitions()
      Returns:
      the definitions for the current virtual host alone, which omits the cluster-wide sections — users, permissions and global parameters are not scoped to a vhost

    • importDefinitions

      public void importDefinitions(String json)
      Applies a definitions document to this broker.

      This is a merge, not a replacement. Everything in the document is created or updated; nothing absent from it is removed. Importing an old backup therefore restores what was lost without deleting what has been added since — which is usually what is wanted, and is not what "restore" sounds like.

      Users in a definitions file carry password hashes rather than passwords, so importing one recreates accounts that accept the original passwords. Treat the file as a credential.

      Parameters:
      json - a definitions document, as produced by exportDefinitions()

    • connections

      public List<ConnectionInfo> connections()
      Returns:
      every client connection to the broker, across all virtual hosts

    • connection

      public Optional<ConnectionInfo> connection(String name)
      Parameters:
      name - the broker's name for the connection, from ConnectionInfo.name()
      Returns:
      it, or empty if it has since closed

    • closeConnection

      public void closeConnection(String name, String reason)
      Closes a client connection.

      The client sees a connection closed by the broker, with the reason given. A well-built client will reconnect immediately, which makes this a way to force a reconnection — to move a client off a node before maintenance, for instance — rather than a way to keep something disconnected.

      Parameters:
      name - from ConnectionInfo.name()
      reason - told to the client, and worth making specific

    • channels

      public List<ChannelInfo> channels()
      Returns:
      every open channel, across all connections

    • consumers

      public List<ConsumerInfo> consumers()
      Returns:
      every consumer in the current virtual host, with the queue each is attached to

    • declareQueue

      public void declareQueue(String name, boolean durable, Map<String,Object> arguments)
      Declares a queue.

      Note that AMQP can do this, and usually should: a queue an application depends on is best declared by that application, at startup, so it exists before the first publish. This exists for the cases AMQP cannot serve — provisioning a queue for a team that has not deployed yet, or restoring one from a backup.

      Parameters:
      name - the queue
      durable - whether it survives a broker restart
      arguments - queue arguments, or an empty map

    • deleteQueue

      public void deleteQueue(String name)
      Parameters:
      name - the queue to delete, with everything in it

    • purgeQueue

      public void purgeQueue(String name)
      Discards every message in a queue, keeping the queue.

      Irreversible, and there is no confirmation. It is genuinely the right tool for a dead-letter queue full of messages that have been analysed and cannot be replayed — and it is exactly as fast on a queue holding something important.

      Parameters:
      name - the queue to empty

    • declareExchange

      public void declareExchange(String name, String type, boolean durable, Map<String,Object> arguments)
      Parameters:
      name - the exchange
      type - direct, topic, fanout or headers
      durable - whether it survives a broker restart
      arguments - exchange arguments, or an empty map

    • deleteExchange

      public void deleteExchange(String name)
      Parameters:
      name - the exchange to delete. Its bindings go with it

    • bindQueue

      public void bindQueue(String source, String queue, String routingKey, Map<String,Object> arguments)
      Binds a queue to an exchange.
      Parameters:
      source - the exchange
      queue - the queue
      routingKey - the routing key or pattern
      arguments - binding arguments, which is where a headers exchange keeps its rules

    • bindExchange

      public void bindExchange(String source, String destination, String routingKey, Map<String,Object> arguments)
      Binds an exchange to another exchange.
      Parameters:
      source - the upstream exchange
      destination - the downstream exchange
      routingKey - the routing key or pattern
      arguments - binding arguments

    • unbind

      public void unbind(BindingInfo binding)
      Removes a binding.

      Takes the whole BindingInfo rather than a routing key because a binding is identified by its properties key, which arrives already percent-encoded and must not be encoded again.

      Parameters:
      binding - one returned by bindings() or bindingsForQueue(String)

    • operatorPolicies

      public List<PolicyInfo> operatorPolicies()
      Policies an operator sets and a user cannot override.

      Same shape as policies() and a different purpose: where both apply, the operator policy wins. These are the guard rails — a maximum queue length across a shared broker, say — that survive a team applying its own policy.

      Returns:
      every operator policy in the current virtual host

    • putOperatorPolicy

      public void putOperatorPolicy(String name, String pattern, Map<String,Object> definition, int priority)
      Parameters:
      name - the policy name
      pattern - a regular expression matched against queue or exchange names
      definition - the settings to apply
      priority - higher wins among matching operator policies

    • deleteOperatorPolicy

      public void deleteOperatorPolicy(String name)
      Parameters:
      name - the operator policy to remove

    • vhostLimits

      public List<LimitInfo> vhostLimits()
      Returns:
      the limits set on every virtual host

    • setVhostLimit

      public void setVhostLimit(String limit, long value)
      Parameters:
      limit - LimitInfo.MAX_CONNECTIONS or LimitInfo.MAX_QUEUES
      value - the ceiling

    • clearVhostLimit

      public void clearVhostLimit(String limit)
      Parameters:
      limit - the limit to remove, restoring "no limit"

    • userLimits

      public List<LimitInfo> userLimits()
      Returns:
      the limits set on every user

    • setUserLimit

      public void setUserLimit(String user, String limit, long value)
      Parameters:
      user - the user
      limit - max-connections or max-channels
      value - the ceiling

    • clearUserLimit

      public void clearUserLimit(String user, String limit)
      Parameters:
      user - the user
      limit - the limit to remove

    • topicPermissions

      public List<TopicPermissionInfo> topicPermissions()
      Routing-key authorisation on topic exchanges.

      Separate from permissions(), and easy to miss: a user with no topic permissions is unrestricted on topic exchanges. An empty list here means nothing is being enforced, however carefully the ordinary permissions are set.

      Returns:
      every topic permission on the broker

    • grantTopic

      public void grantTopic(String user, String exchange, String write, String read)
      Parameters:
      user - the user
      exchange - the topic exchange
      write - a routing-key pattern this user may publish with
      read - a routing-key pattern this user may bind and consume with

    • revokeTopic

      public void revokeTopic(String user)
      Parameters:
      user - the user whose topic permissions in this virtual host are removed, which restores unrestricted access rather than removing it

    • whoami

      public CurrentUser whoami()
      Returns:
      the user these credentials belong to, and its tags. Worth calling at startup: a missing tag reported here is far easier to act on than the 403 it becomes later

    • nodes

      public List<NodeInfo> nodes()
      Returns:
      every node in the cluster

    • clusterName

      public String clusterName()
      Returns:
      the cluster's name

    • featureFlags

      public List<FeatureFlagInfo> featureFlags()
      Returns:
      every feature flag and its state. A flag whose stability is required and whose state is not enabled will block the next major upgrade

    • deprecatedFeaturesInUse

      public List<DeprecatedFeatureInfo> deprecatedFeaturesInUse()
      Returns:
      the deprecated features this broker is actually using. The upgrade readiness list: everything on it stops working at some future version, and an empty list is the answer you want before upgrading

    • globalParameters

      public List<GlobalParameterInfo> globalParameters()
      Returns:
      the cluster-wide parameters, which are not scoped to a virtual host. These use GlobalParameterInfo rather than ParameterInfo because their values are not objects: internal_cluster_id is a string and cluster_tags is an array, and every broker has both

    • rebalanceQueues

      public void rebalanceQueues()
      Moves queue leaders so each node leads a fair share.

      After a rolling restart every quorum queue and stream tends to be led from whichever node came back first, and that node does most of the work. This asks the broker to spread the leaders out again. POST /api/rebalance/queues.

      Asynchronous. The call returns once the broker has accepted the request, not when the leaders have moved; the outcome is reported only in the broker's log. Each leader that moves makes its queue briefly unavailable, so this is a maintenance operation rather than something to run on a timer.

    • extensions

      public List<String> extensions()
      Returns:
      the JavaScript files that management plugins have registered with the UI — the plugin list as the management API sees it, so stream.js here means rabbitmq_stream_management is enabled. GET /api/extensions

    • authAttempts

      public List<AuthAttemptInfo> authAttempts(String node)
      Login attempts one node has counted, per protocol.

      The only record of a failed login: it never becomes a connection, so nothing else lists it. Counted per node, which is why a node is named — NodeInfo.name().

      Parameters:
      node - a node name, such as rabbit@host
      Returns:
      the counters, or empty when there is no such node

    • authAttemptsBySource

      public List<AuthAttemptInfo> authAttemptsBySource(String node)
      Login attempts one node has counted, per remote address and username.

      Empty unless the broker is told to track them. The setting is track_auth_attempt_source in the rabbit application, off by default, and has no rabbitmq.conf key on 3.13 or 4.x: it goes in advanced.config as [{rabbit, [{track_auth_attempt_source, true}]}]. Off because every distinct address and username becomes an entry, and a scanner trying many usernames would grow the table without limit.

      Parameters:
      node - a node name
      Returns:
      the counters, or empty when source tracking is off or there is no such node

    • resetAuthAttempts

      public void resetAuthAttempts(String node)
      Zeroes one node's login counters, the per-protocol totals and the per-source entries alike — after a misconfigured client is fixed, say, so a failure that follows is a new one.
      Parameters:
      node - a node name

    • streamConnections

      public List<StreamConnectionInfo> streamConnections()
      Connections over the stream protocol, across all virtual hosts.

      These need the rabbitmq_stream_management plugin, and without it every stream method here answers an empty list — the same rule as shovels(): a plugin that is not enabled is a fact about the broker, not a failure. extensions() containing stream.js is how to tell the two kinds of empty apart.

      The broker reports stream connections, publishers and consumers through its statistics, so one that appeared within the last statistics interval — five seconds by default — is not listed yet.

      Returns:
      every stream protocol connection

    • streamPublishers

      public List<StreamPublisherInfo> streamPublishers()
      Returns:
      the stream publishers in this virtual host

    • streamPublishers

      public List<StreamPublisherInfo> streamPublishers(String stream)
      Parameters:
      stream - a stream in this virtual host
      Returns:
      the publishers writing to it

    • streamConsumers

      public List<StreamConsumerInfo> streamConsumers()
      There is no per-stream form of this, unlike streamPublishers(String): /api/stream/consumers/{vhost}/{stream} answers 400 on 3.13 and 4.x alike. Filter on StreamConsumerInfo.stream() instead.
      Returns:
      the stream consumers in this virtual host

    • vhost

      public String vhost()
      Returns:
      the virtual host this client reads

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object