Package org.acemq.workloads.scenario

Class ProducerNode

java.lang.Object
org.acemq.workloads.scenario.ProducerNode
All Implemented Interfaces:
Node
public final class ProducerNode extends Object implements Node
One source of load, aimed at one exchange.

Several of them is the point. "Two services publishing to the same exchange, one of them ten times louder" is a shape a single publisher block cannot describe, and it is the shape that produces the failures people are actually trying to reproduce.

Say the rate, not the threads. rate(long) is the number that means something — messages a second, offered on a schedule. How many threads it takes to offer that is arithmetic, and threads(int) is there for when the arithmetic is wrong.

  • Method Details

    • to

      public ProducerNode to(String exchange, String routingKey)
      Parameters:
      exchange - where it publishes, empty for the default exchange
      routingKey - the key it publishes under
      Returns:
      this producer

    • routingKeys

      public ProducerNode routingKeys(String... keys)
      Publishes across several keys, one after another.

      What makes a topic exchange behave like a topic exchange. A producer on a single key measures one binding however many bindings the exchange has.

      Parameters:
      keys - the keys, used in turn
      Returns:
      this producer

    • rate

      public ProducerNode rate(long messagesPerSecond)
      Parameters:
      messagesPerSecond - the offered rate, on a schedule
      Returns:
      this producer

    • unthrottled

      public ProducerNode unthrottled()
      As fast as it can, with no schedule.

      Finds the ceiling, and makes the latency meaningless while doing it: an unthrottled generator stalls when the broker stalls, so it stops offering load exactly when the load was the interesting part.

      Returns:
      this producer

    • threads

      public ProducerNode threads(int threads)
      Parameters:
      threads - how many threads offer this producer's rate, or 0 to work it out
      Returns:
      this producer

    • messageSize

      public ProducerNode messageSize(int bytes)
      Parameters:
      bytes - how big each message body is
      Returns:
      this producer

    • confirms

      public ProducerNode confirms(boolean confirms)
      Parameters:
      confirms - whether to wait for the broker to confirm
      Returns:
      this producer

    • maxMessages

      public ProducerNode maxMessages(long max)
      Parameters:
      max - stop after this many messages, whatever the clock says
      Returns:
      this producer

    • maxInFlight

      public ProducerNode maxInFlight(int inFlight)
      Parameters:
      inFlight - how many publishes may be outstanding before it waits
      Returns:
      this producer

    • enabled

      public ProducerNode enabled(boolean enabled)
      Parameters:
      enabled - whether it publishes at all
      Returns:
      this producer

    • expect

      public ProducerNode expect(Consumer<Expect> objectives)
      What this producer is asked to prove.
      Parameters:
      objectives - what to require of it
      Returns:
      this producer

    • expectations

      public Expect expectations()
      Returns:
      what this producer is asked to prove, empty when nothing was asked

    • name

      public String name()
      Specified by:
      name in interface Node
      Returns:
      what this node is called

    • exchange

      public String exchange()

    • routingKeys

      public List<String> routingKeys()
      Returns:
      the keys it publishes under, never empty: an unset key is the empty one

    • routingKey

      public String routingKey()
      Returns:
      the first key, which is the only one for a producer that named one

    • rate

      public long rate()

    • isUnthrottled

      public boolean isUnthrottled()

    • threadCount

      public int threadCount()
      How many threads this producer actually needs.

      Worked out from the rate when nobody said: a thread can offer a few tens of thousands a second before the schedule starts slipping, and a run whose generator could not keep up is a run that measured the generator. Somebody who knows better says threads(int).

      Returns:
      the thread count to use

    • messageSize

      public int messageSize()

    • payload

      public ProducerNode payload(Payload payload)
      What the messages are made of.

      The size is enough for nearly everything, and this is for the case where it is not: Payload.ofRandomBytes(int) gives every message a different body, which is what to use when a broker, a filesystem or a link between them might be compressing or deduplicating identical bodies and flattering the result.

      Parameters:
      payload - the payload, or null for the plain one of messageSize(int) bytes
      Returns:
      this producer

    • payload

      public Payload payload()
      Returns:
      what the messages are made of

    • confirms

      public boolean confirms()

    • maxMessages

      public long maxMessages()

    • maxInFlight

      public int maxInFlight()

    • isEnabled

      public boolean isEnabled()
      Specified by:
      isEnabled in interface Node
      Returns:
      whether it takes part in the run

    • toString

      public String toString()
      Overrides:
      toString in class Object