Real-time Spike Detection with SR-CNN

To detect sudden changes, I tried an SR‑CNN(Spectral Residual and Convolutional Neural Network) model that looks at a small window of recent values. It assigns an anomaly score to each point, and if the newest score crosses a threshold, the system treats it as a spike.

The sample application uses ML.NET SR‑CNN to processes a rolling buffer (ConcurrentQueue) of recent samples and publishes to an Message Queue Telemetry Transport(MQTT) topic anomalies as they are detected.

private sealed class TopicRuntime
{
   public object Lock { get; } = new();
   public Queue<Model.TimeSeriesData> Buffer { get; }
   public int MaxBufferSize { get; }
   public ITransformer? Transformer { get; set; }

   public TopicRuntime(Model.TopicConfiguration configuration)
   {
      MaxBufferSize = configuration.SrCnnSettings.WindowSize+ configuration.SrCnnSettings.LookaheadWindowSize+ configuration.SrCnnSettings.BackAddWindowSize;

      Buffer = new Queue<Model.TimeSeriesData>(MaxBufferSize);
   }
}

Every MQTT topic gets its own little SR‑CNN workspace: a lock to keep inference thread‑safe, a buffer that holds just enough recent samples for the model to work, and a transformer that runs the spike‑detection pipeline. The buffer size is calculated from the _applicationSettings windowSize, LookaheadWindowSize, and BackAddWindowSize settings so the model always sees the sequence length it has trained for.

var topicEstimator = _TopicEstimators.GetOrAdd(subscribedTopic, key => new TopicRuntime(_applicationSettings.SubscribedTopics[key]));

List<Model.SpikePrediction> spikePredictions;
lock (topicEstimator.Lock)
{
   topicEstimator.Buffer.Enqueue(new Model.TimeSeriesData { Value = value });
   while (topicEstimator.Buffer.Count > topicEstimator.MaxBufferSize) topicEstimator.Buffer.Dequeue();

   if (topicEstimator.Buffer.Count < subscribedTopicSettings.SrCnnSettings.WindowSize)
   {
      Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Not enough data for prediction (have {topicEstimator.Buffer.Count}, need {subscribedTopicSettings.SrCnnSettings.WindowSize})");
      return;
   }

   var data = _mlContext.Data.LoadFromEnumerable(topicEstimator.Buffer);

   if (topicEstimator.Transformer is null)
   {
      var estimator = _mlContext.Transforms.DetectAnomalyBySrCnn(
         outputColumnName: nameof(Model.SpikePrediction.Prediction),
         inputColumnName: nameof(Model.TimeSeriesData.Value),
         windowSize: subscribedTopicSettings.SrCnnSettings.WindowSize,
         backAddWindowSize: subscribedTopicSettings.SrCnnSettings.BackAddWindowSize,
         lookaheadWindowSize: subscribedTopicSettings.SrCnnSettings.LookaheadWindowSize,
         averagingWindowSize: subscribedTopicSettings.SrCnnSettings.AveragingWindowSize,
         judgementWindowSize: subscribedTopicSettings.SrCnnSettings.JudgementWindowSize,
         threshold: subscribedTopicSettings.SrCnnSettings.Threshold);
      topicEstimator.Transformer = estimator.Fit(data);
   }

   var transformed = topicEstimator.Transformer.Transform(data);
   spikePredictions = [.. _mlContext.Data.CreateEnumerable<Model.SpikePrediction>(transformed, reuseRowObject: false)];
}

SR‑CNN needs a warmup period. Early on, it doesn’t have enough history, so its baseline is shaky and it tends to over‑react, flagging lots of spikes. After the model has seen enough samples (WindowSize + LookaheadWindowSize + BackAddWindowSize), it settles down, it understands the normal noise and recognises patterns. But, I have found if it runs for a very long time, it can become almost too stable, making it slower to react to sudden changes.

To keep things flexible, the input transformer uses CS‑Script to turn the message payload (in this example JSON) into a C# object and then pulls out the one value (a float) the SR‑CNN model needs. In this case, it reads the Cm field of the Seeedstudio Ultrasonic Ranger and returns it as a floating‑point number. Each device type (in this example a Seeedstudio SKU 101991042) gets its own tiny script, making the system easy to extend.

//---------------------------------------------------------------------------------
// Copyright (c) May 2026, devMobile Software
//
/*
{
   "ClientID:"Device123",
   "Mm":269,
   "Cm":26.8999996,
   "Temperature":24.2999992
}
*/
using System; // Donot remove this as required for InvalidOperationException
using devMobile.IoT.MqttTransformers;

internal class SKU101991042
{    
   public string ClientID { get; set; } = string.Empty;
   public int Mm { get; set; }
   public float Cm { get; set; }
   public float Temperature { get; set; }
}

public class InputSKU101991042 : IInputMessageTransformer
{
   public float Transform(string topic, byte[] payload)
   {
      var json = System.Text.Encoding.UTF8.GetString(payload);

      var obj = System.Text.Json.JsonSerializer.Deserialize<SKU101991042>(json) ?? throw new InvalidOperationException("Failed to deserialize payload");

      return obj.Cm;
   }
}

The output transformer takes the spike‑detection result and turns it into a simple JSON message that is published to the specified MQTT topic. It builds a data transfer object (DTO) containing the detection type, topic, value, raw score, and magnitude, then serialises it and returns the UTF‑8 bytes. Each CS-Script transformer script defines exactly how the spike result should look, making the system flexible and easy to extend.

//---------------------------------------------------------------------------------
// Copyright (c) May 2026, devMobile Software
//
using System.Text.Json;
using System.Text.Json.Serialization; // Do not remove this using directive as it is required for the JsonIgnoreCondition

using devMobile.IoT.MqttTransformers;


public class SpikeOutputTransformer : ISpikeOutputMessageTransformer
{
   private static readonly JsonSerializerOptions _serializerOptions = new() { DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull };

   public byte[] Transform(string topic, double value, double rawScore, double magnitude)
   {
      var obj = new
      {
         DetectionType = "Spike",
         Topic = topic,
         Value = value,
         RawScore = rawScore,
         Magnitude = magnitude
      };

      string payload = JsonSerializer.Serialize(obj, _serializerOptions);

      return System.Text.Encoding.UTF8.GetBytes(payload);
   }
}

The MQTTX application from EMQX provides a clear view of the messages published by the spike‑detection pipeline. Each alert arrives as a JSON payload containing the detection type, topic name, measured value, raw SR‑CNN score, and spike magnitude. This makes it easy to monitor devices in close to “real time” and verify that spike events are being reliably detected and alerted correctly.

In my testing, SR‑CNN consistently produced more reliable results with my device data than IID or SSA‑based anomaly‑detection approaches. The residual CNN architecture handled the noise characteristics and temporal patterns far better, giving fewer false positives and more stable detections especially across long‑running sessions.

This blog post assumes you have read my earlier spike and change point detection posts.

Mistral Vibe Code Generated Mistral Chat Completion API Client

For this series of posts, I asked GitHub Copilot, Anthropic Claude Code, and Mistral Vibe Code to generate a Mistral Chat Completion client using the same prompt….

Generate C# DTOs and a ChatCompletionClient for https://docs.mistral.ai/api/endpoint/chat. The ChatCompletionClient should
use HttpClientFactory with resilience and should have async methods where suitable. The DTOs should be split up into files
for shared, buffered, and streaming.

Generate a console application which uses ChatCompletionClient

Then I took the initial implementation and prompted GitHub Copilot, Anthropic Claude Code, and Mistral Vibe Code to review them. The aim was to see how each tool thinks, what they prioritise, and then how their feedback differs when given identical code.

TL;DR – the code was fine.

Github Copilot

Overall the library is well-structured, cleanly documented, and follows familiar patterns (Options, IHttpClientFactory, Polly).

Anthropic Claude

The client is a small, focused wrapper around POST /v1/chat/completions, supporting both a buffered call (CompleteAsync) and an SSE streaming call (StreamAsync). Overall design is clean: immutable JsonSerializerOptions, an immutable record request type mutated via with, DI registration through IHttpClientFactory with a resilience pipeline, and polymorphic DTOs for messages/content/tools that mirror the Mistral (OpenAI-compatible) wire format. No correctness-blocking bugs were found; findings below are mostly design nits and a few things worth double-checking against real API behavior.

Mistral Vibe Code

Overall Rating: ⭐⭐⭐⭐☆ (4.2/5) – Production Ready with Minor Improvements

Github Copilot Review

:question:

Review MistralAI.Client.Copilot – ChatCompletionClient.cs and DTOs

GitHub Copilot

  1. Read ChatCompletionClient.cs, lines 165-514 :white_check_mark:
  2. Code search found relevant results :white_check_mark:
  3. Read ChatCompletionClient.cs, lines 294-514 :white_check_mark:
  4. Read BufferedDto.cs, lines 1-220 :white_check_mark:
  5. Read SharedDto.cs, lines 1-305 :white_check_mark:
  6. Read StreamingDto.cs, lines 1-420 :white_check_mark:

Code Review – ChatCompletionClient & DTOs

Overall the library is well-structured, cleanly documented, and follows familiar patterns (Options, IHttpClientFactory, Polly). Below are the concrete issues and improvements I’d action, roughly ordered by impact.

🔴 Correctness / bugs

  1. Streaming isn’t actually streaming. In StreamCompletionAsync you call _httpClient.PostAsync(...), which buffers the whole response before returning. You must pass HttpCompletionOption.ResponseHeadersRead, otherwise every “streamed” token arrives at once (and the retry policy will hold the full body in memory).
using var req = new HttpRequestMessage(HttpMethod.Post, requestUri) { Content = requestContent };
response = await _retryPolicy.ExecuteAsync(ct =&gt;
    _httpClient.SendAsync(req, HttpCompletionOption.ResponseHeadersRead, ct), cancellationToken);
  1. Retry policy + single StringContent is broken. StringContent is disposed once sent, so any retry after the first attempt POSTs an empty/disposed body. The HttpRequestMessage/content must be re-created inside the ExecuteAsync delegate (or use HttpClientHandler-level Polly via AddPolicyHandler, which handles this).

  2. IDisposable + IHttpClientFactory conflict. The class implements IDisposable and holds _disposed, implying it disposes _httpClient. HttpClients from IHttpClientFactory must not be disposed by the consumer — the factory owns the handler lifetime. Either:

    • drop IDisposable, or
    • dispose only in the HttpClient-injected overload and track ownership with a flag.
  3. Mutating a factory-provided HttpClient in the constructor. Setting Timeout, DefaultRequestHeaders.Authorization, and User-Agent on httpClientFactory.CreateClient() re-configures a shared/named client each time and, worse, appending User-Agent via Add(...) will throw InvalidOperationException on second construction against the same named client. Configure this once through services.AddHttpClient(c =&gt; …) and remove it from the ctor.

  4. ReadLineAsync treats SSE as line-oriented text. Real SSE:

    • allows data: (no space) — your Substring(6) is off by one in that case,
    • allows multi-line data: chunks concatenated per event,
    • has comment lines starting with : (heartbeat/keep-alive) that must be ignored,
    • has event:, id:, retry: fields. Consider System.Net.ServerSentEvents.SseParser (built-in) instead of hand-rolling.
  5. Polymorphic streaming events. You deserialize into a single CompletionEvent type, yet StreamingDto.cs defines CompletionEventBase with TextCompletionEvent, ToolCallEvent, … discriminated by type. Without [JsonPolymorphic]/[JsonDerivedType] on the base, the derived data is lost. Either:

    • Add polymorphism attributes on CompletionEventBase and deserialize as the base, or
    • Keep a flat CompletionEvent DTO and drop the base/derived hierarchy — currently you have both, which is confusing.
  6. Fallback path assumes Text exists on CompletionEvent.
    new CompletionEvent { Type = "unknown", Text = line } implies a flat DTO; reconcile with issue #6.

  7. No Retry-After handling on 429. Polly’s WaitAndRetryAsync uses only 2^n. Mistral (like OpenAI) returns Retry-After — honour it.

  8. response.Content.ReadAsStringAsync() / ReadAsStreamAsync() don’t take the cancellationToken. These overloads exist on .NET 8 — pass the token so hanging responses can be cancelled.

  9. HttpClient.Timeout fights with cancellation. With Polly retries + streaming, a global 120 s timeout aborts long streams. Use HttpClient.Timeout = Timeout.InfiniteTimeSpan and enforce per-request timeout via a linked CancellationTokenSource.

🟠 API / design

  1. Massive DTO duplication. ChatCompletionRequest (Buffered) and ChatCompletionStreamRequest (Streaming) are byte-for-byte identical apart from the default of Stream. Introduce a shared base (or single DTO) in DTOs.Shared and delete ~130 lines. The Stream boolean is the only real difference and can be set by the client method.

  2. Stop and ToolChoice typed as object?. These serialize fine but deserialize poorly and lose IntelliSense. Use OneOf&lt;string, List&gt;, a small wrapper type, or JsonElement.

  3. Metadata is Dictionary. Mistral’s schema is Dictionary.

  4. MessageBase.Content is string? only. Mistral supports multimodal content (array of {type, text|image_url} parts) on user messages. Model as object? or OneOf&lt;string, List&gt; to be forward-compatible.

  5. AssistantMessage missing prefix/reasoning/tool_call_id variants. ChatCompletionChoice.Prefix exists but not on the message itself, which is where the API places it for message replay.

  6. GuardrailConfig is an empty placeholder. Either implement it or omit the property so serializers don’t emit "guardrails": [].

  7. ResponseFormat.Schema should be JsonElement/typed to avoid double serialization surprises via object.

  8. MistralAiException — add the three standard Exception ctors ((), (string), (string, Exception)), and consider a strongly typed MistralApiError model (type, message, param, code) parsed from the body instead of the raw string.

  9. Validation. MistralAiOptions.ApiKey defaults to ""; the constructor happily builds a client with no auth. Add IValidateOptions or throw in the ctor.

  10. GetJsonOptions() (not shown) should be cached in a static readonly field — creating JsonSerializerOptions per call is expensive and defeats the internal metadata cache.

  11. CreateRequest / CreateStreamRequest just build DTOs. They add little over new ChatCompletionRequest { … } and inflate the surface area. Consider removing, or make them extension methods.

  12. Second ctor mutates a caller-owned HttpClient. Prefer treating an injected HttpClient as read-only and route auth via HttpRequestMessage.Headers.Authorization on each request.

🟡 Style / minor

  • line.StartsWith("data: ", StringComparison.Ordinal) — add ordinal comparison; ditto for the [DONE] sentinel.
  • Use _httpClient.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("MistralAI.Client", "1.0")) instead of Add("User-Agent", …).
  • Consider sealed on ChatCompletionClient; extension is unlikely and it simplifies IDisposable.
  • MistralAiOptions.EnableStreaming is declared but never used.
  • XML doc for CreateStreamRequest says stream param exists — it doesn’t.
  • Naming: CompletionEvent vs CompletionEventBase vs TextCompletionEvent — pick one convention (drop Base, since it’s abstract).
  • Consider ArgumentNullException.ThrowIfNull(request) (.NET 8 idiom).

Suggested next steps

  1. Fix the streaming/retry correctness issues (#1, #2, #5, #6) — those change behaviour.
  2. Collapse the duplicated request DTOs (#11) — biggest maintenance win.
  3. Move HttpClient configuration into DI registration and drop IDisposable (#3, #4).
  4. Cache JsonSerializerOptions and switch to source-generated serialization if AOT/perf is a goal (#20).

Happy to draft any of these changes — tell me which to tackle first.

The next review was done with Anthropic’s Claude Code

Code Review: MistralAI.Client.CopilotChatCompletionClient.cs & DTOs

Scope: src/Copilot.PoC/MistralAI.Client.Copilot/ Files reviewed:

  • ChatCompletionClient.cs
  • IChatCompletionClient.cs
  • ChatCompletionRequest.cs (request DTOs)
  • ChatCompletionResponse.cs (response DTOs)
  • Streaming/StreamingDTOs.cs
  • MistralClientOptions.cs, MistralServiceCollectionExtensions.cs (for context)

Summary

The client is a small, focused wrapper around POST /v1/chat/completions, supporting both a buffered call (CompleteAsync) and an SSE streaming call (StreamAsync). Overall design is clean: immutable JsonSerializerOptions, an immutable record request type mutated via with, DI registration through IHttpClientFactory with a resilience pipeline, and polymorphic DTOs for messages/content/tools that mirror the Mistral (OpenAI-compatible) wire format. No correctness-blocking bugs were found; findings below are mostly design nits and a few things worth double-checking against real API behavior.


ChatCompletionClient.cs

Strengths

  • JsonOptions is a single static, cached JsonSerializerOptions instance (line 18) — correct, since re-creating JsonSerializerOptions per call is a well-known perf trap with System.Text.Json.
  • CompleteAsync/StreamAsync both null-check request at the boundary (ArgumentNullException.ThrowIfNull), consistent with API-boundary validation.
  • request with { Stream = false/true } cleanly forces the correct value regardless of what the caller set, without mutating the caller’s object.
  • StreamAsync uses HttpCompletionOption.ResponseHeadersRead so the response isn’t buffered before the SSE loop starts — necessary for real streaming.
  • Malformed SSE chunks are swallowed with a comment explaining why (// Skip malformed chunks rather than aborting the whole stream.) — an intentional, documented tradeoff rather than silent-failure-by-accident.
  • EnsureSuccessAsync reads the error body only on failure, and wraps it into HttpRequestException with the status code attached (statusCode: response.StatusCode) — good for callers that pattern-match on StatusCode.

Findings / things to double check

  1. ChatCompletionRequest.Stream is public but always overwritten. The DTO exposes a settable Stream property, documented as "Overridden internally by the client depending on the method called." (ChatCompletionRequest.cs:26). Any value the caller sets is silently discarded by both CompleteAsync and StreamAsync. This isn’t a bug, but it’s a slightly surprising public API — a caller could reasonably expect setting Stream = true and calling CompleteAsync to do something, when it’s actually a no-op. Consider either making the setter internal/removing it from the public request shape, or asserting/ignoring rather than silently stomping it.

  2. Silent chunk-skip on JsonException has no observability hook. StreamAsync (lines 104–113) catches JsonException and continues with no logging or counter. For a PoC this is fine, but if the wire format ever drifts (e.g., a new event type Mistral adds), failures will be invisible — chunks just quietly vanish. Consider at least a ILogger debug-level log if one is easy to thread through, since this is the one place errors are deliberately suppressed rather than propagated.

  3. SSE parsing only recognizes data: lines. The reader loop only reacts to lines starting with data: and otherwise continues (lines 89–92), which correctly skips blank lines and SSE comment/event: lines. This matches Mistral’s OpenAI-compatible SSE format (each event is a single data: {...} line), so it should be fine in practice — just flagging that it assumes single-line JSON payloads per event; a server that ever wrapped a JSON payload across multiple data: lines (some SSE producers do this) would need concatenation logic that isn’t present here.

  4. No cancellation-specific handling. CancellationToken is threaded through correctly (ConfigureAwait(false) + [EnumeratorCancellation]), but there’s no explicit catch/rethrow around OperationCanceledException — that’s actually correct behavior (let it propagate), just noting it was checked.


Request DTOs (ChatCompletionRequest.cs)

Strengths

  • ChatCompletionRequest is a sealed record with required members for Model/Messages, giving compile-time enforcement of the two truly mandatory fields.
  • Polymorphic ChatMessage/ContentPart hierarchies use [JsonPolymorphic] + [JsonDerivedType] correctly keyed off role and type respectively, matching the API’s discriminator fields.
  • MessageContent cleanly models the “string OR array of parts” duality the API allows, with implicit conversions (string, ContentPart[], List) making call sites ergonomic, and a custom JsonConverter handling both read and write shapes.
  • GuardrailConfig.AdditionalProperties via [JsonExtensionData] is a sensible forward-compatibility escape hatch for a field Mistral is likely to extend.

Findings

  1. Naming collision: ToolChoice property vs. ToolChoice static helper class. ChatCompletionRequest.ToolChoice (an object? property, line 41) and the top-level public static class ToolChoice (line 239) share the identifier ToolChoice in the same namespace (Mistral.Client.Shared). This compiles and works — new ChatCompletionRequest { ToolChoice = ToolChoice.Auto } resolves the right-hand side to the static class since object-initializer right-hand expressions aren’t evaluated in the initializer’s member scope — but it’s a readability trap: IntelliSense/go-to-definition on ToolChoice inside that expression can be momentarily ambiguous to a human reader, and a future refactor that moves code around could make this genuinely ambiguous or require global::/qualification. Consider renaming the helper class (e.g. ToolChoiceValues or ToolChoices) to remove the shadow.

  2. ToolChoice property typed as object?. Reasonable given the API accepts either a string enum or an object literal, and the ToolChoice helper class exists specifically to build valid values — but it does mean nothing stops a caller from passing an arbitrary/invalid object that will only fail at the HTTP layer. Given this is a PoC and the alternative (a discriminated-union style type) is meaningfully more code, this looks like an acceptable, deliberate tradeoff rather than an oversight.

  3. MessageContentJsonConverter.Write prefers Text over Parts if both are set. Not reachable through the public implicit-conversion surface (only one of Text/Parts is ever set), but since MessageContent has public init accessors, nothing stops object-initializer code from setting both fields directly, at which point Parts would be silently dropped on serialize. Low risk given current usage patterns, just noting the type doesn’t defensively guard against that state.

  4. Minor: IReadOnlyList has no implicit conversion, only ContentPart[] and List do (lines 122–124). Not a bug — just means a caller holding an IReadOnlyList from elsewhere has to materialize it to one of the two supported types first.


Response DTOs (ChatCompletionResponse.cs)

  • Straightforward, mutable POCOs (get; set;) matching typical System.Text.Json deserialization targets — appropriately different from the request side (which is a record built by the caller) since these are populated by the deserializer, not constructed by consumers.
  • ChatCompletionChoice.Message is typed as AssistantMessage? (line 21), correctly narrowing the polymorphic ChatMessage base to the one concrete type a response choice can actually contain.
  • Choices and other list-typed members default to [] rather than null (e.g., ChatCompletionResponse.cs:12), which avoids null-checks for consumers even before deserialization overwrites them — good defensive default.
  • No SystemFingerprint/LogProbs/similar fields are modeled. If Mistral’s response includes them and callers might want them, they’re currently unreachable (no [JsonExtensionData] catch-all on the response types, unlike GuardrailConfig on the request side). Worth confirming this is an intentional scope decision for the PoC.

Streaming DTOs (Streaming/StreamingDTOs.cs)

  • ChatCompletionChunk/ChatCompletionChoiceChunk/DeltaMessage correctly mirror the non-streaming shapes where structurally identical (UsageInfo, MessageContent, ToolCall are reused rather than duplicated), which is good — no drift risk between streaming/non-streaming content models.
  • The XML doc on DeltaMessage (lines 28–32) correctly documents the aggregation contract callers must implement (accumulate Content, merge ToolCalls by Index) — useful since the client itself provides no aggregation helper, so this is the only place that contract is written down. Worth keeping in sync with any future consumer-side aggregation utility.
  • Minor cosmetic: namespace Mistral.Client.Shared.Streaming; (line 3) has a double space after the namespace keyword. Harmless, but worth a quick formatter pass.

Supporting types (context only)

  • MistralClientOptions / MistralServiceCollectionExtensions: registration wires IChatCompletionClient as a singleton backed by a named HttpClient with Timeout = Timeout.InfiniteTimeSpan and delegates time-boxing to the standard resilience handler (AttemptTimeout 90s, TotalRequestTimeout 5m) — a sensible choice given long-running generations and streaming responses, and the comment explaining why the infinite client timeout is safe here is a good practice (explains a non-obvious decision rather than just stating what the code does).

Overall assessment

No functional bugs found. The code is well-structured for a PoC: immutable request model, cached serializer options, correct streaming/cancellation plumbing, and DTOs that faithfully mirror the API’s polymorphic wire format. The main things worth a follow-up decision are the ToolChoice naming collision (cosmetic but worth a rename) and whether the public, always-overridden Stream property on ChatCompletionRequest is intentional API surface or should be tightened.

The final review was done with Mistral’s Vibe Code

📊 MistralAI.Client.Copilot Code Review

> Project Type: .NET 10.0 Library | Purpose: Mistral AI Chat Completion API Client > Last Reviewed: 2026-07-20 | Reviewed By: Mistral Vibe



📁 Project Structure

MistralAI.Client.Copilot/
├── MistralAI.Client.Copilot.csproj         # Project configuration (net10.0)
├── IChatCompletionClient.cs               # Interface definition
├── ChatCompletionClient.cs                # Main client implementation
├── ChatCompletionRequest.cs               # Request DTOs &amp; Message types
├── ChatCompletionResponse.cs              # Response DTOs
├── MistralClientOptions.cs                # Configuration options
├── MistralServiceCollectionExtensions.cs # DI registration
└── Streaming/
    └── StreamingDTOs.cs                   # Streaming response chunk DTOs

Strengths

1. Clean Architecture

  • Separation of Concerns: Clear separation between interfaces (IChatCompletionClient), implementation (ChatCompletionClient), and DTOs
  • Dependency Injection: Proper use of IHttpClientFactory and DI container extensions
  • Modular Design: Streaming logic isolated in dedicated namespace

2. Modern C# Features

  • Records: Extensive use of record types for immutable DTOs (ChatCompletionRequest, ChatCompletionResponse)
  • Pattern Matching: Effective use in MessageContentJsonConverter.Read()
  • Async Streams: Proper implementation of IAsyncEnumerable for streaming
  • Init-only Properties: Consistent use of init accessors for immutable DTOs
  • File-scoped Namespaces: Clean namespace declarations

3. Robust Error Handling

  • Argument Validation: Uses ArgumentNullException.ThrowIfNull() for public method parameters
  • HTTP Error Handling: EnsureSuccessAsync() provides detailed error messages with status codes and response bodies
  • Streaming Resilience: Gracefully handles malformed JSON chunks in streaming without breaking the entire stream
  • Cancellation Support: Proper CancellationToken propagation throughout async methods

4. API Design

  • Interface-based: Client consumes IChatCompletionClient interface, enabling mocking
  • Flexible Configuration: MistralClientOptions allows custom base URLs and API keys
  • Resilience Pipeline: Built-in retry, circuit breaker, timeout, and rate limiting via AddStandardResilienceHandler

5. JSON Serialization

  • System.Text.Json: Uses modern, high-performance JSON serializer
  • Polymorphic Deserialization: Proper use of [JsonPolymorphic] and [JsonDerivedType] for message role discrimination
  • Custom Converters: MessageContentJsonConverter handles string vs array polymorphism
  • Snake Case Naming: Consistent JSON naming policy matching Mistral API conventions
  • Null Handling: Proper JsonIgnoreCondition.WhenWritingNull configuration

6. Streaming Implementation

  • SSE Parsing: Correctly handles Server-Sent Events format (data: prefix, [DONE] sentinel)
  • Memory Efficient: Uses StreamReader for line-by-line processing
  • Chunk Validation: Skips malformed chunks instead of failing

7. Documentation

  • XML Documentation: Comprehensive XML comments on public APIs
  • Constants for Magic Strings: FinishReason, ToolChoice classes for well-known values

🔍 DTO Analysis

Request DTOs (ChatCompletionRequest.cs)

DTO Purpose Design Quality
ChatCompletionRequest Main request body ✅ Comprehensive, all Mistral API parameters covered
ChatMessage (abstract) Base message type ✅ Polymorphic with role discriminator
SystemMessage, UserMessage, AssistantMessage, ToolMessage Message types ✅ Proper inheritance hierarchy
MessageContent Flexible content (string or parts) ✅ Uses custom converter for polymorphism
ContentPart (abstract) Content part types ✅ Extensible with derived types
TextContentPart, ImageUrlContentPart, etc. Specific content types ✅ Covers all Mistral content types
Tool, FunctionDefinition, ToolCall, FunctionCall Tool integration ✅ Complete tool calling support
ResponseFormat, JsonSchemaSpec Response formatting ✅ JSON schema support
UsageInfo Token usage tracking ✅ Standard structure
GuardrailConfig Guardrail configuration ✅ Extensible with [JsonExtensionData]
Prediction Prediction hints ✅ Simple and effective

Response DTOs

DTO File Design Quality
ChatCompletionResponse ChatCompletionResponse.cs ✅ Matches Mistral API response structure
ChatCompletionChoice ChatCompletionResponse.cs ✅ Includes message and finish reason
ChatCompletionChunk StreamingDTOs.cs ✅ Streaming chunk with usage info
ChatCompletionChoiceChunk StreamingDTOs.cs ✅ Delta-based streaming
DeltaMessage StreamingDTOs.cs ✅ Incremental message content

⚠️ Areas for Improvement

1. Namespace Organization

  • Issue: Inconsistent namespace usage
    • IChatCompletionClient uses Mistral.Client
    • ChatCompletionClient uses Mistral.Client.Shared
    • This creates confusion about which namespace to use
  • 💡 Recommendation: Standardize on Mistral.Client or MistralAI.Client throughout

2. Nullability Annotations

  • ⚠️ Issue: Incomplete nullability annotations
    • Many DTO properties lack ? for nullable reference types
    • Example: ChatCompletionRequest.Model is required string but could benefit from explicit nullability
  • 💡 Recommendation: Add consistent nullability annotations, especially for optional fields

3. Record vs Class Usage

  • ⚠️ Issue: Mixed use of record and class for DTOs
    • ChatCompletionRequest is a record (good for immutability)
    • Message types (SystemMessage, UserMessage) are class with setters
  • 💡 Recommendation: Consider making all DTOs immutable record types for consistency

4. JSON Serialization Edge Cases

  • ⚠️ Issue: MessageContentJsonConverter doesn’t handle all possible JSON scenarios
    • No validation that exactly one of Text or Parts is set
    • Could throw on unexpected JSON structures
  • 💡 Recommendation: Add validation in converter to ensure data integrity

5. Error Handling in Streaming

  • ⚠️ Issue: Silent skipping of malformed chunks
    • Line 111-112: continue on JsonException
    • No logging or telemetry for skipped chunks
  • 💡 Recommendation: Add optional logging/delegate for error handling

6. Configuration Validation

  • ⚠️ Issue: Minimal validation in MistralClientOptions
    • No validation that ApiKey is not empty when provided
    • No validation that BaseAddress ends with trailing slash (as documented)
  • 💡 Recommendation: Add validation in property setters or use validation library

7. HTTP Client Configuration

  • ⚠️ Issue: Hardcoded timeout values in resilience pipeline
    • 90s attempt timeout, 180s sampling duration, 5min total timeout
    • These may not be appropriate for all use cases
  • 💡 Recommendation: Make timeouts configurable via MistralClientOptions

8. Missing Features

  • ⚠️ Missing: No support for custom headers
  • ⚠️ Missing: No retry-after header handling
  • ⚠️ Missing: No built-in rate limit detection

🐛 Potential Bugs

1. JSON Serialization Issue

File: ChatCompletionRequest.cs, Line 137

Parts = JsonSerializer.Deserialize&lt;List&gt;(ref reader, options),

Issue: If deserialization fails, this will throw JsonException which isn’t caught Impact: Could break entire request parsing Fix: Wrap in try-catch similar to streaming logic

2. Content Part Initialization

File: ChatCompletionRequest.cs, Line 137

Parts = JsonSerializer.Deserialize&lt;List&gt;(ref reader, options),

Issue: Creates mutable List but property expects IReadOnlyList Impact: Type mismatch – should use ToList().AsReadOnly() or change property type

3. Usage Info in Streaming

File: StreamingDTOs.cs Issue: UsageInfo appears on terminal chunk, but no validation that it’s only set once Impact: Could lead to incorrect token counting if multiple usage objects received Fix: Document this behavior or add validation


📈 Performance Considerations

Aspect Status Notes
Async/Await ✅ Excellent Proper use throughout
Memory Usage ✅ Good Streaming implementation is memory-efficient
JSON Serialization ✅ Good Uses System.Text.Json
HTTP Client ✅ Good Uses IHttpClientFactory for connection pooling
Object Allocation ⚠️ Minor Could reduce allocations in hot paths

Recommendation: Consider using ArrayPool for buffer management in high-throughput scenarios


🔧 Testing Recommendations

Unit Tests Needed

  1. Happy Path: Normal request/response flow
  2. Error Scenarios: HTTP 4xx/5xx responses
  3. Streaming: Chunk parsing, malformed JSON handling
  4. Serialization: All DTO serialization/deserialization
  5. Edge Cases: Empty messages, null values, boundary conditions
  6. Configuration: Invalid options, missing API key
  7. Resilience: Retry logic, circuit breaker behavior

Integration Tests Needed

  1. Real API Calls: Test against Mistral API (with test API key)
  2. Mock Server: Test with local mock server for controlled scenarios

📝 Code Style & Best Practices

Following .NET Best Practices

Practice Status
Async method naming (Async suffix)
Cancellation token support
IDisposable pattern for resources
Argument validation
Nullable reference types enabled ✅ (but needs more annotations)
Consistent naming conventions
XML documentation

Style Issues

  • ⚠️ Brace Style: Inconsistent – some files use same-line braces, others use new-line
  • ⚠️ Indentation: Mostly consistent (4 spaces), but verify throughout
  • ⚠️ Line Length: Some lines exceed typical 100-120 character limits

🛡️ Security Considerations

Aspect Status Notes
API Key Handling ✅ Good Supports both configured and custom auth headers
HTTPS ✅ Good Defaults to https://api.mistral.ai/
Input Validation ✅ Good Argument null checks
Error Messages ⚠️ Caution Error messages include response body (could expose sensitive data)
Rate Limiting ✅ Good Built-in via resilience pipeline

Recommendation: Consider sanitizing error messages in production to avoid exposing sensitive API response data


🚀 Usage Example

// Configuration
var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMistralChatClient(options =&gt;
{
    options.ApiKey = "your-api-key";
    options.BaseAddress = new Uri("https://api.mistral.ai/");
});

// Usage
var client = serviceProvider.GetRequiredService();

// Non-streaming
var request = new ChatCompletionRequest
{
    Model = "mistral-large",
    Messages = new List
    {
        new UserMessage { Content = "Hello, how are you?" }
    }
};

var response = await client.CompleteAsync(request);

// Streaming
await foreach (var chunk in client.StreamAsync(request))
{
    if (chunk.Choices.FirstOrDefault()?.Delta?.Content is { } content)
    {
        Console.Write(content);
    }
}

📊 Summary & Rating

Category Score Notes
Architecture ⭐⭐⭐⭐⭐ Excellent separation of concerns, DI-friendly
Code Quality ⭐⭐⭐⭐☆ Very good, minor inconsistencies
Error Handling ⭐⭐⭐⭐☆ Robust, could improve logging
Performance ⭐⭐⭐⭐☆ Good, streaming is efficient
Documentation ⭐⭐⭐⭐☆ Good XML docs, could use more examples
Testing ⭐⭐☆☆☆ Needs comprehensive test coverage
Security ⭐⭐⭐⭐☆ Good foundation, watch error messages

Overall Rating: ⭐⭐⭐⭐☆ (4.2/5) – Production Ready with Minor Improvements


Action Items

High Priority

  1. [P1] Fix namespace inconsistency between Mistral.Client and Mistral.Client.Shared
  2. [P1] Fix MessageContentJsonConverter to handle all edge cases properly
  3. [P1] Add validation for MistralClientOptions.ApiKey and BaseAddress

Medium Priority

  1. [P2] Add comprehensive unit tests
  2. [P2] Standardize all DTOs as immutable records
  3. [P2] Make resilience timeouts configurable
  4. [P2] Add nullability annotations to all properties

Low Priority

  1. [P3] Add logging for streaming errors
  2. [P3] Consider adding telemetry support
  3. [P3] Add integration tests with mock server

Review completed using Mistral Vibe | Generated: 2026-07-20

Real-time IoT Change Point Detection with ML.NET

This post covers change point detection, which is better suited for identifying persistent shifts in behaviour rather than momentary spikes. Specifically, comparing the two approaches available in ML.NET: Independent and identically distributed (IID) and Singular Spectrum Analysis (SSA) Change Point Detection

Change point detection is useful because IoT devices don’t always fail dramatically. Often, there’s a gradual or sustained change:

  • Temperature sensor starts reading consistently higher
  • A vibration sensor baseline shifts due to wear
  • Power consumption increases due to cooling issues

The sample application implements IID detection using DetectIidChangePoint and SSA with DetectChangePointBySsa.

IID assumes that: Incoming data points are independent and drawn from the same distribution. When the statistical properties shift (mean, variance), a change point is flagged.

AlgorithmLimitationsBest for
IIDDoesn’t account for trends or seasonality
Sensitive to noise
Fast detection
Simple signals
Low computational overhead
SSAComputationally expensive
Requires tuning
Signals with trends or seasonality
Noisy real-world IoT data
More robust long-term monitoring

SSA assumes that: The observed time‑series can be decomposed into a combination of trend, periodic components(seasonality), and noise. A change point is flagged when these components no longer match the historical structure of the series

BehaviourIIDSSA
Initial sensitivityHighModerate
Long-term behaviourAlways sensitiveAdapts
Noise tolerancePoorGood
Drift detectionKeeps firingFires once, then stops

In the application’s program.cs roughly the first 90 lines discusses changes to configuration to tune the models

var changePointEngine = _changePointEngines.GetOrAdd(subscribedTopic, _ =>
   new Lazy<TimeSeriesPredictionEngine<Model.TimeSeriesData, Model.ChangePointPrediction>>(() =>
   {
      try
      {
         switch (subscribedTopicSettings.DetectionMode)
         {
            case Model.DetectionMode.IID:
               var empty = _MLContext.Data.LoadFromEnumerable(new List<Model.TimeSeriesData>());

               IidChangePointEstimator iidPipe = _MLContext.Transforms.DetectIidChangePoint(
                              outputColumnName: nameof(Model.ChangePointPrediction.Prediction),
                              inputColumnName: nameof(Model.TimeSeriesData.Value),
                              confidence: subscribedTopicSettings.IIDSettings.Confidence,
                              changeHistoryLength: subscribedTopicSettings.IIDSettings.ChangeHistoryLength);

               var iidModel = iidPipe.Fit(empty);
               var iidEngine = iidModel.CreateTimeSeriesEngine<Model.TimeSeriesData, Model.ChangePointPrediction>(_MLContext);

               Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Initialized IID change point engine for '{subscribedTopic}' conf:{subscribedTopicSettings.IIDSettings.Confidence}, changeHistory:{subscribedTopicSettings.IIDSettings.ChangeHistoryLength}, AnomalySide:{subscribedTopicSettings.IIDSettings.AnomalySide}");
               return iidEngine;

            case Model.DetectionMode.SSA:
               SsaChangePointEstimator ssaPipe = _MLContext.Transforms.DetectChangePointBySsa(
                                 outputColumnName: nameof(Model.ChangePointPrediction.Prediction),
                                 inputColumnName: nameof(Model.TimeSeriesData.Value),
                                 confidence: subscribedTopicSettings.SSASettings.Confidence,
                                 changeHistoryLength: subscribedTopicSettings.SSASettings.ChangeHistoryLength,
                                 trainingWindowSize: subscribedTopicSettings.SSASettings.TrainingWindowSize,
                                 seasonalityWindowSize: subscribedTopicSettings.SSASettings.SeasonalityWindowSize);

               var dataView = _MLContext.Data.LoadFromEnumerable(new List<Model.TimeSeriesData>());
               var ssaModel = ssaPipe.Fit(dataView);
               var ssaEngine = ssaModel.CreateTimeSeriesEngine<Model.TimeSeriesData, Model.ChangePointPrediction>(_MLContext);

               Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Initialized SSA change point engine for '{subscribedTopic}' conf:{subscribedTopicSettings.SSASettings.Confidence}, changeHistory:{subscribedTopicSettings.SSASettings.ChangeHistoryLength}, trainingWindow:{subscribedTopicSettings.SSASettings.TrainingWindowSize}, seasonalityWindow:{subscribedTopicSettings.SSASettings.SeasonalityWindowSize}, AnomalySide:{subscribedTopicSettings.SSASettings.AnomalySide}");
               return ssaEngine;
            default:
               throw new NotSupportedException($"Detection mode {subscribedTopicSettings.DetectionMode} is not supported.");
         }

As part of a refactoring of configuration settings post my Real-time IoT Spike and Change Point Detection with ML.NET post the IID and SSA parameters have been split into separate classes

{
   "ApplicationSettings": {
      /*   */
      "SubscribedTopics": {
         "device/AshleyS1/distance": {
            "InputQualityOfService": 0,
            "OutputTopic": "alerts,ashley/changepoint",
            "OutputQualityOfService": 1,
            "ContentType": "application/json",
            "InputMessageTransformFile": "Transforms/InputSKU101991042.cs",
            "OutputMessageTransformFile": "Transforms/ChangePointOutput.cs",

            "DetectionMode": "IID", // IID or SSA
            "IIDSettings": {
               "Confidence": 95.0,
               "ChangeHistoryLength": 20,
               "AnomalySide": "TwoSided" //Positive, Negative,TwoSided
            },
            "SSASettings": {
               "Confidence": 95.0,
               "ChangeHistoryLength": 20,
               "SeasonalityWindowSize": 1,
               "TrainingWindowSize": 800,
               "AnomalySide": "TwoSided" //Positive, Negative,TwoSided
            }
         },
         "device/AshleyS72/distance": {
            "InputQualityOfService": 0,
            "OutputTopic": "alerts,ashley/changepoint",
            "OutputQualityOfService": 1,
            "ContentType": "application/json",
            "InputMessageTransformFile": "Transforms/InputSKU101991042.cs",
            "OutputMessageTransformFile": "Transforms/ChangePointOutput.cs",

            "DetectionMode": "IID",
            "IIDSettings": {
               "Confidence": 55.0,
               "ChangeHistoryLength": 10,
               "AnomalySide": "Negative" //Positive, Negative,TwoSided
            },
            "SSASettings": {
               "Confidence": 95.0,
               "ChangeHistoryLength": 20,
               "SeasonalityWindowSize": 2,
               "TrainingWindowSize": 300,
               "AnomalySide": "TwoSided" //Positive, Negative,TwoSided
            }
         }
      }
   }
}

In ML.NET time-series detectors (SSA and IID), the model doesn’t start detecting immediately. They first need to observe enough data to establish what “normal” looks like.

ParameterDescriptionTypical IoT Values
changeHistoryLengthNumber of points to evaluate stability30-100
trainingWindowSizeHistorical learning window (SSA)100-500
seasonalWindowSizeHistorical learning window (SSA)depends on cycle
confidenceSensitivity threshold90-99
AnomalySideStatistically unusual signalPositive, Negative, TwoSided

SSA requires a training window: trainingWindowSize: 200, this means the model needs roughly 200 samples before it is stable and before that predictions are unreliable and should be ignored. IID uses changeHistoryLength: 50 this means roughly 50 points are used to estimate the distribution and this is effectively the minimum history needed.

Warmup Matters: In the early stages the model is very sensitive as it has little context, and the baseline is weak which results in lots of detections and false positives (the detection below was after trainingWindowSize values). In the middle stage i.e. after enough samples the noise characteristics have been learned and patterns recognised. In the late stage after the model has been running for a long time it is very stable and less sensitive to change which can be an issue.

To cope with this (in a production system), I would ignore early predictions, for IID: changeHistoryLength and for SSA: trainingWindowSize + changeHistoryLength. To improve startup time preloading and SSA model with historical data or a synthetic base line is worth considering.

If an SSA model has been running for a longtime, or after a major change, reset the model to restore sensitivity. If sensitivity reduces over time it maybe worth considering “Dual Detectors: SSA for long-term structure, and IID for short-term sensitivity.

//---------------------------------------------------------------------------------
// Copyright (c) May 2026, devMobile Software
//
using System.Text.Json;
using System.Text.Json.Serialization; // Do not remove this using directive as it is required for the JsonIgnoreCondition

using devMobile.IoT.MqttTransformers;

public class ChangePointOutputTransformer : IChangePointOutputMessageTransformer
{
   private static readonly JsonSerializerOptions _serializerOptions = new() { DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull };

   public byte[] Transform(string sourceTopic, double value, double rawScore, double pValue, double martingale)
   {
      var obj = new
      {
         DetectionType = "ChangePoint",
         Value = value,
         SourceTopic = sourceTopic,
         RawScore = rawScore,
         PValue = pValue,
         Martingale = martingale,
      };

      string payload = JsonSerializer.Serialize(obj, _serializerOptions);

      return System.Text.Encoding.UTF8.GetBytes(payload);
   }
}

In the most recent version of the output transform ChangePoint = true was changed to DetectionType = “ChangePoint” to make it easier to identify change point messages for processing.

NOTE: SSA will not reliably detect slow, continuous movement – even if the total change is large

I was “tripped up” by this, it is not a bug, it’s an algorithm design choice to optimise for detecting regime changes, not motion. SSA is excellent at detecting when a system changes, but once it has adapted, it will treat that new state as completely normal.”

Real-time IoT Spike and Change Point Detection with ML.NET

Internet of Things(IoT) sensors generate streams of continuous numeric data. I wanted to detect spikes and change points without building full a Machine Learning (ML) pipeline per device. My approach is to subscribe to Message Queue Telemetry Transport (MQTT) topics using HiveMQ .NET Client Library, extract each message’s numeric value using CS-Script, process it with an ML.NET time-series prediction engine then finally publish an alert to a MQTT Topic when a spike is detected.

This post details the plumbing for detecting change points and spikes;

The next post will cover spike detection with Independent and identically distributed (IID) using DetectIidSpike, and Singular Spectrum Analysis (SSA) with DetectSpikeBySsa.

The final post will cover change point detection with Independent and identically distributed (IID) using DetectIidChangePoint. and Singular Spectrum Analysis (SSA) with DetectChangePointBySsa .

The configuration is stored in appsettings.json and when running on a desktop sensitive information AddUserSecrets() keeps credentials out of source control. The HiveMQ client configured with the builder pattern: The WithClientId, WithBroker, WithPort, WithCleanStart, WithAutomaticReconnect and WithUseTls are loaded from the appsettings.json. To stop the compiler whining about “deprecated methods” there are compile-time #if guards for username/password and certificate blocks.

// Load configuration
var configuration = new ConfigurationBuilder()
         .AddJsonFile("appsettings.json", optional: false, reloadOnChange: true)
         .AddUserSecrets<Program>()
         .Build();

_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>() ?? throw new Exception("ApplicationSettings not configured");

// HiveMQ client options
var optionsBuilder = new HiveMQClientOptionsBuilder()
         .WithClientId(_applicationSettings.ClientId)
         .WithBroker(_applicationSettings.Host)
         .WithPort(_applicationSettings.Port)
#if HIVEMQ_USERNAME_AND_PASSWORD_SUPPORT
         .WithUserName(_applicationSettings.UserName)
         .WithPassword(_applicationSettings.Password)
#endif
         .WithCleanStart(_applicationSettings.CleanStart)
         .WithAutomaticReconnect(_applicationSettings.AutomaticReconnect)
         .WithUseTls(_applicationSettings.UseTls);

#if HIVEMQ_CERTIFICATE_SUPPORT
if (!string.IsNullOrWhiteSpace(_applicationSettings.ClientCertificateFileName))
{
      optionsBuilder.WithClientCertificate(_applicationSettings.ClientCertificateFileName,_applicationSettings.ClientCertificatePassword);
   }
#endif

#if HIVEMQ_USERNAME_AND_PASSWORD_SUPPORT
if (!string.IsNullOrWhiteSpace(_applicationSettings.UserName))
{
   optionsBuilder = optionsBuilder.WithPassword(_applicationSettings.Password);
}
#endif

The ApplicationSettings.SubscribedTopics are loaded into a dictionary with the MQTT topic as the key. Each subscription has per-topic configuration (detection mode, confidence, window sizes, QoS, output topic).

The HIVEMQ_CERTIFICATE_SUPPORT and HIVEMQ_USERNAME_AND_PASSWORD_SUPPORT exclude calls to WithUserName, WithPassword and WithClientCertificate(fileName, password). This is required (from version 0.39 of the HiveMQtt .NETClient) even though SecureString is considered obsolete in .NET for cross-platform work.

Converting a plain string from appsettings.json into a SecureString is a PITA as you must iterate characters and AppendChar one by one, then call MakeReadOnly(). SecureString limits the window during which the plaintext lives in managed heap memory,
reducing exposure to memory dumps. I don’t think that trade-off is worth the ceremony is debatable for most IoT scenarios.

Each subscribed topic has two scripts (InputMessageTransformFile and OutputMessageTransformFile) which CSScript.Evaluator.LoadFile(path) compiles, caches and instantiates a C# script at runtime

The IInputMessageTransformer, ISpikeOutputMessageTransformer follow the same pattern for the inbound and outbound payloads. The interfaces decouple the spike detection core from message format. A new or updated .cs script file to handle a new sensor payload can be deployed without touching the host application.

Exception handling is non-optional: script compilation errors surface as exceptions. The try/catch around the foreach that loads transformers is the right place to fail fast at startup rather than silently skipping a topic.

Console.CancelKeyPress sets e.Cancel = true (which suppresses the default termination) and calls cts.Cancel(). Await Task.Delay(Timeout.Infinite, cts.Token), the program does nothing in Main while the event-driven OnMessageReceived handler does all the work. When Ctrl+C fires, Task.Delay throws TaskCanceledException, which is caught in the outer try/catch block and prints a clean shutdown message. Everything else falls through to finally, which disconnects the HiveMQ client.

// one engine per topic; keeps rolling state for IID detector
private static readonly ConcurrentDictionary<string, Lazy<TimeSeriesPredictionEngine<Model.TimeSeriesData, Model.SpikePrediction>>> _spikeEngines = new();

// lock per topic because TimeSeriesPredictionEngine is not thread-safe
private static readonly ConcurrentDictionary<string, object> _engineLocks = new();

...

var spikeEngine = _spikeEngines.GetOrAdd(subscribedTopic, _ =>
   new Lazy<TimeSeriesPredictionEngine<Model.TimeSeriesData, Model.SpikePrediction>>(() =>
or...
var changePointEngine = _changePointEngines.GetOrAdd(subscribedTopic, _ =>
   new Lazy<TimeSeriesPredictionEngine<Model.TimeSeriesData, Model.ChangePointPrediction>>(() =>
{

...

}, LazyThreadSafetyMode.PublicationOnly)).Value;

The _spikeEngines/_changePointEngines is a ConcurrentDictionary keyed by topic string, GetOrAdd with a Lazy<> factory ensures each engine is initialised exactly once, even under concurrent message arrival. The LazyThreadSafetyMode.PublicationOnly means multiple threads may race to construct the value but only one result survives.

Each engine carries rolling internal state (the sliding window of past values), so isolation per topic is essential. The “mixing” of values from different sensors(topics) into one engine would produce nonsense predictions. The DetectionMode enumeration (IID / SSA) is checked inside the Lazy factory, branching to the appropriate ML.NET estimator. The engine type is the same either way (TimeSeriesPredictionEngine) only the output transform and pipeline construction differs.

Two separate try/catch blocks wrap the two transforms in each program, so failures (runtime errors like, bad JSON, missing field, type mismatch etc.) don’t crash the handler or the engine.

float value;
try
{
   value = subscribedTopicSettings.InputMessageTransformer.Transform(subscribedTopic, e.PublishMessage.Payload);
}
catch (Exception ex)
{
   Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Input transform failed: {ex.Message}");
   return;
}
foreach (string topic in topics)
{
   Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Publishing to:{topic}");

   var message = new MQTT5PublishMessage(topic, subscribedTopicSettings.OutputQualityOfService)
   {
      ContentType = subscribedTopicSettings.ContentType,
      Payload = payload
   };

   try
   {
      var resultPublish = await client.PublishAsync(message);

      Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} Published:{resultPublish.QoS1ReasonCode} {resultPublish.QoS2ReasonCode}");
   }
   catch (Exception ex)
   {
      Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss:fff} client.PublishAsync failed: {ex.Message}");
      return;
   }
}

The async void OnMessageReceived wrapper catches any unhandled exception from OnMessageReceivedCoreAsync. This is necessary because the async void “swallows” exceptions that would otherwise terminate the process silently. This approach is required because a bad/broken payload from one device should never take down detection for all other topics.

Christchurch Azure User Group Session April 2026

Faster, Cheaper, Scalable: Architecting High-Performance Azure Apps with Caching

Details

“There are 2 hard problems in computer science: cache invalidation, naming things, and off-by-1 errors.” — Leon Bambrick

Join us as Microsoft MVP Bryn Lewis shows us how caching is the ultimate “cheat code” for cloud architecture. When implemented correctly, it’s the fastest way to slash your Azure consumption costs, reduce database contention, and keep your application responsive under massive load. But move beyond simple lookups – your deployment model and caching strategy can make or break your app’s reliability. In this session, we’ll move from the browser edge to the distributed core:

  • Optimizing the Edge: Leverage RFC-standard HTTP semantics to offload traffic to CDNs and browsers, cutting ingress/egress costs before requests even reach your App Service.
  • Saving Compute: See how ASP.NET Core Output Caching rescues your CPU from redundant work, allowing you to scale out less frequently and save on your monthly Azure bill.
  • Modern Object Strategies: A deep dive into HybridCache and FusionCache. We’ll compare L1/L2 strategies and master “the dark arts” of stampede protection and cache invalidation to ensure high availability.
  • The Power of Azure Cache for Redis: We’ll close by configuring Redis as a distributed L2 cache, ensuring your cloud applications stay fast, synchronized, and resilient across multiple instances.

The code I used to double check my assumptions is available on GitHub. This repo demonstrates various .NET caching strategies (FusionCache, HybridCache, OutputCache, Redis, etc.) against a real Azure SQL Server backend.

Dapper Extensions

All the demo projects use my DapperExtensions project, so every cache benchmark hits the database through the same resilient layer, meaning the retry logic “should never” skew the results.

DIYCache – Rolling your own cache in 80 lines

The cache is a ConcurrentDictionary<string,CacheItem<T>>> registered as a singleton. The GET endpoint uses the cache-aside pattern, return if valid and not expired, otherwise hit the database, and store the result with a configurable TTL, then return. A companion DELETE endpoint evicts a specific entry with a single TryRemove call. The cache has no background eviction, or stampede protection, and the size is “unbounded”

Fusion Cache – Scale with configuration

FusionCache is a read through cache with Fast L1/Shared L2 support which hides the checking, fetching, and storing in three separate steps, with a factory lambda to manage the process. Cache invalidation uses tags, each entry is stamped at write time, and a single RemoveByTagAsync call evicts every matching entry. In the sample project Stack Exchange Redis is opt-in via configuration. Add the required connection string and FusionCache becomes a two-tier cache: fast in-memory L1 backed by distributed Redis L2. Add a backplane connection string and invalidation signals propagate across all running instances. The same code works as a single-process cache in development and a fully distributed one in production with no code changes. Realistically it was what I was hoping HybridCache would be

HTTP Head – RFC 9111 IETF “HTTP Caching”.

The HTTPHead project shows how HTTP’s HEAD method and ETags can eliminate unnecessary data transfers. When a client fetches a neighborhood record via GET, it receives an ETag derived from the Azure SQL Server rowversion (replaces the TimeStamp which has been deprecated) column. On subsequent checks, it sends that ETag to the HEAD endpoint, which queries only the version column and returns 304 Not Modified or 200 OK no payload needed. The PUT endpoint uses optimistic concurrency, rejecting updates where the ETag no longer matches. This ensures clients only download data that has actually changed.

Hybrid Cache – If only

HybridCache is a two-tier cache that sits in front of both an in-process L1 cache and an optional Redis L2 cache behind a single GetOrCreateAsync call. In the sample code NeighborHood lookups are cached with a 5-minute in-memory expiry and a 30-minute Stack Exchange Redis expiry, so repeated requests within the same process never leave the machine, while distributed deployments still share a warm cache across instances.

Hybrid Cache Serialization – When less sent to the L2 Cache is more

The HybridCacheSerialization project extends the HybridCache sample by swapping the default JSON serializer for others like Neuecc MessagePack. HybridCache exposes an IHybridCacheSerializer interface, so developers can plug in different serialisers. In the sample the Data Transfer Object(DTO) is decorated with [MessagePackObject] and [Key(n)] attributes to control the binary layout (MessagePack message format is supported by many languages). The payoff is compact, fast binary payloads stored in Stack Exchange Redis instead of verbose JSON. This is worthwhile when cached objects are large, retrieved frequently, or bandwidth between app and cache is a latency/jitter/cost concern.

Object Cache – Barely sufficient

The ObjectCache project is the simplest (non-DIY) option using just IMemoryCache. Neighborhood lookups are wrapped in GetOrCreateAsync: a hit returns the cached object instantly, a miss queries Azure SQL Server and caches the result for 5 minutes. In this example a database miss isn’t just returned as NotFound and forgotten, this is cached too, for 1 minute, so a flood of requests for a non-existent record won’t hammer the database. A DELETE endpoint lets callers evict a specific entry on demand.

Output Cache – Avoiding regeneration, but don’t cross the streams.

The OutputCache project demonstrates ASP.NET Core’s OutputCaching middleware, a response-level cache that stores the fully serialized HTTP responses rather than the underlying objects. Output caching short-circuits the entire endpoint and serves the cached bytes directly. The project has named policies (“short”, “medium”, “neighborhood”) defined at startup and applied to endpoints with .CacheOutput(), inline policies defined inline as a lambda, Stack Exchange Redis can be dropped in
as the backing store with no code changes

MIDDLEWARE ORDER MATTERS- Place AFTER authentication/authorization so user identity and policies respected

Redis Cache – Old school and amazingly Fast

The RedisCache project goes bare-metal, using the Stack Exchange Redis IConnectionMultiplexer directly rather than any .NET caching abstraction. The cache-aside pattern used, check Redis first, then fall back to the database on a miss, then write the result back with a 30-second TTL. This sample uses source-generated JSON serializationvia JsonSerializerContext: serialization and deserialization use pre-compiled code paths rather than runtime reflection, which keeps allocation low and throughput high on the hot path. This also enables Ahead of Time(AoT) compilation support.

ResponseCache – RFC 9110 IETF “HTTP Semantics”

The ResponseCache project covers ASP.NET Core’s older ResponseCaching middleware, which caches responses based on standard HTTP Cache-Control headers rather than any framework-specific API. The endpoint sets Cache-Control: public, max-age=90 directly on the response headers and the middleware handles the rest. ResponseCache has largely been replaced by Output Cache though it matters when managing the caching behaviour of downstream proxies and Content Delivery Networks(CDNs), because the Cache-Control headers it emits are understood by the full HTTP stack.

Response Compression – When less sent to the client is more

The ResponseCompression middleware is server-side complement to caching that reduces payload size rather than request database traffic. The sample supports Gzip (faster,universally supported) and Brotli (better compression ratio, higher CPU cost), with an optionalflag to tune the trade-off between speed and size.

The application/json content-type isn’t compressed by default so it must be added explicitly to the Multipurpose Internet Mail Extension(MIME) type list; EnableForHttps must be opted into deliberately since compressing encrypted responses can expose reflected secrets (the CRIME/BREACH attacks); and Azure App Service containers apply their own platform-level gzip, so enabling this middleware there risks double-compression. Clients must send Accept-Encoding: gzip for compression as it’s not automatic.

The full source is available in the CHCAzureUGC202604 repository alongside the caching demos it supports

ONNX Tensor loading Initial Comparison

This is the second in a series of posts from my session at the Agent Camp – Christchurch about using Open Neural Network Exchange(ONNX) for processing Moving Picture Experts Group (MPEG) video and Pulse Code Modulation(PCM) audio streams.

These benchmarks use Ultralytics Yolo26 standard object detection model input image size of 640*640pixels.

var _tensor= new DenseTensor<float>(new[] { 1, 3, modelH, modelW });

The original nested loop: multi-dimensional [0,c,y,x] indexer, with divide by 255f. This is the baseline to measure all other implementations against.

[Benchmark(Baseline = true, Description = "Baseline: indexer + / 255f")]
public void Baseline()
{
   for (int y = 0; y < modelH; y++)
      for (int x = 0; x < modelW; x++)
      {
          var c = _letterboxed.GetPixel(x, y);

         _tensor[0, 0, y, x] = px.Red / 255f;
         _tensor[0, 1, y, x] = px.Green / 255f;
         _tensor[0, 2, y, x] = px.Blue / 255f;
      }
}

The implementation bypasses the multi-dimensional [0,c,y,x] indexer entirely with Span<> over the tensor’s backing buffer. Channel planes are at offsets 0, planeSize, and 2*planeSize. Then a single loop reads each pixel once; writes to all three planes interleaved.

[Benchmark(Description = "Buffer span: flat index, interleaved")]
public void BufferSpan()
{
   SKColor[] pixels = _letterboxed.Pixels;
   const float scaler = 1 / 255f;
   int planeSize = _modelW* _modelW;
   Span<float> buf = _tensor.Buffer.Span;

   for (int i = 0; i < planeSize; i++)
   {
      SKColor px = pixels[i];
      buf[i] = px.Red * scaler;
      buf[planeSize + i] = px.Green * scaler;
      buf[2 * planeSize + i] = px.Blue * scaler;
   }
}

This implementation slices the flat buffer into three non-overlapping channel spans, it then runs three separate sequential loops, one for each colour. This Combines the benefits of span (no indexer overhead, JIT can also auto-vectorise) and with split loops which the JIT can eliminate per-element bounds checks after the slice.

   [Benchmark(Description = "Buffer span split: 3× sequential flat loops")]
   public void BufferSpanSplit()
   {
      SKColor[] pixels = _letterboxed.Pixels;
      const float scaler = 1 / 255f;
      int planeSize = _modelW* _modelH;
      Span<float> buf = _tensor.Buffer.Span;

      Span<float> rPlane = buf.Slice(0, planeSize);
      Span<float> gPlane = buf.Slice(planeSize, planeSize);
      Span<float> bPlane = buf.Slice(2 * planeSize, planeSize);

      for (int i = 0; i < planeSize; i++) rPlane[i] = pixels[i].Red * scaler;
      for (int i = 0; i < planeSize; i++) gPlane[i] = pixels[i].Green * scaler;
      for (int i = 0; i < planeSize; i++) bPlane[i] = pixels[i].Blue * scaler;
   }

The minimal difference in performance of the two fastest implementations of the benchmark suite running on my development box was a surprise. It will be interesting to see how the performance of the different implementations changes on my Seeedstudio EdgeBox RPi 200 which has a different instruction set (esp. ARM NEON Single Instruction, Multiple Data (SIMD) extensions) and memory caching model

These benchmarks should be treated as indicative not authoritative 

SkiaSharp and ImageSharp Initial Comparison

This is the first in a series of posts from my session at the Agent Camp – Christchurch about using Open Neural Network Exchange(ONNX) for processing Moving Picture Experts Group (MPEG) video and Pulse Code Modulation(PCM) audio streams.

For processing video streams one of the first steps is extracting individual Joint Photographic Experts Group(JPEG) images from MPEG Real-Time Streaming Protocol(RTSP) stream. The jpeg images then have to transformed into an ONNX DenseTensor<float> in the correct format for the Ultralytics Yolo26 model. These image processing posts will use Ultralytics Yolo26 standard Small object detection model which has an input image size of 640*640pixels.

I have used both the YoloSharp and YoloDotNet libraries (Thank you Niklas Swärd and dme-compunet I appreciate the amount of effort you have put in). Both these libraries have support for object detection, instance segmentation, oriented bounding boxes detection(OBB), classification and pose estimation. They both have support for different versions, video stream processing, plotting minimum bounding boxes, Non-Maximum Suppression(NMS) for earlier models like YOLOv8 or YOLO11. I just need object detection (none of the other model types, plotting minimum boxes etc.) to work as fast as possible on my Seeedstudio EdgeBox RPi 200.

First step, was to use Benchmark.Net compare the performance of Six Labors ImageSharp (used by YoloSharp) and SkiaSharp (used by YoloDotNet). Six Labors ImageSharp  is a high-performance, fully managed, 2D graphics API whereas SkiaSharp is a wrapper for Google’s Skia 2D Graphics Library.

ImageSharp Benchmark
SkiaSharp Benchmark

The initial comparison running on my development box (will benchmark on my Seeedstudio EdgeBox RPi 200.) was roughly what I was expecting though the SkaiSharp 2560×1440 mean duration was a bit odd. I think that the difference in the amount of memory allocated is because SkaiSharp’s memory is allocated by the native code. Both benchmarks need some refactoring to improve repeatability on my different platforms.

These benchmarks should be treated as indicative not authoritative 

Seeedstudio XIAO ESP32 S3 RS-485 test harness(nanoFramework)

As part of a project to read values from a MODBUS RS-485 sensor using a RS-485 Breakout Board for Seeed Studio XIAO and a Seeed Studio XIAO ESP32-S3 I built a .NET nanoFramework version of the Arduino test harness described in this wiki post.

This took a bit longer than I expected mainly because running two instances of Visual Studio 2026 was a problem (running Visual Studio 2022 for one device and Visual Studio 2026 for the other, though not 100% confident this was an issue) as there were some weird interactions.

using nanoff to flash a device with the latest version of ESP32_S3_ALL_UART

As I moved between the Arduino tooling and flashing devices with nanoff the serial port numbers would change watching the port assignments in Windows Device Manager was key.

Windows Device manager displaying the available serial ports

Rather than debugging both the nanoFramework RS485Sender and RS485Receiver applications simultaneously, I used the Arduino RS485Sender and RS485 Receiver application but had similar issues with the port assignments changing.

Arduino RS485 Sender application
The nanoFramework sender application
public class Program
{
   static SerialPort _serialDevice;

   public static void Main()
   {
      Configuration.SetPinFunction(Gpio.IO06, DeviceFunction.COM2_RX);
      Configuration.SetPinFunction(Gpio.IO05, DeviceFunction.COM2_TX);
      Configuration.SetPinFunction(Gpio.IO02, DeviceFunction.COM2_RTS);

      Debug.WriteLine("RS485 Sender: ");

      var ports = SerialPort.GetPortNames();

      Debug.WriteLine("Available ports: ");
      foreach (string port in ports)
      {
         Debug.WriteLine($" {port}");
      }

      _serialDevice = new SerialPort("COM2");
      _serialDevice.BaudRate = 9600;
      _serialDevice.Mode = SerialMode.RS485;

      _serialDevice.Open();

      Debug.WriteLine("Sending...");
      while (true)
      {
         string payload = $"{DateTime.UtcNow:HHmmss}";

         Debug.WriteLine($"Sent:{DateTime.UtcNow:HHmmss}");

         Debug.WriteLine(payload);

         _serialDevice.WriteLine(payload);

         Thread.Sleep(2000);
      }
   }
}

if I had built the nanoFramework RS485Sender and RS485Receiver applications first debugging the Arduino RS485Sender and RS485Receiver would been similar.

Arduino receiver application displaying messages from the nanoFramework sender application
The nanoFramework Receiver receiving messages from the nanoFramework Sender
public class Program
{
   static SerialPort _serialDevice ;
 
   public static void Main()
   {
      Configuration.SetPinFunction(Gpio.IO06, DeviceFunction.COM2_RX);
      Configuration.SetPinFunction(Gpio.IO05, DeviceFunction.COM2_TX);
      Configuration.SetPinFunction(Gpio.IO02, DeviceFunction.COM2_RTS);

      Debug.WriteLine("RS485 Receiver ");

      // get available ports
      var ports = SerialPort.GetPortNames();

      Debug.WriteLine("Available ports: ");
      foreach (string port in ports)
      {
         Debug.WriteLine($" {port}");
      }

      // set parameters
      _serialDevice = new SerialPort("COM2");
      _serialDevice.BaudRate = 9600;
      _serialDevice.Mode = SerialMode.RS485;

      // set a watch char to be notified when it's available in the input stream
      _serialDevice.WatchChar = '\n';

      // setup an event handler that will fire when a char is received in the serial device input stream
      _serialDevice.DataReceived += SerialDevice_DataReceived;

      _serialDevice.Open();

      Debug.WriteLine("Waiting...");
      Thread.Sleep(Timeout.Infinite);
   }

   private static void SerialDevice_DataReceived(object sender, SerialDataReceivedEventArgs e)
   {
      SerialPort serialDevice = (SerialPort)sender;

      switch (e.EventType)
      {
         case SerialData.Chars:
         //break;

         case SerialData.WatchChar:
            string response = serialDevice.ReadExisting();
            Debug.Write($"Received:{response}");
            break;
         default:
            Debug.Assert(false, $"e.EventType {e.EventType} unknown");
            break;
      }
   }
}

The changing of serial port numbers while running different combinations of Arduino and nanoFramework environments concurrently combined with the sender and receiver applications having to be deployed to the right devices (also initially accidentally different baud rates) was a word of pain, and with the benefit of hindsight I should have used two computers.

Azure Event Grid nanoFramework Client – Publisher

Building a .NET nanoFramework application for testing Azure Event Grid MQTT Broker connectivity that would run on my Seeedstudio EdgeBox ESP100 and Seeedstudio Xiao ESP32S3 devices took a couple of hours. Most of that time was spent figuring out how to generate the certificate and elliptic curve private key

Create an elliptic curve private key

 openssl ecparam -name prime256v1 -genkey -noout -out device.key

Generate a certificate signing request

openssl req -new -key device.key -out device.csr -subj "/CN=device.example.com/O=YourOrg/OU=IoT"

Then use the intermediate certificate and key file from earlier to generate a device certificate and key.

 openssl x509 -req -in device.csr -CA IntermediateCA.crt -CAkey IntermediateCA.key -CAcreateserial -out device.crt -days 365 -sha256

In this post I have assumed that the reader is familiar with configuring Azure Event Grid clients, client groups, topic spaces, permission bindings and routing.

The PEM encoded root CA certificate chain that is used to validate the server
public const string CA_ROOT_PEM = @"-----BEGIN CERTIFICATE-----
CN: CN = Microsoft Azure ECC TLS Issuing CA 03
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
CN: CN = DigiCert Global Root G3
-----END CERTIFICATE-----";

The PEM encoded certificate chain that is used to authenticate the device
public const string CLIENT_CERT_PEM_A = @"-----BEGIN CERTIFICATE-----
-----BEGIN CERTIFICATE-----
 CN=Self signed device certificate
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
 CN=Self signed Intermediate certificate
-----END CERTIFICATE-----";

 The PEM encoded private key of device
public const string CLIENT_KEY_PEM_A = @"-----BEGIN EC PRIVATE KEY-----
-----END EC PRIVATE KEY-----";

My application was “inspired” by the .NET nanoFramework m2mqtt example.

public static void Main()
{
   int sequenceNumber = 0;
   MqttClient mqttClient = null;
   Thread.Sleep(1000); // Found this works around some issues with running immediately after a reset

   bool wifiConnected = false;
   Console.WriteLine("WiFi connecting...");
   do
   {
      // Attempt to connect using DHCP
      wifiConnected = WifiNetworkHelper.ConnectDhcp(Secrets.WIFI_SSID, Secrets.WIFI_PASSWORD, requiresDateTime: true);

      if (!wifiConnected)
      {
         Console.WriteLine($"Failed to connect. Error: {WifiNetworkHelper.Status}");
         if (WifiNetworkHelper.HelperException != null)
         {
            Console.WriteLine($"Exception: {WifiNetworkHelper.HelperException}");
         }

         Thread.Sleep(1000);
      }
   }
   while (!wifiConnected);
   Console.WriteLine("WiFi connected");

   var caCert = new X509Certificate(Constants.CA_ROOT_PEM);

   X509Certificate2 clientCert = null;
   try
   {
      clientCert = new X509Certificate2(Secrets.CLIENT_CERT_PEM_A, Secrets.CLIENT_KEY_PEM_A, string.Empty);
   }
   catch (Exception ex)
   {
      Console.WriteLine($"Client Certificate Exception: {ex.Message}");
   }

   mqttClient = new MqttClient(Secrets.MQTT_SERVER, Constants.MQTT_PORT, true, caCert, clientCert, MqttSslProtocols.TLSv1_2);

   mqttClient.ProtocolVersion = MqttProtocolVersion.Version_5;

   bool mqttConnected = false;
   Console.WriteLine("MQTT connecting...");
   do
   {
      try
      {
         // Regular connect
         var resultConnect = mqttClient.Connect(Secrets.MQTT_CLIENTID, Secrets.MQTT_USERNAME, Secrets.MQTT_PASSWORD);
         if (resultConnect != MqttReasonCode.Success)
         {
            Console.WriteLine($"MQTT ERROR connecting: {resultConnect}");
            Thread.Sleep(1000);
         }
         else
         {
            mqttConnected = true;
         }
      }
      catch (Exception ex)
      {
         Console.WriteLine($"MQTT ERROR Exception '{ex.Message}'");
         Thread.Sleep(1000);
      }
   }
   while (!mqttConnected);
   Console.WriteLine("MQTT connected...");

   mqttClient.MqttMsgPublishReceived += MqttMsgPublishReceived;
   mqttClient.MqttMsgSubscribed += MqttMsgSubscribed;
   mqttClient.MqttMsgUnsubscribed += MqttMsgUnsubscribed;
   mqttClient.ConnectionOpened += ConnectionOpened;
   mqttClient.ConnectionClosed += ConnectionClosed;
   mqttClient.ConnectionClosedRequest += ConnectionClosedRequest;

   string topicPublish = string.Format(MQTT_TOPIC_PUBLISH_FORMAT, Secrets.MQTT_CLIENTID);
   while (true)
   {
      Console.WriteLine("MQTT publish message start...");

      var payload = new MessagePayload() { ClientID = Secrets.MQTT_CLIENTID, Sequence = sequenceNumber++ };

      string jsonPayload = JsonSerializer.SerializeObject(payload);

      var result = mqttClient.Publish(topicPublish, Encoding.UTF8.GetBytes(jsonPayload), "application/json; charset=utf-8", null);

      Debug.WriteLine($"MQTT published ({result}): {jsonPayload}");

      Thread.Sleep(100);
   }
}

I then configured my client (Edgebox100Z) and updated the “secrets.cs” file

Azure Event Grid MQTT Broker Clients

The application connected to the Azure Event Grid MQTT broker and started publishing the JSON payload with the incrementing sequence number.

Visual Studio debugger output of JSON payload publishing

The published messages were “routed” to an Azure Storage Queue where they could be inspected with a tool like Azure Storage Explorer.

Azure Event Grid MQTT Broker metrics with messages published selected

I could see the application was working in the Azure Event Grid MQTT broker metrics because the number of messages published was increasing.

Azure Event Grid Arduino Client – Publisher

The Arduino application for testing Azure Event Grid MQTT Broker connectivity worked on my Seeedstudio EdgeBox ESP100 and Seeedstudio Xiao ESP32S3 devices, so the next step was to modify it to publish some messages.

The first version generated the JSON payload using an snprintf which was a bit “nasty”

static uint32_t sequenceNumber = 0;

void loop() {
  mqttClient.loop();

  Serial.println("MQTT Publish start");

  char payloadBuffer[64];

  snprintf(payloadBuffer, sizeof(payloadBuffer), "{\"ClientID\":\"%s\", \"Sequence\": %i}", MQTT_CLIENTID, sequenceNumber++);

  Serial.println(payloadBuffer);

  if (!mqttClient.publish(MQTT_TOPIC_PUBLISH, payloadBuffer, strlen(payloadBuffer))) {
    Serial.print("\nMQTT publish failed:");        
    Serial.println(mqttClient.state());    
  }
  Serial.println("MQTT Publish finish");

  delay(60000);
}

I then configured my client (Edgebox100A) and updated the “secrets.h” file

Azure Event Grid MQTT Broker Clients

The application connected to the Azure Event Grid MQTT broker and started publishing the JSON payload with the incrementing sequence number.

Arduino IDE serial monitor output of JSON payload publishing

The second version generated the JSON payload using ArduinoJson library.

static uint32_t sequenceNumber = 0;

void loop() {
  mqttClient.loop();

  Serial.println("MQTT Publish start");

  // Create a static JSON document with fixed size
  StaticJsonDocument<64> doc;

  doc["Sequence"] = counter++;
  doc["ClientID"] = MQTT_CLIENTID;

  // Serialize JSON to a buffer
  char jsonBuffer[64];
  size_t n = serializeJson(doc, jsonBuffer);

  Serial.println(jsonBuffer);

  if(!mqttClient.publish(MQTT_TOPIC_PUBLISH, jsonBuffer, n))
  {
    Serial.println(mqttClient.state());    
  }

  Serial.println("MQTT Publish finish");

  delay(2000);
}

I could see the application was working in the Azure Event Grid MQTT broker metrics because the number of messages published was increasing.

Azure Event Grid MQTT Broker metrics with messages published selected

The published messages were “routed” to an Azure Storage Queue where they can be inspected with a tool like Azure Storage Explorer.

Azure Storage Explorer displaying a message’s payload

The message payload is in Base64 encoded so I used copilot convert it to text.

Microsoft copilot decoding the Base64 payload

In this post I have assumed that the reader is familiar with configuring Azure Event Grid clients, client groups, topic spaces, permission bindings and routing.

Bonus also managed to slip in a reference to copilot.