improved AMQP examples

Author: AntyaDev

examples/Demo/AMQP/ClientPool/ClientPoolAmqpExample.cs

@@ -25,21 +25,23 @@ public void Run()
         byte[] message = [];
         var usePersistence = false;
 
-        var scenario = Scenario.Create("amqp_scenario", async ctx =>
+        var scenario = Scenario.Create("client_pool_scenario", async ctx =>
         {
+            // get a client from the pool by Scenario InstanceID
             var client = clientPool.GetClient(ctx.ScenarioInfo);
 
             var publish = await Step.Run("publish", ctx, async () =>
             {
-                var scenarioInstanceId = ctx.ScenarioInfo.InstanceId;
-                var prop = new BasicProperties { Persistent = usePersistence };
+                var queueName = $"queue_{ctx.ScenarioInfo.InstanceNumber}";
+                var props = new BasicProperties { Persistent = usePersistence };
 
-                var response = await client.Publish(exchange: "myExchange", routingKey: scenarioInstanceId, basicProperties: prop, body: message);
+                var response = await client.Publish(exchange: "myExchange", routingKey: queueName, props, message);
                 return response;
             });
 
             var receive = await Step.Run("receive", ctx, async () =>
             {
+                // pass the ScenarioCancellationToken to stop waiting for a response if the scenario finish event is triggered
                 var response = await client.Receive(ctx.ScenarioCancellationToken);
                 return response;
             });
@@ -54,19 +56,21 @@ public void Run()
 
             var factory = new ConnectionFactory { HostName = config.AmqpServerUrl };
 
+            // initialize a client and add it to the ClientPool
             for (var i = 0; i < config.ClientCount; i++)
             {
                 var connection = await factory.CreateConnectionAsync();
                 var channel = await connection.CreateChannelAsync();
                 var amqpClient = new AmqpClient(channel);
 
-                var scenarioInstanceId = $"amqp_scenario_{i}";
-                var result = await amqpClient.Connect(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: scenarioInstanceId,
-                        routingKey: scenarioInstanceId, durable: usePersistence);
+                var queueName = $"queue_{i}";
+
+                var result = await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: queueName,
+                        routingKey: queueName, durable: usePersistence);
 
                 if (!result.IsError)
                 {
-                    await amqpClient.Subscribe(queue: scenarioInstanceId);
+                    await amqpClient.Subscribe(queue: queueName);
                     clientPool.AddClient(amqpClient);
                 }
                 else

examples/Demo/AMQP/ClientPool/config.json

@@ -2,12 +2,12 @@
     "TestSuite": "AMQP",
     "TestName": "client_pool",
 
-    "TargetScenarios": [ "amqp_scenario" ],
+    "TargetScenarios": [ "client_pool_scenario" ],
 
     "GlobalSettings": {
         "ScenariosSettings": [
             {
-                "ScenarioName": "amqp_scenario",
+                "ScenarioName": "client_pool_scenario",
 
                 "WarmUpDuration": "00:00:05",
 

examples/Demo/AMQP/IndependentActors/AmqpPublishScenario.cs

@@ -21,16 +21,14 @@ public ScenarioProps Create()
             var publish = await Step.Run("publish", ctx, async () =>
             {
                 var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-                var prop = new BasicProperties
+
+                var props = new BasicProperties
                 {
                     // We include the current timestamp so the consumer can calculate the final latency.
-                    Headers = new Dictionary<string, object>
-                    {
-                        { "timestamp", timestamp }
-                    }
+                    Headers = new Dictionary<string, object> { { "timestamp", timestamp } }
                 };
 
-                return await amqpClient.Publish(exchange: "myExchange", routingKey: "IndependentActors", prop, body: payload);
+                return await amqpClient.Publish(exchange: "myExchange", routingKey: "myQueue", props, payload);
             });
 
             return Response.Ok();
@@ -45,8 +43,8 @@ public ScenarioProps Create()
             var channel = await connection.CreateChannelAsync();
             amqpClient = new AmqpClient(channel);
 
-            await amqpClient.Connect(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: "IndependentActors",
-                routingKey: "IndependentActors");
+            await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: "myQueue",
+                routingKey: "myQueue");
         })
         .WithClean(async ctx =>
         {

examples/Demo/AMQP/IndependentActors/AmqpConsumeScenario.cs

@@ -35,10 +35,10 @@ public ScenarioProps Create()
             var channel = await connection.CreateChannelAsync();
             amqpClient = new AmqpClient(channel);
 
-            await amqpClient.Connect(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: "IndependentActors",
-                routingKey: "IndependentActors");
+            await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: "myQueue",
+                routingKey: "myQueue");
 
-            await amqpClient.Subscribe(queue: "IndependentActors", autoAck: true);
+            await amqpClient.Subscribe(queue: "myQueue", autoAck: true);
         })
         .WithClean(async ctx =>
         {

examples/Demo/AMQP/PingPongAmqpTest.cs

@@ -14,39 +14,39 @@ public void Run()
         var payload = Data.GenerateRandomBytes(200);
         var factory = new ConnectionFactory { HostName = "localhost" };
 
-        var scenario = Scenario.Create("ping_pong_amqp_scenario", async ctx =>
+        var scenario = Scenario.Create("ping_pong_scenario", async ctx =>
         {
-            AmqpClient amqpClient = null;
-
             var connect = await Step.Run("connect", ctx, async () =>
             {
                 var connection = await factory.CreateConnectionAsync();
                 var channel = await connection.CreateChannelAsync();
 
-                amqpClient = new AmqpClient(channel);
-
-                var scenarioInstanceId = ctx.ScenarioInfo.InstanceId;
-
-                return await amqpClient.Connect(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: scenarioInstanceId,
-                    routingKey: scenarioInstanceId);
+                var amqpClient = new AmqpClient(channel);
+                return Response.Ok(payload: amqpClient);
             });
 
+            using var amqpClient = connect.Payload.Value;
+
             var subscribe = await Step.Run("subscribe", ctx, async () =>
             {
                 var queueName = ctx.ScenarioInfo.InstanceId;
+
+                await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: queueName,
+                    routingKey: queueName);
+
                 return await amqpClient.Subscribe(queue: queueName, autoAck: true);
             });
 
             var publish = await Step.Run("publish", ctx, async () =>
             {
                 var queueName = ctx.ScenarioInfo.InstanceId;
-                var prop = new BasicProperties();
-                return await amqpClient.Publish(exchange: "myExchange", routingKey: queueName, prop, body: payload);
+                return await amqpClient.Publish(exchange: "myExchange", routingKey: queueName, body: payload);
             });
 
             var receive = await Step.Run("receive", ctx, async () =>
             {
-                var response = await amqpClient.Receive().AsTask();
+                // Here, we pass the ScenarioCancellationToken to stop waiting for a response if the scenario finish event is triggered
+                var response = await amqpClient.Receive(ctx.ScenarioCancellationToken);
                 return response;
             });
 

examples/Demo/Demo.csproj

@@ -85,7 +85,7 @@
 
         <PackageReference Include="NBomber" Version="6.1.0-beta.1" />
 
-        <PackageReference Include="NBomber.AMQP" Version="0.1.3-beta.2" />
+        <PackageReference Include="NBomber.AMQP" Version="0.2.0" />
         <PackageReference Include="NBomber.Data" Version="6.0.0" />
         <PackageReference Include="NBomber.Http" Version="6.0.2" />
         <PackageReference Include="NBomber.MQTT" Version="0.4.1-beta.2" />

examples/Demo/MQTT/PingPongMqttTest.cs

@@ -45,6 +45,7 @@ public void Run()
 
             var receive = await Step.Run("receive", ctx, async () =>
             {
+                // pass the ScenarioCancellationToken to stop waiting for a response if the scenario finish event is triggered
                 var response = await mqttClient.Receive(ctx.ScenarioCancellationToken);
                 return response;
             });