End-to-End Event Flow Walkthrough

Complete walkthrough showing how device events flow from raw bytes to NATS messages.

Scenario: User swipes card at reader → access granted → door strike releases → door opens → all events published to PQ system.

What you'll learn:

• Complete event processing pipeline
• Address resolution
• State changes vs events
• Thing function methods
• Debugging techniques

Time: ~20 minutes reading + implementation

Architecture Overview

┌─────────────┐     ┌──────────────┐     ┌────────────┐     ┌──────────┐
│   Device    │────▶│  Transport   │────▶│  Protocol  │────▶│  Thing   │
│  (Hardware) │     │  (TCP/Serial)│     │  (Parser)  │     │ (Logic)  │
└─────────────┘     └──────────────┘     └────────────┘     └──────────┘
     Raw bytes         Connection          Event objects      State mgmt
                       management          + routing          + publishing

                                                                     │
                                                                     ▼
                                                            ┌─────────────────┐
                                                            │  NATS Message   │
                                                            │      Bus        │
                                                            └─────────────────┘
                                                                     │
                                                                     ▼
                                                            ┌─────────────────┐
                                                            │   PQ Server     │
                                                            │   (Storage)     │
                                                            └─────────────────┘

Step 1: Device Sends Raw Bytes

Device: Access controller detects card swipe

Device protocol frame (example binary):

Byte:  0x02 0x15 0x00 0x05 0x01 0x12 0x34 0x56 0x78 ... 0x03 0xAB
       STX  Len  Type DevId CardNum(4)         ...     ETX  CRC

Type 0x05 = Access Granted event DevId 0x01 = Door #1 CardNum = 0x12345678

Step 2: Transport Receives Bytes

Transport layer handles physical communication (TCP, or a custom transport for serial/UDP/vendor SDKs).

Transport.cs

public sealed class Transport : IDisposable
{
    private readonly TcpClient _client;
    private readonly Channel<DeviceEvent> _eventChannel;

    public async Task MessagePumpLoop(CancellationToken ct)
    {
        var stream = _client.GetStream();
        var buffer = new byte[4096];

        while (!ct.IsCancellationRequested)
        {
            // read from network
            int bytesRead = await stream.ReadAsync(buffer, ct);

            if (bytesRead == 0)
            {
                // connection closed
                await HandleDisconnect(ct);
                break;
            }

            // deserialize to event object
            var evt = DeserializeEvent(buffer.AsSpan(0, bytesRead));

            if (evt != null)
            {
                // push to channel for protocol processing
                await _eventChannel.Writer.WriteAsync(evt, ct);
            }
        }
    }

    private DeviceEvent? DeserializeEvent(ReadOnlySpan<byte> data)
    {
        // validate frame (STX, length, CRC, ETX)
        if (!ValidateFrame(data))
            return null;

        // parse header
        byte eventType = data[2];
        byte deviceId = data[3];
        var receivedAt = DeviceTimestamp.FromDateTime(timeProvider.GetUtcNow().UtcDateTime);

        return eventType switch
        {
            0x05 => new AccessGrantedEvent
            {
                Timestamp = receivedAt,
                DeviceId = deviceId,
                CardNumber = ReadCardNumber(data),
                EventType = DeviceEventType.AccessGranted
            },
            0x06 => new DoorOpenedEvent
            {
                Timestamp = receivedAt,
                DeviceId = deviceId,
                EventType = DeviceEventType.DoorOpened
            },
            // ... other event types
            _ => new UnknownEvent
            {
                Timestamp = receivedAt,
                DeviceId = deviceId,
                RawData = data.ToArray()
            }
        };
    }

    private static uint ReadCardNumber(ReadOnlySpan<byte> data)
    {
        // read 4 bytes starting at offset 4
        return BinaryPrimitives.ReadUInt32BigEndian(data.Slice(4, 4));
    }
}

Key points:

• Infinite loop reading from network
• Frame validation (CRC, length, delimiters)
• Deserialization to typed event objects
• Channel writer for async processing

Step 3: Protocol Routes Event

Protocol layer receives events from channel and routes to appropriate Thing.

Protocol.cs

public sealed class Protocol : ProtocolBase
{
    private readonly ILogger<Protocol> _logger;
    private readonly Transport _transport;

    protected override async Task HandleEventLoop(CancellationToken ct)
    {
        // read from transport channel
        await foreach (var evt in _transport.EventChannel.ReadAllAsync(ct))
        {
            _logger.LogDebug(
                "Received event: Type={Type}, DeviceId={DeviceId}",
                evt.EventType,
                evt.DeviceId
            );

            try
            {
                await RouteEvent(evt, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(
                    ex,
                    "Failed to route event: Type={Type}, DeviceId={DeviceId}",
                    evt.EventType,
                    evt.DeviceId
                );
            }
        }
    }

    private async Task RouteEvent(DeviceEvent evt, CancellationToken ct)
    {
        switch (evt)
        {
            case AccessGrantedEvent accessEvt:
                await HandleAccessGranted(accessEvt, ct);
                break;

            case AccessDeniedEvent deniedEvt:
                await HandleAccessDenied(deniedEvt, ct);
                break;

            case DoorOpenedEvent doorEvt:
                await HandleDoorOpened(doorEvt, ct);
                break;

            case DoorClosedEvent doorEvt:
                await HandleDoorClosed(doorEvt, ct);
                break;

            case UnknownEvent unknownEvt:
                await HandleUnknown(unknownEvt, ct);
                break;

            default:
                _logger.LogWarning(
                    "Unhandled event type: {Type}",
                    evt.GetType().Name
                );
                break;
        }
    }
}

Key points:

• Event loop reading from channel
• Type-based routing to specific handlers
• Error isolation per event
• Logging for debugging

Step 4: Address Resolution

Find Thing instance by device address.

Protocol.cs (continued)

private async Task HandleAccessGranted(AccessGrantedEvent evt, CancellationToken ct)
{
    // resolve device ID to Thing
    var reader = await FindReaderByAddress(evt.DeviceId, ct);

    if (reader == null)
    {
        _logger.LogWarning(
            "No reader found for address {Address}",
            evt.DeviceId
        );
        return;
    }

    // lookup credential in local cache or PQ API
    var (personId, credentialId) = await ResolveCredential(evt.CardNumber, ct);

    // Reader publishes its own event (Thing must publish, not Protocol)
    await reader.PublishAccessGranted(
        personId,
        credentialId,
        evt.CardNumber,
        evt.Timestamp,
        ct
    );

    _logger.LogInformation(
        "Access granted: Reader={ReaderId}, Person={PersonId}, Card={CardNumber}",
        reader.ThingId,
        personId,
        evt.CardNumber
    );

    // find associated door and unsecure it
    var door = await FindDoorForReader(reader.ThingId, ct);
    if (door != null)
    {
        // state change: secured → unsecured
        await door.Unsecured(evt.Timestamp);

        _logger.LogInformation(
            "Door unsecured: DoorId={DoorId}",
            door.ThingId
        );
    }
}

private async Task<ReaderThing?> FindReaderByAddress(byte deviceId, CancellationToken ct)
{
    // query registry for Thing with matching address
    return await ThingRegistry.FindByAddress<ReaderThing>(deviceId, ct);
}

private async Task<(string? personId, string? credentialId)> ResolveCredential(
    uint cardNumber,
    CancellationToken ct)
{
    // check local credential cache
    if (_credentialCache.TryGetValue(cardNumber, out var cached))
    {
        return (cached.PersonId, cached.CredentialId);
    }

    // query PQ API
    var credential = await _pqApi.GetCredentialByNumber(cardNumber, ct);

    if (credential != null)
    {
        // cache for future lookups
        _credentialCache[cardNumber] = credential;
        return (credential.PersonId, credential.Id);
    }

    return (null, null);
}

Key points:

• Address-based Thing lookup
• Credential resolution (cache + API)
• Reader publishes its own event via public method
• Door uses function method (state change)

Step 5: Thing Processes Event

Thing applies state change and publishes events.

ReaderThing.cs

public class ReaderThing : ThingBase
{
    public ReaderFunction Reader { get; }

    public ReaderThing(string thingId, IServiceProvider services)
        : base(thingId, services)
    {
        Reader = new ReaderFunction(this, new ReaderConfig());
    }

    /// <summary>
    /// publishes access granted event
    /// IMPORTANT: Thing must publish its own events (not Protocol)
    /// </summary>
    public async Task PublishAccessGranted(
        string? personId,
        string? credentialId,
        uint cardNumber,
        DeviceTimestamp timestamp,
        CancellationToken ct)
    {
        // PublishEvent is protected - only Thing can call it
        // This ensures status tracking and internal state remain consistent
        await PublishEvent(
            PqEvent.Access.Granted
                .At(timestamp, this)
                .WithIdentity(personId, credentialId)
                .WithParameter("card_number", cardNumber)
                .Build(),
            ct
        );
    }

    /// <summary>
    /// publishes access denied event
    /// </summary>
    public async Task PublishAccessDenied(
        string reason,
        DeviceTimestamp timestamp,
        CancellationToken ct)
    {
        await PublishEvent(
            PqEvent.Access.Denied
                .At(timestamp, this)
                .WithReason(reason)
                .Build(),
            ct
        );
    }
}

Key pattern:

• Thing has public methods for event publishing
• Protocol calls these public methods
• Inside, Thing calls protected PublishEvent
• This keeps Thing's status and state tracking correct

DoorThing.cs

public class DoorThing : ThingBase
{
    private readonly IDoorFunction _doorFunc;
    private readonly ILogger<DoorThing> _logger;

    public DoorFunction Door { get; }
    public PowerFunction Power { get; }

    public DoorThing(string thingId, IServiceProvider services, ILogger<DoorThing> logger)
        : base(thingId, services)
    {
        _logger = logger;

        Door = new DoorFunction(this, new DoorConfig
        {
            HasPositionSensor = true,
            HasLockSensor = false,
            HasRexSensor = true,
            DoorlongopenTimeoutSeconds = 30,
            AutoRelockTimeoutSeconds = 5
        });

        Power = new PowerFunction(this, new PowerConfig());

        _doorFunc = (IDoorFunction)Door;
    }

    // called by Protocol when access granted
    public async Task Unsecure(DeviceTimestamp timestamp, CancellationToken ct)
    {
        // inside Thing, use direct method call
        await Unsecured(timestamp);

        // function method automatically:
        // 1. updates internal state: secured_closed → unsecured_closed
        // 2. publishes event: pq.event.access.door.unsecured
        // 3. updates Thing status for UI

        _logger.LogInformation(
            "Door unsecured: {ThingId}, State={State}",
            ThingId,
            _doorFunc.CurrentState
        );

        // schedule auto-relock timer
        ScheduleAutoRelock(timestamp, ct);
    }

    private void ScheduleAutoRelock(DateTimeOffset grantTime, CancellationToken ct)
    {
        _ = Task.Run(async () =>
        {
            await Task.Delay(
                TimeSpan.FromSeconds(_doorFunc.AutoRelockTimeoutSeconds),
                ct
            );

            // re-secure strike after timeout; this is adapter-generated, not device-reported
            var relockedAt = DeviceTimestamp.FromDateTime(timeProvider.GetUtcNow().UtcDateTime);
            await Secured(relockedAt);

            _logger.LogInformation(
                "Door auto-relocked: {ThingId}",
                ThingId
            );
        }, ct);
    }
}

Three things happen inside Door.Unsecured():

1. State change: pq.state.access.secured_closed → pq.state.access.unsecured_closed
2. Event publish: pq.event.access.door.unsecured to NATS
3. Status update: UI shows "Door Unsecured" badge

Step 6: Door Position Sensor Triggers

User pushes door open → contact sensor opens → device sends event.

Protocol.cs (door opened handler)

private async Task HandleDoorOpened(DoorOpenedEvent evt, CancellationToken ct)
{
    var door = await FindDoorByAddress(evt.DeviceId, ct);

    if (door == null)
    {
        _logger.LogWarning(
            "No door found for address {Address}",
            evt.DeviceId
        );
        return;
    }

    // position change: door leaf opened
    await door.Opened(evt.Timestamp);

    _logger.LogInformation(
        "Door opened: DoorId={DoorId}, State={State}",
        door.ThingId,
        door.DoorFunction.CurrentState
    );
}

What happens:

1. State change: unsecured_closed → unsecured_open
2. Event publish: pq.event.access.door.opened
3. Status: UI shows "Door Open"

Step 7: Complete Flow Summary

Timeline

t=0.0s: Card swipe detected
  ├─ Device → bytes [0x02 0x15 0x00 0x05 ...]
  └─ Transport → AccessGrantedEvent

t=0.1s: Protocol routes event
  ├─ Address resolution (deviceId=1 → reader-main-entry)
  ├─ Credential lookup (card=12345678 → person-uuid)
  └─ Publish: pq.event.access.granted

t=0.2s: Door unsecures
  ├─ Protocol → door.Door.Unsecured()
  ├─ State: secured_closed → unsecured_closed
  ├─ Publish: pq.event.access.door.unsecured
  └─ Schedule auto-relock (5s timer)

t=0.5s: User pushes door
  ├─ Device → bytes [0x02 0x15 0x00 0x06 ...]
  ├─ Transport → DoorOpenedEvent
  └─ Protocol → door.Door.Opened()

t=0.6s: Door position change
  ├─ State: unsecured_closed → unsecured_open
  ├─ Publish: pq.event.access.door.opened
  └─ Status: "Door Open"

t=2.0s: Door closes
  ├─ Device → bytes [0x02 0x15 0x00 0x07 ...]
  ├─ State: unsecured_open → unsecured_closed
  └─ Publish: pq.event.access.door.closed

t=5.2s: Auto-relock timer fires
  ├─ door.Door.Secured()
  ├─ State: unsecured_closed → secured_closed
  ├─ Publish: pq.event.access.door.secured
  └─ Status: "Door Secured"

Events Published

Time	Event	Thing	Identity	Properties
0.1s	pq.event.access.granted	reader-main-entry	person-uuid, cred-uuid	card_number: 12345678
0.2s	pq.event.access.door.unsecured	door-main-entry	-	-
0.6s	pq.event.access.door.opened	door-main-entry	-	-
2.0s	pq.event.access.door.closed	door-main-entry	-	-
5.2s	pq.event.access.door.secured	door-main-entry	-	-

State Transitions

Door state machine:

secured_closed (t=0)
  ↓ Unsecured()
unsecured_closed (t=0.2s)
  ↓ Opened()
unsecured_open (t=0.6s)
  ↓ Closed()
unsecured_closed (t=2.0s)
  ↓ Secured() [auto-relock]
secured_closed (t=5.2s)

Debugging Techniques

1. Log Raw Bytes

private DeviceEvent? DeserializeEvent(ReadOnlySpan<byte> data)
{
    _logger.LogTrace(
        "Raw frame: {Hex}",
        Convert.ToHexString(data)
    );

    // ... deserialization
}

Output:

TRACE: Raw frame: 021500050112345678...03AB

2. Log Event Objects

private async Task RouteEvent(DeviceEvent evt, CancellationToken ct)
{
    _logger.LogDebug(
        "Event: {@Event}",
        evt  // structured logging
    );

    // ... routing
}

Output:

{
  "EventType": "AccessGranted",
  "DeviceId": 1,
  "CardNumber": 12345678,
  "Timestamp": "2025-01-15T10:30:00Z"
}

3. Log Address Resolution

private async Task<DoorThing?> FindDoorByAddress(byte deviceId, CancellationToken ct)
{
    var door = await ThingRegistry.FindByAddress<DoorThing>(deviceId, ct);

    if (door == null)
    {
        _logger.LogWarning(
            "Address resolution failed: DeviceId={DeviceId}, Type={Type}",
            deviceId,
            nameof(DoorThing)
        );
    }
    else
    {
        _logger.LogDebug(
            "Address resolved: DeviceId={DeviceId} → ThingId={ThingId}",
            deviceId,
            door.ThingId
        );
    }

    return door;
}

4. Log State Changes

public class DoorFunction
{
    private DoorState _state;

    public async Task SetState(DoorState newState, CancellationToken ct)
    {
        var oldState = _state;
        _state = newState;

        _logger.LogInformation(
            "State transition: {ThingId} {OldState} → {NewState}",
            Thing.ThingId,
            oldState,
            newState
        );

        // ... event publishing
    }
}

5. Trace NATS Messages

Use NATS CLI to monitor message bus:

# subscribe to all events
nats sub "pq.event.>"

# filter to door events only
nats sub "pq.event.access.door.>"

# subscribe with verbose output
nats sub "pq.event.>" --verbose

6. Breakpoint Locations

VS Code / Rider / Visual Studio:

Transport layer:

• MessagePumpLoop() - after ReadAsync() (raw bytes received)
• DeserializeEvent() - after parsing (event object created)

Protocol layer:

• RouteEvent() - start of routing logic
• Handler methods (HandleAccessGranted(), etc.) - specific event processing
• Address resolution (FindByAddress()) - Thing lookup

Thing layer:

• Function methods (Door.Unsecured(), etc.) - state change execution
• PublishEvent() - event publishing

Common Failure Points

1. Frame Validation Fails

Symptom: Events never reach Protocol

Cause: CRC mismatch, wrong frame format

Solution:

if (!ValidateFrame(data))
{
    _logger.LogError(
        "Frame validation failed: {Hex}",
        Convert.ToHexString(data)
    );
    return null;
}

2. Address Resolution Fails

Symptom: "No door found for address X"

Cause: Device address doesn't match Thing address property

Solution: Verify adapter-registration.yaml address property matches device hardware address:

properties:
  device_address:
    type: int
    required: true

Check Thing creation:

var door = new DoorThing(
    thingId: "door-1",
    address: 5  // must match device
);

3. Event Not Published

Symptom: State changes but no NATS message

Cause: Wrong PqEvent type or missing Build() call

Solution:

// ❌ Wrong - missing Build()
await PublishEvent(
    PqEvent.Access.Granted.At(timestamp, thingId),
    ct
);

// ✅ Correct
await PublishEvent(
    PqEvent.Access.Granted
        .At(timestamp, thingId)
        .Build(),
    ct
);

4. State Machine Stuck

Symptom: State doesn't transition

Cause: Missing function method call or wrong target state

Solution: Always use function methods for state changes:

// ❌ Wrong - manual state manipulation
door.State = DoorState.Unsecured;

// ✅ Correct - function method
await door.Unsecured(timestamp);

5. Credential Lookup Fails

Symptom: personId and credentialId are null

Cause: Credential not synced to adapter

Solution: Implement AccessSynchronization function and sync credentials first:

// check cache
if (!_credentialCache.ContainsKey(cardNumber))
{
    _logger.LogWarning(
        "Unknown credential: CardNumber={CardNumber}",
        cardNumber
    );

    // publish unknown credential event
    await PublishEvent(
        PqEvent.Access.Credential.Unknown
            .At(timestamp, reader)
            .WithParameter("card_number", cardNumber)
            .Build(),
        ct
    );
}

Performance Considerations

1. Channel Buffering

Configure channel capacity for burst handling:

_eventChannel = Channel.CreateBounded<DeviceEvent>(new BoundedChannelOptions(1000)
{
    FullMode = BoundedChannelFullMode.Wait
});

2. Async Processing

Avoid blocking in event handlers:

// ❌ Blocking - delays other events
Thread.Sleep(1000);

// ✅ Async - allows concurrent processing
await Task.Delay(1000, ct);

3. Credential Cache

Cache frequently-used credentials:

private readonly MemoryCache _credentialCache = new(new MemoryCacheOptions
{
    SizeLimit = 10000,  // max entries
    ExpirationScanFrequency = TimeSpan.FromMinutes(5)
});

// cache with sliding expiration
_credentialCache.Set(cardNumber, credential, new MemoryCacheEntryOptions
{
    Size = 1,
    SlidingExpiration = TimeSpan.FromHours(1)
});

See Also

• Concepts:
• Events - Event taxonomy and structure
• Protocol - Protocol layer patterns

• Things - Thing lifecycle and management

• Patterns:

• Event Translation - Mapping device events
• Address Resolution - Finding Things

• Event Publishing - Function methods vs PqEvent

• Reference:

• Generated Code API - PqEvent builders and shortcuts

• Functions Registry - All function methods

• Tutorials:

• Hello World - Minimal adapter
• YAML Configuration - Complete registration