Synchronous request-response

The flagship feature of SyMqtt is its ability to execute synchronous, bidirectional data exchanges over inherently asynchronous MQTT infrastructures. By pairing an outbound topic and an inbound response topic into a unified SyMqttChannel, you can issue requests and suspend execution until the matching response returns.

---

The core execution mechanism

When you initiate a request through a channel, SyMqtt coordinates several components seamlessly behind the scenes:

1. Task completion promise: The channel wraps the request inside a SyMqttRequestAwaiter and creates a .NET TaskCompletionSource.
2. Asynchronous thread release: The internal MQTT receive loop is configured with TaskCreationOptions.RunContinuationsAsynchronously. This critical optimization ensures that processing subsequent incoming network messages from the broker is never choked or blocked while awaiting user-side logic execution.
3. Thread suspension: The calling thread yields using await, hanging safely until the target device responds, a manual cancellation token is triggered, or a strict timeout window expires.

Method signature

The orchestrator exposes the FetchAsync method to manage this coordinated lifecycle:

public Task<GenericResult<SerialMessage>> FetchAsync(
    string channelId,
    SerialMessage serialMessage,
    int timeoutMilliseconds,
    SyMqttSendMessageOptions messageOptions,
    SyMqttDeliveryOptions deliveryOptions = null,
    CancellationToken cancellationToken = default
)

Request-Response sequence diagram

sequenceDiagram
    autonumber
    actor App as Calling Thread (User Code)
    participant Channel as SyMqttChannel
    participant Sender as SyMqttSenderTopicHolder
    participant Receiver as SyMqttReceiverTopicHolder
    actor Broker as MQTT Broker

    App->>Channel: FetchAsync(requestMessage)
    Note over Channel: 1. Maps MessageId in tracking dict.<br/>2. Creates TaskCompletionSource promise.
    
    Channel->>Sender: Pass message to sender
    Sender->>Broker: Publish raw payload ( on topic "factory/floor1/cmd" )
    
    Note over App: Thread yields safely via await
    Channel-->>App: Returns uncompleted Task promise

    Note over Receiver, Broker: ... Remote controller processes the operation ...

    Broker->>Receiver: Incoming raw payload ( on topic "factory/floor1/res" )
    
    Receiver->>Channel: Route payload to parent Channel
    Note over Channel: 1. SerialMessage.Parse(payload)<br/>2. Extracts tracking MessageId.<br/>3. TryRemove() from dictionary.
    
    Channel-->>App: TaskCompletionSource.SetResult(GenericResult)<br/>(Unblocks the calling user thread)

Step-by-step implementation

Here is the complete workflow to register a channel and perform a synchronous request-response call:

using System;
using System.Threading;
using System.Threading.Tasks;
using SyMqtt;

// 1. Instantiate the channel (passing the parent box instance as the first parameter)
var controlChannel = new SyMqttChannel(
    mqttBox,
    channelId: "conveyor_belt_1",
    senderTopicHolder: new SyMqttSenderTopicHolder("conveyor_cmd", "factory/floor1/conveyor1/cmd"),
    receiverTopicHolder: new SyMqttReceiverTopicHolder("conveyor_res", "factory/floor1/conveyor1/res")
);

// 2. Register the channel into the box prior to opening the connection
mqttBox.AddChannel(controlChannel);

// ... Initialize and Open Connection ...

// 3. Create the outbound payload using the factory method to generate a clean tracking ID
var requestMessage = SerialMessage.NewMessage("SET_SPEED:150");

// 4. Execute the call with a definitive timeout and an optional cancellation token
int timeoutMs = 3000;
var rpcResult = await mqttBox.FetchAsync(
    channelId: "conveyor_belt_1",
    serialMessage: requestMessage,
    timeoutMilliseconds: timeoutMs,
    messageOptions: null,
    deliveryOptions: null,
    cancellationToken: CancellationToken.None
);

// 5. Evaluate the combined results safely
if (rpcResult.Success)
{
    SerialMessage response = rpcResult.Value;
    Console.WriteLine($"Command executed successfully! Response: {response.MessageString}");
}
else
{
    // Evaluate failure (timeout, network drop or manual token cancellation)
    Console.WriteLine($"RPC Communication failed: {rpcResult.ErrorDescription}");
}

---

Advanced execution and delivery options

Just like asynchronous publishing, you can fine-tune your synchronous FetchAsync calls by passing configuration objects.

1. SyMqttSendMessageOptions

Used to alter application-layer routing parameters for the outbound request leg.

Property	Type	Default Value	Description
TargetTopicSuffix	string	null	A unique string appended automatically as a trailing postfix (delimited by /) to the channel’s request topic, allowing you to target a specific hardware node dynamically.

2. SyMqttDeliveryOptions

Controls transport-layer flags. Properties marked for MQTT v5 require initialization with SyMqttProtocolVersion.Mqtt500.

Property	Type	Default Value	Description
QualityOfServiceLevel	int	0	MQTT QoS level (0, 1, or 2) utilized during the request transmission leg.
Retain	bool	false	When true, instructs the broker to cache the request message.
MessageExpiryIntervalInSeconds	uint?	null	MQTT v5 only. Lifetime of the request message in seconds inside the broker before it expires.
SubscriptionIdentifier	uint?	null	MQTT v5 only. An identifier associated with the client’s subscription tracking.
TopicAlias	ushort?	null	MQTT v5 only. A numeric alias used to optimize network bandwidth for the long topic string.

Fallback Mechanism: If you pass null for the deliveryOptions parameter, FetchAsync automatically falls back to the values defined in the global DefaultDeliveryOptions property of your SyMqttBox instance.

---

Targeted unicast routing in channels

In large-scale deployments, you often have multiple identical physical controllers listening to the same command infrastructure. To prevent messages from acting as a global broadcast, SyMqtt supports targeted routing via the TargetTopicSuffix property inside SyMqttSendMessageOptions.

When passed to the FetchAsync method, the engine dynamically modifies the outbound leg of the transaction:

1. It appends the TargetTopicSuffix string as a postfix (separated by /) to the channel’s Request Topic path.
2. The target controller, programmed with its own unique ID, intercepts only the messages containing its specific postfix stamp.
3. Important lifecycle note: The return leg (Response Topic) remains bound to the base configuration path, as the tracking correlation token inside the SerialMessage envelope is already handling the unique thread alignment.

This creates a highly scalable point-to-point architecture over a single channel layout template:

using SyMqtt;

var unicastOptions = new SyMqttSendMessageOptions
{
    TargetTopicSuffix = "PLC_Node_22"
};

// If the TopicsPrefix is "smartcontrol", the request is routed strictly to:
// "smartcontrol/factory/floor1/conveyor1/cmd/PLC_Node_22"
// The response will still land on: "smartcontrol/factory/floor1/conveyor1/res"
var result = await mqttBox.FetchAsync(
    channelId: "conveyor_belt_1",
    serialMessage: requestMessage,
    timeoutMilliseconds: 3000,
    messageOptions: unicastOptions
);

---

Extracting response network metadata

When a response returns successfully, SyMqtt captures structural details from the transport layer and injects them directly into the SerialMessage.Metadata property as a SyMqttResponseMessageInfo object.

This allows you to inspect advanced broker states or properties without polluting the core plain-text payload string. You can access properties like:

• PublishTopic: The exact physical topic string the message landed on (useful when tracking wildcard subscriptions).
• Qos: The Quality of Service level utilized by the broker during the return leg.
• UserProperties: A dictionary containing native MQTT v5 User Properties headers, safely aggregated to handle duplicate metadata keys gracefully by keeping the latest value.

Here is how you can read transport metadata from a successful response:

if (rpcResult.Success && rpcResult.Value.Metadata is SyMqttResponseMessageInfo networkInfo)
{
    Console.WriteLine($"Response physical topic: {networkInfo.PublishTopic}");
    Console.WriteLine($"Transport QoS: {networkInfo.Qos}");

    if (networkInfo.UserProperties.TryGetValue("server_timestamp", out string ts))
    {
        Console.WriteLine($"Broker processed timestamp: {ts}");
    }
}

---

Error, timeout, and cancellation resiliency

The FetchAsync routine is completely safe against hanging operations. If a remote controller goes offline, loses power, or drops its network connection midway through an execution sequence:

1. Eviction: The internal dictionary tracking the MessageId drops the request wrapper safely via a TryRemove operation, preventing memory leaks.
2. Timeout Unblocking: If the timer expires before a response is received, the calling thread is unblocked, and a clean GenericResult is returned with an explicit error description (e.g., "Request timed out after 3000 ms").
3. Instant Cancellation: If the passed .NET CancellationToken is cancelled mid-flight, the thread unblocks immediately (without waiting for the remaining timeout duration), cleans up the internal mapping, and returns a dedicated cancellation error status.