1"use strict";(globalThis.webpackChunknbomber_docs=globalThis.webpackChunknbomber_docs||[]).push([[2003],{6968(e,n,a){a.r(n),a.d(n,{assets:()=>m,contentTitle:()=>h,default:()=>u,frontMatter:()=>l,metadata:()=>t,toc:()=>p});const t=JSON.parse('{"id":"protocols/amqp","title":"AMQP","description":"AMQP (Advanced Message Queuing Protocol) is an open standard for messaging middleware that enables applications to communicate with each other by sending messages in a reliable and interoperable way. It is a protocol designed to provide a robust messaging infrastructure for distributed systems, enabling asynchronous communication between different components.","source":"@site/docs/protocols/amqp.md","sourceDirName":"protocols","slug":"/protocols/amqp","permalink":"/docs/protocols/amqp","draft":false,"unlisted":false,"editUrl":"https://github.com/PragmaticFlow/NBomber.Docs/blob/dev/docs/protocols/amqp.md","tags":[],"version":"current","sidebarPosition":4,"frontMatter":{"id":"amqp","title":"AMQP","sidebar_position":4},"sidebar":"tutorialSidebar","previous":{"title":"MQTT","permalink":"/docs/protocols/mqtt"},"next":{"title":"Datadog","permalink":"/docs/reporting/realtime/datadog"}}');var i=a(4848),s=a(8453);const r=a.p+"assets/images/amqp-e042cbe7824d9363663942ae8f3fb00f.jpg",o=a.p+"assets/images/amqp_client_pool-571665ffd171f33164f6a16749fb776c.jpg",c=a.p+"assets/images/amqp_independent_actors-9432f9d5e1c55a21ea9fac55ec645d05.jpg",l={id:"amqp",title:"AMQP",sidebar_position:4},h=void 0,m={},p=[{value:"AMQP API",id:"amqp-api",level:2},{value:"Original RabbitMQ.Client",id:"original-rabbitmqclient",level:3},{value:"Examples",id:"examples",level:2},{value:"Ping Pong Example",id:"ping-pong-example",level:3},{value:"ClientPool Example",id:"clientpool-example",level:3},{value:"Independent Actors Example",id:"independent-actors-example",level:3}];function d(e){const n={a:"a",admonition:"admonition",blockquote:"blockquote",code:"code",em:"em",h2:"h2",h3:"h3",img:"img",li:"li",ol:"ol",p:"p",pre:"pre",strong:"strong",ul:"ul",...(0,s.R)(),...e.components};return(0,i.jsxs)(i.Fragment,{children:[(0,i.jsx)("center",{children:(0,i.jsx)("img",{src:r,width:"70%",height:"70%"})}),"\n",(0,i.jsx)(n.p,{children:"AMQP (Advanced Message Queuing Protocol) is an open standard for messaging middleware that enables applications to communicate with each other by sending messages in a reliable and interoperable way. It is a protocol designed to provide a robust messaging infrastructure for distributed systems, enabling asynchronous communication between different components."}),"\n",(0,i.jsxs)(n.p,{children:["To work with AMQP protocol, NBomber provides ",(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber.AMQP",children:"NBomber.AMQP"})," plugin that includes functionality for sending and receiving messages, including tracking of data transfer and status codes."]}),"\n",(0,i.jsx)(n.admonition,{type:"warning",children:(0,i.jsx)(n.p,{children:"This package is experimental and might be subject to breaking API changes in the future. While we intend to keep experimental packages as stable as possible, we may need to introduce breaking changes."})}),"\n",(0,i.jsxs)(n.admonition,{type:"info",children:[(0,i.jsxs)(n.p,{children:["To install ",(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber.AMQP",children:"NBomber.AMQP"})," package you should execute the following ",(0,i.jsx)(n.em,{children:"dotnet"})," command:"]}),(0,i.jsxs)(n.p,{children:[(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber.AMQP",children:(0,i.jsx)(n.img,{src:"https://github.com/PragmaticFlow/NBomber.AMQP/actions/workflows/build.yml/badge.svg",alt:"build"})}),"\n",(0,i.jsx)(n.a,{href:"https://www.nuget.org/packages/nbomber.amqp/",children:(0,i.jsx)(n.img,{src:"https://img.shields.io/nuget/v/nbomber.amqp.svg",alt:"NuGet"})})]}),(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-code",children:"dotnet add package NBomber.AMQP\n"})})]}),"\n",(0,i.jsx)(n.h2,{id:"amqp-api",children:"AMQP API"}),"\n",(0,i.jsxs)(n.p,{children:["AMQP plugin provides a wrapper over the popular library ",(0,i.jsx)(n.a,{href:"https://www.rabbitmq.com/client-libraries/dotnet",children:"RabbitMQ.Client"}),". The wrapper implements basic methods for publishing and receiving messages."]}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'var scenario = Scenario.Create("amqp_scenario", async ctx =>\n{\n using var amqpClient = new AmqpClient(channel);\n\n // Declares an AMQP exchange and queue, then binds the queue to the exchange using the specified routing key.\n await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: queueName, routingKey: queueName);\n\n // Subscribes to the specified AMQP queue by adding a consumer.\n await amqpClient.Subscribe(queue: "queueName"); \n\n // Publishes a message to the specified AMQP exchange using the given routing key.\n await amqpClient.Publish(exchange: "myExchange", routingKey: "routingKey", body: payload);\n\n // Awaits and receives a message from the subscribed queue.\n var response = await amqpClient.Receive();\n\n // Disconnect the current client from the broker\n await amqpClient.Disconnect();\n\n // Gets the total number of messages received by the client.\n var receivedCount = amqpClient.MsgReceivedCount\n\n return Response.Ok();\n});\n'})}),"\n",(0,i.jsx)(n.h3,{id:"original-rabbitmqclient",children:"Original RabbitMQ.Client"}),"\n",(0,i.jsxs)(n.p,{children:["NBomber AMQP plugin is basically a wrapper over the popular library ",(0,i.jsx)(n.a,{href:"https://www.rabbitmq.com/client-libraries/dotnet",children:"RabbitMQ.Client"}),". If you need to work with the original ",(0,i.jsx)(n.em,{children:"API"})," from ",(0,i.jsx)(n.em,{children:"RabbitMQ.Client"})," library, you can use public ",(0,i.jsx)(n.code,{children:"AmqpChannel"})," property. All native methods are available for usage."]}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'using var client = new AmqpClient(channel);\n\nvar originalClient = client.AmqpChannel;\n\nawait originalClient.ExchangeDeclareAsync(exchange: "myExchange", type: ExchangeType.Direct);\n'})}),"\n",(0,i.jsx)(n.h2,{id:"examples",children:"Examples"}),"\n",(0,i.jsx)(n.p,{children:"Here you will find some useful examples of working with the AMQP protocol and NBomber to cover different workloads."}),"\n",(0,i.jsx)(n.h3,{id:"ping-pong-example",children:"Ping Pong Example"}),"\n",(0,i.jsx)(n.p,{children:"This is a basic example meant to demonstrate the API usage. In this example, we create an AMQP client that:"}),"\n",(0,i.jsxs)(n.ol,{children:["\n",(0,i.jsx)(n.li,{children:"Connects to the broker"}),"\n",(0,i.jsx)(n.li,{children:"Subscribes to its own queue (self topic) using the scenario instance ID (ctx.ScenarioInfo.InstanceId)"}),"\n",(0,i.jsx)(n.li,{children:"Publishes a message to self topic"}),"\n",(0,i.jsx)(n.li,{children:"Awaits and receives a message from the self topic"}),"\n",(0,i.jsx)(n.li,{children:"Disconnects"}),"\n"]}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'var payload = Data.GenerateRandomBytes(200);\nvar factory = new ConnectionFactory { HostName = "localhost" };\n\nvar scenario = Scenario.Create("ping_pong_scenario", async ctx =>\n{\n var connect = await Step.Run("connect", ctx, async () =>\n {\n var connection = await factory.CreateConnectionAsync();\n var channel = await connection.CreateChannelAsync();\n\n var amqpClient = new AmqpClient(channel);\n return Response.Ok(payload: amqpClient);\n });\n\n using var amqpClient = connect.Payload.Value;\n\n var subscribe = await Step.Run("subscribe", ctx, async () =>\n {\n var queueName = ctx.ScenarioInfo.InstanceId;\n\n await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, queue: queueName,\n routingKey: queueName);\n\n return await amqpClient.Subscribe(queue: queueName, autoAck: true);\n });\n\n var publish = await Step.Run("publish", ctx, async () =>\n {\n var queueName = ctx.ScenarioInfo.InstanceId;\n return await amqpClient.Publish(exchange: "myExchange", routingKey: queueName, body: payload);\n });\n\n var receive = await Step.Run("receive", ctx, async () =>
1\n {\n // pass the ScenarioCancellationToken to stop waiting for a response if the scenario finish event is triggered\n var response = await amqpClient.Receive(ctx.ScenarioCancellationToken);\n return response;\n });\n\n var disconnect = await Step.Run("disconnect", ctx, async () =>\n {\n await amqpClient.Disconnect();\n return Response.Ok();\n });\n\n return Response.Ok();\n})\n'})}),"\n",(0,i.jsx)(n.p,{children:(0,i.jsxs)(n.em,{children:["You can find the complete example by this ",(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber/blob/dev/examples/Demo/AMQP/PingPongAmqpTest.cs",children:"link"}),"."]})}),"\n",(0,i.jsx)(n.h3,{id:"clientpool-example",children:"ClientPool Example"}),"\n",(0,i.jsx)("center",{children:(0,i.jsx)("img",{src:o,width:"70%",height:"70%"})}),"\n",(0,i.jsxs)(n.p,{children:["In this example, we create and initialize a list of AMQP clients, which remain active throughout the entire load test using a ",(0,i.jsx)(n.a,{href:"../nbomber/client-pool",children:"ClientPool"})," abstraction. Each client continuously and sequentially performs two steps:"]}),"\n",(0,i.jsxs)(n.ul,{children:["\n",(0,i.jsxs)(n.li,{children:[(0,i.jsx)(n.strong,{children:"publishes a message"})," to self topic"]}),"\n",(0,i.jsxs)(n.li,{children:[(0,i.jsx)(n.strong,{children:"receives a message"})," from the self topic"]}),"\n"]}),"\n",(0,i.jsx)(n.p,{children:"All clients run in parallel to simulate multiple connected clients. This test measures how active concurrent connections impact the performance of the message broker. Also it measures both the throughput and latency for publish and receive operations."}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'// Create a ClientPool, which will be used to hold active AMQP connections/clients\nvar clientPool = new ClientPool<AmqpClient>();\n\nbyte[] message = [];\nvar usePersistence = false;\n\nvar scenario = Scenario.Create("client_pool_scenario", async ctx =>\n{\n // get a client from the pool by Scenario InstanceNumber\n // highlight-start\n var client = clientPool.GetClient(ctx.ScenarioInfo);\n // highlight-end\n \n var publish = await Step.Run("publish", ctx, async () =>\n {\n var queueName = $"queue_{ctx.ScenarioInfo.InstanceNumber}";\n var props = new BasicProperties { Persistent = usePersistence };\n\n var response = await client.Publish(exchange: "myExchange", routingKey: queueName, props, message);\n return response;\n });\n\n var receive = await Step.Run("receive", ctx, async () =>\n {\n // pass the ScenarioCancellationToken to stop waiting for a response if the scenario finish event is triggered\n var response = await client.Receive(ctx.ScenarioCancellationToken);\n return response;\n }); \n\n return Response.Ok();\n})\n.WithInit(async context =>\n{ \n var config = context.CustomSettings.Get<CustomScenarioSettings>();\n\n for (var i = 0; i < config.ClientCount; i++)\n {\n // initialize a client and add it to the ClientPool \n var connection = await factory.CreateConnectionAsync();\n var channel = await connection.CreateChannelAsync();\n var amqpClient = new AmqpClient(channel);\n\n var queueName = $"queue_{i}";\n\n var result = await amqpClient.DeclareQueue(exchange: "myExchange", exchangeType: ExchangeType.Direct, \n queue: queueName, routingKey: queueName, durable: usePersistence);\n\n if (!result.IsError)\n {\n await amqpClient.Subscribe(queue: queueName);\n // highlight-start\n clientPool.AddClient(amqpClient);\n // highlight-end\n }\n else\n throw new Exception("client can\'t connect to the AMQP broker"); \n }\n})\n.WithClean(ctx =>\n{\n clientPool.DisposeClients(client => client.Dispose());\n return Task.CompletedTask;\n});\n'})}),"\n",(0,i.jsx)(n.p,{children:(0,i.jsxs)(n.em,{children:["You can find the complete example by this ",(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber/blob/dev/examples/Demo/AMQP/ClientPool/ClientPoolAmqpExample.cs",children:"link"}),"."]})}),"\n",(0,i.jsx)(n.h3,{id:"independent-actors-example",children:"Independent Actors Example"}),"\n",(0,i.jsx)("center",{children:(0,i.jsx)("img",{src:c,width:"100%",height:"100%"})}),"\n",(0,i.jsx)(n.p,{children:"There could be situations where you have an independent Publisher/Writer and Consumer/Reader, each w
1orking separately. In such cases, you may want to measure both the throughput and the message latency from the publisher to the consumer."}),"\n",(0,i.jsx)(n.p,{children:"This particular case is usually tricky to cover with any load testing framework."}),"\n",(0,i.jsxs)(n.blockquote,{children:["\n",(0,i.jsxs)(n.p,{children:["Imagine two messages being published by two independent publishers ",(0,i.jsx)(n.strong,{children:"at the same time"}),", say at ",(0,i.jsx)(n.code,{children:'"00:00:10"'}),". The consumer waits to receive the messages \u2014 the first received message will have its latency measured correctly, ",(0,i.jsx)(n.strong,{children:"but the second message will be processed immediately after the first one"}),". This will distort the latency measurement for the second message, making it appear as if there was zero delay."]}),"\n"]}),"\n",(0,i.jsx)(n.p,{children:"Example:"}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'// It\'s a pseudo-code just to show how Publisher and Consumer scenarios could be modeled.\n// Publisher: sends messages to a topic\nvar publish_scenario = Scenario.Create("publish_scenario", async ctx =>\n{\n var payload = Data.GenerateRandomBytes(200);\n var prop = new BasicProperties();\n\n return await amqpClient.Publish(exchange: "myExchange", routingKey: "myRoutingKey", prop, payload); \n})\n.WithLoadSimulations( \n // we run 2 Publisher(s), which will send messages concurrently\n // highlight-start\n Simulation.KeepConstant(2, TimeSpan.FromSeconds(30)) \n // highlight-end\n);\n\n// Consumer: listens for messages and measures latency\nvar consume_scenario = Scenario.Create("consume_scenario", async ctx =>\n{\n var message = await amqpClient.Receive(); // The second message will be received immediately after the first one,\n // because both messages were published at the same time. \n // This will distort the latency measurement for the second message,\n return Response.Ok(); // making it appear as if there was zero delay \n})\n.WithLoadSimulations( \n // we run 1 Consumer\n // highlight-start\n Simulation.KeepConstant(1, TimeSpan.FromSeconds(30)) \n // highlight-end\n);\n\nNBomberRunner\n .RegisterScenarios(publish_scenario, consume_scenario)\n .Run();\n'})}),"\n",(0,i.jsxs)(n.admonition,{type:"info",children:[(0,i.jsx)(n.p,{children:"In order to fix this measurement, we need to do a few things:"}),(0,i.jsxs)(n.ul,{children:["\n",(0,i.jsxs)(n.li,{children:["On the Publisher side: each message should be enriched with a current ",(0,i.jsx)(n.code,{children:"timestamp"}),"."]}),"\n",(0,i.jsxs)(n.li,{children:["On the Consumer side: we read the received message's ",(0,i.jsx)(n.code,{children:"timestamp"})," and calculate the latency:"]}),"\n"]}),(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:"var customLatency = DateTime.UtcNow - msg.Timestamp\n"})}),(0,i.jsxs)(n.ul,{children:["\n",(0,i.jsxs)(n.li,{children:["We instruct NBomber to use the custom latency via ",(0,i.jsx)(n.code,{children:"Response.Ok(customLatency: value)"})]}),"\n"]})]}),"\n",(0,i.jsx)(n.p,{children:"Example:"}),"\n",(0,i.jsx)(n.pre,{children:(0,i.jsx)(n.code,{className:"language-csharp",children:'var publish_scenario = Scenario.Create("publish_scenario", async ctx =>\n{ \n // highlight-start\n var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();\n // highlight-end\n\n var props = new BasicProperties\n {\n // We include the current timestamp so the consumer can calculate the final latency.\n Headers = new Dictionary<string, object> { { "timestamp", timestamp } }\n };\n\n return await amqpClient.Publish(exchange: "myExchange", routingKey: "myRoutingKey", props, payload); \n})\n.WithLoadSimulations( \n Simulation.KeepConstant(2, TimeSpan.FromSeconds(30)) \n);\n\nvar consume_scenario = Scenario.Create("consume_scenario", async ctx =>\n{\n var message = await amqpClient.Receive(ctx.ScenarioCancellationToken);\n\n // Final latency is computed by subtracting the current time from the timestamp in the header.\n var timestampMs = (long)message.Payload.Value.BasicProperties.Headers["timestamp"];
1\n var latency = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - timestampMs;\n\n // highlight-start\n return Response.Ok(customLatencyMs: latency, sizeBytes: message.SizeBytes); \n // highlight-end\n})\n.WithLoadSimulations( \n Simulation.KeepConstant(1, TimeSpan.FromSeconds(30)) \n);\n\nNBomberRunner\n .RegisterScenarios(publish_scenario, consume_scenario)\n .Run();\n'})}),"\n",(0,i.jsx)(n.p,{children:(0,i.jsxs)(n.em,{children:["You can find the complete example by this ",(0,i.jsx)(n.a,{href:"https://github.com/PragmaticFlow/NBomber/tree/dev/examples/Demo/AMQP/IndependentActors",children:"link"}),"."]})})]})}function u(e={}){const{wrapper:n}={...(0,s.R)(),...e.components};return n?(0,i.jsx)(n,{...e,children:(0,i.jsx)(d,{...e})}):d(e)}},8453(e,n,a){a.d(n,{R:()=>r,x:()=>o});var t=a(6540);const i={},s=t.createContext(i);function r(e){const n=t.useContext(s);return t.useMemo(function(){return"function"==typeof e?e(n):{...n,...e}},[n,e])}function o(e){let n;return n=e.disableParentContext?"function"==typeof e.components?e.components(i):e.components||i:r(e.components),t.createElement(s.Provider,{value:n},e.children)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.