Channels¶
Channels are message pipelines that carry messages between framework components.
What is a Channel?¶
+------------------+ +------------------+
| Producer | | Consumer |
| (writes messages)|--->[ Channel ]---->|(reads messages) |
+------------------+ +------------------+
Channel = async queue that connects producers to consumers
Channels are:
- Pipelines - messages flow through them
- Async - non-blocking writes and reads
- Buffered - can hold messages temporarily
- Typed - each channel carries specific message type
Four Main Channels¶
Adapter
+----------------------------------------------------------+
| |
| INBOUND (from PQ) OUTBOUND (to PQ) |
| |
| +------------------+ +------------------+ |
| | DataCommand | | OutboundEvent | |
| | Channel | | Channel | |
| +------------------+ +------------------+ |
| | ^ |
| v | |
| +------------------+ +------------------+ |
| | ControlCommand | | OutboundStatus | |
| | Channel | | Channel | |
| +------------------+ +------------------+ |
| | ^ |
| v | |
| [Command Handlers] [Event Publishers] |
| |
+----------------------------------------------------------+
| Channel | Direction | Carries | Purpose |
|---|---|---|---|
| DataCommandChannel | PQ -> Adapter | Commands | Operational requests (unlock, sync) |
| ControlCommandChannel | PQ -> Adapter | Commands | Admin requests (restart, configure) |
| OutboundEventChannel | Adapter -> PQ | Events | Something happened (access granted) |
| OutboundStatusChannel | Adapter -> PQ | Status | Current state (online, door locked) |
Message Flow¶
Inbound (Commands)¶
PQ System
|
v (NATS message)
+-------------------+
| NatsCommand |
| Receiver |
+-------------------+
|
v (writes to channel)
+-------------------+
| DataCommand |
| Channel |
+-------------------+
|
v (processor reads)
+-------------------+
| DataCommand |
| Processor |
+-------------------+
|
v (calls handler)
+-------------------+
| Your Command |
| Handler |
+-------------------+
Outbound (Events)¶
Your Code
|
| PublishEvent()
v
+-------------------+
| OutboundEvent |
| Channel |
+-------------------+
|
v (publisher reads)
+-------------------+
| Event |
| Publisher |
+-------------------+
|
v (NATS message)
PQ System
Why Separate Channels?¶
Data vs Control Commands¶
+------------------+ +------------------+
| DataCommand | | ControlCommand |
| Channel | | Channel |
+------------------+ +------------------+
| |
v v
Normal priority High priority
(unlock, sync) (restart, stop)
Control commands can interrupt/override data commands.
Events vs Status¶
+------------------+ +------------------+
| OutboundEvent | | OutboundStatus |
| Channel | | Channel |
+------------------+ +------------------+
| |
v v
High volume Low volume
Critical delivery Latest-only OK
(access.granted) (heartbeat)
Events must all be delivered. Status only needs latest value.
Channel Capacity¶
Channels have limited buffer:
+--------------------------------------------------+
| Channel |
| [msg][msg][msg][msg][msg][msg][msg][msg] |
| ^ ^ |
| | | |
| Write pointer Read pointer |
+--------------------------------------------------+
If buffer fills up:
- Writer waits (backpressure)
- Or message dropped (overflow)
Configure depth warning:
var eventChannel = new OutboundEventChannel(
depthWarningThreshold: 1000, // warn if queue > 1000 messages
logger: logger);
Using Channels in Your Code¶
Publishing Events¶
public class Protocol
{
private readonly IAdapterCommunication _communication;
public async Task OnDeviceEvent(DeviceAccessLog log, CancellationToken ct)
{
// create event message
var eventMessage = new EventMessage(
eventType: "access.granted",
severity: "info",
adapterId: _adapterId,
nodeId: doorId,
data: TransformToJson(log));
// publish to channel (framework handles NATS)
await _communication.PublishEventAsync(eventMessage, ct);
}
}
Publishing Status¶
public async Task PublishHeartbeat(CancellationToken ct)
{
var status = new StatusMessage(
adapterId: _adapterId,
nodeId: Guid.Empty, // adapter-level
data: JsonSerializer.SerializeToElement(new {
connection = "online",
deviceCount = _connectedDevices.Count
}));
await _communication.PublishStatusAsync(status, ct);
}
Channel Registration¶
Channels are registered in Program.cs:
var builder = Host.CreateApplicationBuilder(args);
// register channels
builder.Services.AddSingleton<DataCommandChannel>(sp =>
new DataCommandChannel(
depthWarningThreshold: 100,
logger: sp.GetRequiredService<ILogger<DataCommandChannel>>()));
builder.Services.AddSingleton<ControlCommandChannel>(sp =>
new ControlCommandChannel(
depthWarningThreshold: 50,
logger: sp.GetRequiredService<ILogger<ControlCommandChannel>>()));
builder.Services.AddSingleton<OutboundEventChannel>(sp =>
new OutboundEventChannel(
depthWarningThreshold: 1000,
logger: sp.GetRequiredService<ILogger<OutboundEventChannel>>()));
builder.Services.AddSingleton<OutboundStatusChannel>(sp =>
new OutboundStatusChannel(
depthWarningThreshold: 500,
logger: sp.GetRequiredService<ILogger<OutboundStatusChannel>>()));
// register orchestrator
builder.Services.AddSingleton<IAdapterCommunication, NatsCommunication>();
IAdapterCommunication¶
Interface that wraps all channels:
public interface IAdapterCommunication
{
// events
Task PublishEventAsync(EventMessage message, CancellationToken ct);
// status
Task PublishStatusAsync(StatusMessage message, CancellationToken ct);
// lifecycle
Task Start(Guid adapterId, string domain, ICommandDispatcher dispatcher, CancellationToken ct);
Task Stop();
}
Your code uses IAdapterCommunication, not channels directly.
Command Processing¶
Commands flow through processor:
+-------------------+
| DataCommand |
| Channel |
+-------------------+
|
v
+-------------------+
| DataCommand |
| Processor |
+-------------------+
|
| for each command:
v
+-------------------+
| CommandExecutor | <-- your handler
| (delegate) |
+-------------------+
|
v
+-------------------+
| ResponseMessage |
+-------------------+
|
v
+-------------------+
| Return via NATS |
+-------------------+
Register processor:
builder.Services.AddSingleton<DataCommandProcessor>(sp =>
{
var channel = sp.GetRequiredService<DataCommandChannel>();
var protocol = sp.GetRequiredService<Protocol>();
// executor is your handler
CommandExecutor executor = (command, token) =>
protocol.ExecuteCommandAsync(command, token);
return new DataCommandProcessor(channel, executor, logger);
});
Graceful Shutdown¶
On shutdown, channels drain:
Stop signal received
|
v
+-------------------+
| Stop accepting |
| new messages |
+-------------------+
|
v
+-------------------+
| Process remaining |
| messages in queue |
+-------------------+
|
| (up to 30 second timeout)
v
+-------------------+
| Force close |
+-------------------+
Events especially need time to flush - don't lose audit data.
Backpressure¶
When channel fills up:
[Producer] --write--> [FULL Channel] --X-- blocked
Producer waits until:
1. Consumer reads some messages (space available)
2. Timeout expires
3. Cancellation requested
// publish can block if channel full
try
{
await _communication.PublishEventAsync(event, ct);
}
catch (OperationCanceledException)
{
_logger.LogWarning("Event publish cancelled (channel full or shutdown)");
}
Channel Depth Monitoring¶
Monitor for problems:
Normal: [msg][msg][ ][ ][ ] depth: 2
Warning: [msg][msg][msg][msg][msg] depth: 5 (threshold hit)
Critical: [msg][msg][msg][msg][msg]... depth: 100+ (backlog)
When depth warning triggers:
- Something is slow (consumer not keeping up)
- Or burst of events (temporary spike)
Check:
- Is protocol blocking?
- Is NATS connection slow?
- Is device flooding events?
Common Mistakes¶
1. Not Using IAdapterCommunication¶
// WRONG - accessing channels directly
_eventChannel.Writer.TryWrite(message);
// RIGHT - use interface
await _communication.PublishEventAsync(message, ct);
2. Fire and Forget¶
// WRONG - ignoring result
_communication.PublishEventAsync(message, ct); // no await!
// RIGHT - await to handle backpressure
await _communication.PublishEventAsync(message, ct);
3. Not Handling Cancellation¶
// WRONG - ignores shutdown
while (true)
{
await PublishHeartbeat();
await Task.Delay(30000); // ignores ct
}
// RIGHT - respects cancellation
while (!ct.IsCancellationRequested)
{
await PublishHeartbeat(ct);
await Task.Delay(30000, ct);
}
Summary¶
+--------------------------------------------------------------+
| Adapter |
| |
| PQ --> [DataCommandChannel] --> [Processor] --> Handler |
| PQ --> [ControlCommandChannel] --> [Processor] --> Handler |
| |
| Handler --> [OutboundEventChannel] --> [Publisher] --> PQ |
| Handler --> [OutboundStatusChannel] --> [Publisher] --> PQ |
| |
+--------------------------------------------------------------+
Channels are the plumbing. You use IAdapterCommunication to interact with them.
Done with Concepts!
Now you understand the building blocks. Continue to:
- Hello World Tutorial - build your first adapter
- Access Synchronization Tutorial - sync credentials to devices