Package org.acemq.amqp.test

Class InMemoryTransport

java.lang.Object
org.acemq.amqp.test.InMemoryTransport
All Implemented Interfaces:
Transport
public final class InMemoryTransport extends Object implements Transport
A transport that runs entirely in the process, for tests that should not need Docker.

Connect with a memory:// URL, where the host names the broker instance: 


 try (AceMq mq = AceMq.connect("memory://orders")) {
     mq.declareExchange("orders", "topic");
     mq.declareQueue("orders.new");
     mq.bind("orders.new", "orders", "order.*");
 }

Two connections to the same name share state; a different name is a different broker. A test wanting isolation should use a unique name rather than resetting shared state.

This is a fake, not a simulator, and it is honest about the difference. It implements routing, prefetch and settlement, because those are what most tests actually exercise. It does not implement replication, persistence, delayed delivery or dead-lettering, and does not claim those capabilities, so code that depends on them fails here for the same reason it would fail against a broker that lacks them. The conformance suite will run against both this and real brokers, which is what keeps the two from drifting apart.

  • Constructor Details

    • InMemoryTransport

      public InMemoryTransport()

  • Method Details

    • reset

      public static void reset()
      Discards every in-memory broker and its contents.

    • block

      public static void block(String brokerName, String reason)
      Makes a named broker refuse publishes, the way a real one does under a memory or disk alarm.

      Exists because this failure is worth testing and is otherwise almost impossible to reproduce: publishing to a blocked broker does not fail, it simply never returns, and a service that has not been tested against it will hang in production instead of here. 

      
 InMemoryTransport.block("orders", "low on memory");
 assertThrows(ConnectionBlockedException.class, () -> publisher.send(order));
 InMemoryTransport.unblock("orders");

      Publishers wait up to ConnectionConfig.blockedTimeout() before throwing, which is zero by default, so a publish while blocked throws at once. A test that wants a publish to wait for unblock sets a timeout.

      One deliberate difference from RabbitMQ, worth knowing before writing a test around it: here isBlocked() becomes true as soon as this is called, whereas RabbitMQ tells a connection it is blocked only when that connection next publishes. Against a real broker the first message into an alarm is already on the wire before anything is known, so it may or may not arrive; here nothing is ever written, and ConnectionBlockedException.mayHaveBeenPublished() is always false. A test that needs the uncertain case needs a real broker.

      Parameters:
      brokerName - the host part of the memory:// URL
      reason - what to report as the broker's own explanation, such as low on memory

    • unblock

      public static void unblock(String brokerName)
      Makes a named broker accept publishes again, releasing every publisher waiting on it.
      Parameters:
      brokerName - the host part of the memory:// URL

    • scheme

      public String scheme()
      Specified by:
      scheme in interface Transport
      Returns:
      the URL scheme this transport handles, for example amqp or amqp10

    • name

      public String name()
      Specified by:
      name in interface Transport
      Returns:
      a short name for logs and diagnostics, such as rabbitmq

    • capabilities

      public Set<Capability> capabilities()
      Specified by:
      capabilities in interface Transport
      Returns:
      what brokers reachable through this transport can do

    • connect

      public TransportConnection connect(ConnectionConfig config)
      Description copied from interface: Transport
      Opens a connection.
      Specified by:
      connect in interface Transport
      Parameters:
      config - connection settings; ConnectionConfig.scheme() must match Transport.scheme()
      Returns:
      an open connection