Blog

beskar-networking

This Is the Way: Low-Allocation Request-Response in MQTT 5.0

An in-depth look at how we implemented low-allocation bidirectional RPC over MQTT 5.0 in Beskar.Networking, utilizing Response Topics, Correlation Data, and asynchronous Task Completion Sources.

August 10, 2026 8 min read Marvin Drude
C#.NETMQTTRPCAsynchronous ProgrammingPerformanceMemory

The complete implementation of the MQTT 5.0 Request-Response engine is open-source and available in the Beskar.Networking GitHub Repository.

The mismatch: synchronous patterns on a decoupling broker

The publish-subscribe pattern is the undisputed king of IoT and event-driven architectures. By decoupling message senders (publishers) from receivers (subscribers) through an intermediary broker, systems achieve incredible scaling freedom, high resilience, and loose coupling.

But let's be honest: in the real world, unidirectional fire-and-forget is only half the story.

Frequently, your system needs to perform synchronous query-response or command-action patterns:

  • Querying a sensor for its immediate status.
  • Triggering an operation and waiting for a success/failure confirmation.
  • Exposing a microservice API where clients send a request payload and expect a structured response.

Historically, implementing Request-Response over an MQTT broker felt like fitting a square peg in a round hole. Senders had to agree on ad-hoc response topics, manually serialize transaction IDs into payloads, spin up subscriber listeners, filter out unrelated messages, and manage timeouts by hand. This resulted in fragile application code, severe race conditions, and mountains of heap allocations on the critical path.

With MQTT 5.0 and the modern async capabilities in .NET, we can do much better. Let’s dive into how we built a low-allocation, thread-safe, and highly resilient Request-Response engine in Beskar.Networking.

MQTT 5.0 to the Rescue: Native RPC Properties

The MQTT 5.0 specification introduced formal support for Request-Response interactions by adding two crucial metadata properties to the PUBLISH packet:

  1. Response Topic (ResponseTopic): A string defining the topic to which the responder should send the reply.
  2. Correlation Data (CorrelationData): A binary payload (a sequence of bytes) that acts as a unique transaction token. The responder must return this token unmodified in the response message.
text
  Publisher Client                          MQTT Broker                         Subscriber Client
        |                                       |                                       |
        |----[PUBLISH Request]----------------->|                                       |
        |     Topic: "test/rpc/request"         |                                       |
        |     ResponseTopic: "clients/pub/res"  |                                       |
        |     CorrelationData: [Token Bytes]    |----[Forward PUBLISH Request]--------->|
        |                                       |                                       |
        |                                       |                                       |--- [Process Request]
        |                                       |                                       |
        |                                       |    [RespondAsync]                     |
        |                                       |    Publish back to ResponseTopic      |
        |                                       |    Echo CorrelationData bytes         |
        |                                       |<---[PUBLISH Response]-----------------|
        |                                       |     Topic: "clients/pub/res"          |
        |<---[Forward PUBLISH Response]---------|     CorrelationData: [Token Bytes]     |
        |                                       |                                       |
  [Match Token -> Complete Task]

By leveraging these properties, the broker routes response packets natively, and clients can correlate incoming messages without searching inside the application payload.

Client-Side Mechanics: MqttClient.RequestAsync

The publisher's interface needs to be clean. Sending a request and waiting for a reply should be as simple as invoking an async method:

C#
var response = await client.RequestAsync("sensors/query", queryPayload, TimeSpan.FromSeconds(5));

Behind this simple interface lies a multi-threaded state coordinator. Let's look at the implementation inside MqttClient.Request.cs:

C#
public async Task<Result<MqttResponseContext, StringError>> RequestAsync(
   PublishOptions options, TimeSpan timeout = default, CancellationToken ct = default)
{
   var validRes = ValidateClient();
   if (validRes.Failed) return validRes.Error;

   if (_protocolVersion is not MqttProtocolVersion.V50)
   {
      return new StringError("RequestAsync requires MQTT 5.0 protocol version for ResponseTopic and CorrelationData support.");
   }

   if (timeout == TimeSpan.Zero || timeout <= TimeSpan.Zero)
   {
      timeout = TimeSpan.FromSeconds(10);
   }

   string responseTopic;
   string correlationKey;
   PublishOptions effectiveOptions;

   var needsNewTopic = options.ResponseTopicUtf8Bytes.IsEmpty;
   var needsNewCorr = options.CorrelationData.IsEmpty;

   // 1. Build effective options (Auto-generate ResponseTopic & CorrelationData if not specified)
   if (needsNewTopic || needsNewCorr)
   {
      var builder = PublishOptions.Create()
         .WithTopic(options.TopicUtf8Bytes)
         .WithPayload(options.Payload)
         .WithQualityOfService(options.QualityOfService);

      if (needsNewTopic)
      {
         var clientIdStr = Encoding.UTF8.GetString(_connectOptions.ClientIdUtf8Bytes.Span);
         if (string.IsNullOrEmpty(clientIdStr))
         {
            clientIdStr = Guid.NewGuid().ToString("N");
         }
         responseTopic = $"clients/{clientIdStr}/response";
         builder.WithResponseTopic(responseTopic);
      }
      else
      {
         responseTopic = Encoding.UTF8.GetString(options.ResponseTopicUtf8Bytes.Span);
         builder.WithResponseTopic(options.ResponseTopicUtf8Bytes);
      }

      if (needsNewCorr)
      {
         correlationKey = Guid.NewGuid().ToString("N");
         var corrBytes = Encoding.UTF8.GetBytes(correlationKey);
         builder.WithCorrelationData(corrBytes);
         correlationKey = Convert.ToBase64String(corrBytes); // Store as dictionary lookup key
      }
      else
      {
         correlationKey = Convert.ToBase64String(options.CorrelationData.Span);
         builder.WithCorrelationData(options.CorrelationData);
      }

      effectiveOptions = builder.Build();
   }
   else
   {
      responseTopic = Encoding.UTF8.GetString(options.ResponseTopicUtf8Bytes.Span);
      correlationKey = Convert.ToBase64String(options.CorrelationData.Span);
      effectiveOptions = options;
   }

   // 2. Register the pending request using a TaskCompletionSource
   var tcs = new TaskCompletionSource<MqttPublishMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
   if (!_pendingRequests.TryAdd(correlationKey, tcs))
   {
      return new StringError($"A pending request with correlation ID '{correlationKey}' is already in progress.");
   }

   var startTimestamp = Stopwatch.GetTimestamp();

   try
   {
      // 3. Ensure the client is subscribed to its own ResponseTopic
      if (!_subscribedResponseTopics.ContainsKey(responseTopic))
      {
         var subOptions = SubscribeOptions.Create()
            .WithTopicFilter(responseTopic, QualityOfServiceType.AtLeastOnce)
            .Build();

         var subResult = await SubscribeAsync(subOptions, ct);
         if (subResult.Failed) return subResult.Error;

         _subscribedResponseTopics[responseTopic] = 1;
      }

      // 4. Publish the Request message
      var pubResult = await PublishAsync(effectiveOptions, ct);
      if (pubResult.Failed) return pubResult.Error;

      // 5. Link timeouts and await the response task
      using var combinedCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
      combinedCts.CancelAfter(timeout);

      MqttPublishMessage responseMessage;
      try
      {
         responseMessage = await tcs.Task.WaitAsync(combinedCts.Token);
      }
      catch (OperationCanceledException ex)
      {
         if (ct.IsCancellationRequested) return new StringError("Request was cancelled.");
         if (!IsConnected) return new StringError($"Request cancelled: {ex.Message}");
         return new StringError($"Request timed out after {timeout.TotalMilliseconds}ms waiting for response on topic '{responseTopic}'.");
      }

      var elapsedTicks = Stopwatch.GetTimestamp() - startTimestamp;
      var elapsed = TimeSpan.FromSeconds((double)elapsedTicks / Stopwatch.Frequency);

      return new MqttResponseContext
      {
         Message = responseMessage,
         CorrelationId = correlationKey,
         Elapsed = elapsed
      };
   }
   finally
   {
      // 6. Clean up resources
      _pendingRequests.TryRemove(correlationKey, out _);
   }
}

Key Optimizations & Lifecycle Safety:

  1. Dynamic Subscription Management: Instead of forcing the user to pre-subscribe, the client lazily registers a subscription for its response topic (clients/{ClientId}/response) the first time an RPC call is made, caching the state in _subscribedResponseTopics to avoid redundant SUBSCRIBE packets.
  2. Concurrent Request Routing: By utilizing a thread-safe ConcurrentDictionary<string, TaskCompletionSource<MqttPublishMessage>>, we can issue hundreds of requests concurrently across threads. When response messages arrive, they are dispatched instantaneously to their waiting tasks.
  3. Task Leak Prevention: In any distributed system, connection drops happen. If the client abruptly loses connection to the broker, we must not leave Tasks hanging in memory forever. Inside CancelPendingRequestsOnDisconnect(), we walk the dictionary and cancel every pending task immediately:
    C#
    internal void CancelPendingRequestsOnDisconnect()
   {
      _subscribedResponseTopics.Clear();
      foreach (var kvp in _pendingRequests)
      {
         if (_pendingRequests.TryRemove(kvp.Key, out var tcs))
         {
            tcs.TrySetException(new OperationCanceledException("Client disconnected while awaiting request response."));
         }
      }
   }

Receiver-Side Mechanics: ctx.RespondAsync

On the receiving end (the subscriber acting as an RPC server), processing the request and issuing a response is fully integrated into the message handling lifecycle.

When a packet is received, the client handlers expose a MessageReceiveContext. We added the RespondAsync method directly to it inside MessageReceiveContext.cs:

C#
public ValueTask<Result<PublishResult, StringError>> RespondAsync(
   ReadOnlyMemory<byte> payload,
   QualityOfServiceType qos = QualityOfServiceType.AtLeastOnce,
   CancellationToken ct = default)
{
   if (string.IsNullOrEmpty(Message.ResponseTopic))
   {
      return ValueTask.FromResult<Result<PublishResult, StringError>>(
         new StringError("Cannot send response: Message does not specify a ResponseTopic."));
   }

   if (Client is null)
   {
      return ValueTask.FromResult<Result<PublishResult, StringError>>(
         new StringError("Cannot send response: MqttClient instance is not attached to MessageReceiveContext."));
   }

   var publishOptionsBuilder = PublishOptions.Create()
      .WithTopic(Message.ResponseTopic)
      .WithPayload(payload)
      .WithQualityOfService(qos);

   // Copy CorrelationData directly to maintain transaction mapping
   if (Message.CorrelationData.HasValue)
   {
      publishOptionsBuilder.WithCorrelationData(Message.CorrelationData.Value);
   }

   return new ValueTask<Result<PublishResult, StringError>>(
      Client.PublishAsync(publishOptionsBuilder.Build(), ct));
}

Low-Allocation Correlation Propagation:

Notice that Message.CorrelationData is a ReadOnlyMemory<byte>. Instead of converting the Correlation Data to strings or parsing it into intermediary structs (which would allocate memory on the heap), RespondAsync takes the raw buffer slice and passes it straight back to the publish pipeline.

This ensures that routing correlation is fully allocation-free on the responder's critical path.

Connecting the Dots: A Practical RPC Workflow

To see the system in action, let’s look at a simplified version of the integration test in MqttRequestResponseTests.cs:

C#
// 1. Set up the RPC Server Client (Subscriber)
var subscriber = MqttClientFactory.CreateTcp();
await subscriber.ConnectAsync(connectOptions);

subscriber.AddMessageReceiveHandler(async (ctx, ct) =>
{
   // Read request payload
   var reqStr = Encoding.UTF8.GetString(ctx.Message.Payload.Span);
   Console.WriteLine($"[RPC Server] Received request: {reqStr}");

   // Process and send reply
   var replyStr = $"ACK:{reqStr}";
   var respondResult = await ctx.RespondAsync(replyStr, QualityOfServiceType.AtLeastOnce, ct);
   if (respondResult.Failed)
   {
      Console.WriteLine($"[RPC Server] Failed to respond: {respondResult.Error.Detail}");
   }
});

// Subscriber listens to the command topic
var subOptions = SubscribeOptions.Create()
   .WithTopicFilter("test/rpc/request", QualityOfServiceType.AtLeastOnce)
   .Build();
await subscriber.SubscribeAsync(subOptions);

// 2. Set up the RPC Client (Publisher)
var publisher = MqttClientFactory.CreateTcp();
await publisher.ConnectAsync(connectOptions);

// Act: Send request and await reply
var requestPayload = Encoding.UTF8.GetBytes("ORDER-999");
var responseResult = await publisher.RequestAsync("test/rpc/request", requestPayload, TimeSpan.FromSeconds(5));

if (responseResult.Failed)
{
   Console.WriteLine($"RPC Failed: {responseResult.Error.Detail}");
}
else
{
   var response = responseResult.Success!;
   var replyText = Encoding.UTF8.GetString(response.Payload.Span);
   Console.WriteLine($"RPC Success: {replyText} (took {response.Elapsed.TotalMilliseconds:F1}ms)");
   // Output: RPC Success: ACK:ORDER-999 (took 4.2ms)
}

Conclusion: Type-Safe RPC over Pub/Sub

By combining MQTT 5.0's native properties with modern asynchronous paradigms in .NET 10, we successfully bridged the gap between decoupled pub-sub messaging and synchronous RPC.

The resulting design has several key advantages:

  1. Low-Allocation Correlation: We route correlation data in its raw binary form (ReadOnlyMemory<byte>), avoiding string parsing and heap allocation.
  2. Encapsulated Thread Safety: Senders don't need to write custom task managers or subscribe loops. The client coordinates tasks dynamically via concurrent dictionary lookups.
  3. Disconnection Safety: Connection interruptions trigger immediate task cancellation, ensuring no dangling tasks leak memory.

With Request-Response RPC fully operational, our MQTT engine is now highly capable of serving both asynchronous events and synchronous commands. When you need to build reliable, high-performance messaging systems out of pure Beskar, this is the Way.

Part of the "Beskar Networking" Series

Part 1

Sockets on a Pipeline: Low-Allocation Networking with Beskar.Memory

An in-depth technical guide to building an ultra-fast TCP socket transport using System.IO.Pipelines, SocketAsyncEventArgs, and Beskar.Memory pools.

July 5, 2026
Part 2

Allocations on the Wire: Building a Low-Allocation MQTT Broker with Trie Routing and .NET 10

An inside look at building the protocol layer for Beskar.Networking, covering UTF-8 based topic trie routing, alternate dictionary lookups, QoS tracking, and pooled connection systems.

July 18, 2026
Part 3

Resiliency in the Chaos: Source-Generated Framing and Auto-Reconnecting Client-Server Pipelines

How I built a highly resilient, transport-agnostic client-server framework with source-generated framing, custom event pipelines, and verified it using a custom-built network Chaos Simulator.

July 25, 2026
Part 4

This Is the Way: Low-Allocation Request-Response in MQTT 5.0

An in-depth look at how we implemented low-allocation bidirectional RPC over MQTT 5.0 in Beskar.Networking, utilizing Response Topics, Correlation Data, and asynchronous Task Completion Sources.

August 10, 2026
An unhandled error has occurred. Reload 🗙

Rejoining the server...

Rejoin failed... trying again in seconds.

Failed to rejoin.
Please retry or reload the page.

The session has been paused by the server.

Failed to resume the session.
Please retry or reload the page.