Introduction

Gatling AMQP support

CAUTION: This is not official library!
- but using 'io.gatling.amqp' for FQCN to deal with 'private[gatling]', sorry.
inspired by https://github.com/fhalim/gatling-rabbitmq forked from https://github.com/maiha/gatling-amqp (thanks!)

Library

Current libraries version

scala 2.12.6 (no support for scala 2.11.x)
amqp-client-4.9.0
gatling-2.3.1
gatling-sbt-2.2.2 (whit dependency to SBT [1.2.3))

Usage

Import library in build.sbt

libraryDependencies += "io.github.dieselr" %% "gatling-amqp" % "0.11"

handy cli (use AmqpProtocol as console utility) [feature from 0.6]

scala> import io.gatling.amqp.Predef._
scala> amqp.declare(queue("q3", durable = true)).run

publish (normal)

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .poolSize(10)

  val req = PublishRequest("q1", "{foo:1}")

  val scn = scenario("AMQP Publish").repeat(1000) {
    exec(amqp("Publish").publish(req))
  }

  setUp(scn.inject(rampUsers(10) over (1 seconds)))
    .protocols(amqpProtocol)

publish (with persistent)

PublishRequest.persistent make request DeliveryMode(2)
known issue: persistent reset request's properties

  val req = PublishRequest("q1", "{foo:1}").persistent

publish (with confirmation)

set "confirmMode()" in protocol that invokes "channel.confirmSelect()"

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .poolSize(10)
    .confirmMode()

  val req = PublishRequest("q1", "{foo:1}")

  val scn = scenario("AMQP Publish(ack)").repeat(1000) {
    exec(amqp("Publish").publish(req))
  }

  setUp(scn.inject(rampUsers(10) over (1 seconds)))
    .protocols(amqpProtocol)

declare queues

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .declare(queue("q1", durable = true, autoDelete = false))

declare exchange and binding

  val x = exchange("color", "direct", autoDelete = false)
  val q = queue("orange")
  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .declare(x)
    .declare(q)
    .bind(x, q, routingKey = "orange")

full code: src/test/scala/io/gatling/amqp/PublishingSimulation.scala

consume (auto acked)

!!!It is not Supported after migration to newest libraries

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("amqp")
    .port(5672)
    .auth("guest", "guest")

  val scn = scenario("AMQP Publish(ack)").exec {
    amqp("Consume").consume("q1", autoAck = true)
  }

  setUp(scn.inject(atOnceUsers(1)))
    .protocols(amqpProtocol)

full code: src/test/scala/io/gatling/amqp/ConsumingSimulation.scala

consume (manual acked)

not implemented yet

working with session context

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .poolSize(10)
    .confirmMode()

  val value = 1

  val scn = scenario("AMQP Publish(ack)").repeat(1000) {
    exec(amqp("Publish").publish(
      session => 
        PublishRequest("q1", s"{foo:${value}")
      )
    )
  }

  setUp(scn.inject(rampUsers(10) over (1 seconds)))
    .protocols(amqpProtocol)

Note: All publish anc consume methods has support for usage of session and alias replacing from session attribute.

Run

sbt directly

% sbt
> gatling:test

% sbt
> gatling:testOnly io.gatling.amqp.PublishingSimulation

% sbt
> gatling:testOnly io.gatling.amqp.ConsumingSimulation

Note: try sbt -J-Xmx8192m -J-XX:MaxPermSize=256m for publishing massive messages

shell script to store gatling stdout logs and simulation sources

!!! Need to be checked if it is working

% ./run p foo

stored in "stats/p/foo"

Benchmark

After migration to newest libraries version no benchmark was run. Reference to old Benchmark.

TODO

declare exchanges, queues and bindings in action builder context (to test declaration costs)
make AmqpProtocol immutable
make Builder mutable
mandatory
consume action (manual ack)
publish library to public repository
enrich support for different signature of methods: publish and consume
migrate to new gatling-3.x version
migrate to new amqp-client-5.x version
remove usage of library pl.project13.scala:rainbow

License

released under the MIT License.

dieselr / gatling-amqp 0.11