From bf603102533a6d92ff6735be5d4fe83729084afb Mon Sep 17 00:00:00 2001 From: Daniel Meza Date: Fri, 23 Jan 2026 18:12:41 -0500 Subject: [PATCH 01/14] Enhance TLS configuration and error handling in MqttClientService --- Source/Services/Mqtt/MqttClientService.cs | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/Source/Services/Mqtt/MqttClientService.cs b/Source/Services/Mqtt/MqttClientService.cs index 80ab17c..bd10544 100644 --- a/Source/Services/Mqtt/MqttClientService.cs +++ b/Source/Services/Mqtt/MqttClientService.cs @@ -96,7 +96,7 @@ public async Task Connect(ConnectionItemViewModel item) o.WithUri(item.ServerOptions.Host); }); } - + if (item.ServerOptions.SelectedTlsVersion.Value != SslProtocols.None) { clientOptionsBuilder.WithTlsOptions(o => @@ -104,7 +104,12 @@ public async Task Connect(ConnectionItemViewModel item) o.WithSslProtocols(item.ServerOptions.SelectedTlsVersion.Value); o.WithIgnoreCertificateChainErrors(item.ServerOptions.IgnoreCertificateErrors); o.WithIgnoreCertificateRevocationErrors(item.ServerOptions.IgnoreCertificateErrors); - o.WithCertificateValidationHandler(item.ServerOptions.IgnoreCertificateErrors ? _ => true : null); + + if (item.ServerOptions.IgnoreCertificateErrors) + { + o.WithCertificateValidationHandler(context => true); + o.WithAllowUntrustedCertificates(); + } if (!string.IsNullOrEmpty(item.SessionOptions.CertificatePath)) { @@ -136,11 +141,11 @@ public async Task Connect(ConnectionItemViewModel item) { return await _mqttClient.ConnectAsync(clientOptionsBuilder.Build(), timeout.Token); } - catch (OperationCanceledException) + catch (OperationCanceledException ex) { if (timeout.IsCancellationRequested) { - throw new MqttCommunicationTimedOutException(); + throw new MqttCommunicationTimedOutException(ex); } throw; From d540f63141c398c6db69dc3ff0c4c7f17654487e Mon Sep 17 00:00:00 2001 From: Daniel Meza Date: Fri, 23 Jan 2026 19:15:36 -0500 Subject: [PATCH 02/14] Enhance TLS options in MqttClientService for improved security handling --- Source/Services/Mqtt/MqttClientService.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Source/Services/Mqtt/MqttClientService.cs b/Source/Services/Mqtt/MqttClientService.cs index bd10544..d6cf8dc 100644 --- a/Source/Services/Mqtt/MqttClientService.cs +++ b/Source/Services/Mqtt/MqttClientService.cs @@ -101,6 +101,7 @@ public async Task Connect(ConnectionItemViewModel item) { clientOptionsBuilder.WithTlsOptions(o => { + o.UseTls(); o.WithSslProtocols(item.ServerOptions.SelectedTlsVersion.Value); o.WithIgnoreCertificateChainErrors(item.ServerOptions.IgnoreCertificateErrors); o.WithIgnoreCertificateRevocationErrors(item.ServerOptions.IgnoreCertificateErrors); @@ -108,7 +109,9 @@ public async Task Connect(ConnectionItemViewModel item) if (item.ServerOptions.IgnoreCertificateErrors) { o.WithCertificateValidationHandler(context => true); - o.WithAllowUntrustedCertificates(); + o.WithAllowUntrustedCertificates(true); + o.WithIgnoreCertificateChainErrors(true); + o.WithIgnoreCertificateRevocationErrors(true); } if (!string.IsNullOrEmpty(item.SessionOptions.CertificatePath)) From 2eda8b36d03515a96a69e1a2600d224b120f7faa Mon Sep 17 00:00:00 2001 From: Daniel Meza Date: Wed, 11 Feb 2026 22:03:56 -0500 Subject: [PATCH 03/14] Add high-throughput guide for handling 8K+ MQTT messages in desktop UI applications This document outlines performance considerations, optimization strategies, and implementation details for managing high-frequency MQTT messages in a UI context. It covers dispatcher queue behavior, memory management, and recommended settings for various throughput scenarios, along with alternative architectures for extreme throughput. --- Source/App.axaml | 1 + Source/App.axaml.cs | 48 +- Source/Controls/AutoGrid.cs | 15 +- Source/Extensions/CollectionTrimExtensions.cs | 29 + Source/Logging/LogObserverExceptionHandler.cs | 46 + Source/Main/MainView.axaml | 99 +- Source/Main/MainView.axaml.cs | 17 + Source/Main/MainViewModel.cs | 122 +- .../BoundedChannelFullModeOption.cs | 23 + .../Pages/Connection/ConnectionItemView.axaml | 2 + .../Connection/ConnectionItemViewModel.cs | 2 + .../Connection/ConnectionOptionsView.axaml | 148 +++ .../Connection/ConnectionPageViewModel.cs | 41 +- .../MessageProcessingOptionsViewModel.cs | 166 +++ .../Connection/MessageProcessingProfile.cs | 82 ++ .../Export/InflightPageItemExportService.cs | 3 +- Source/Pages/Inflight/InflightPageView.axaml | 3 +- .../Pages/Inflight/InflightPageView.axaml.cs | 47 +- .../Pages/Inflight/InflightPageViewModel.cs | 101 +- Source/Pages/Log/LogPageView.axaml | 5 +- Source/Pages/Log/LogPageViewModel.cs | 88 +- .../PacketInspectorPageView.axaml | 7 +- .../PacketInspectorPageViewModel.cs | 162 ++- .../Pages/PacketInspector/PacketViewModel.cs | 2 +- .../SubscriptionsPageViewModel.cs | 1 + .../TopicExplorerItemViewModel.cs | 2 +- .../TopicExplorer/TopicExplorerPageView.axaml | 47 +- .../TopicExplorerPageView.axaml.cs | 64 +- .../TopicExplorerPageViewModel.cs | 258 +++- .../TopicExplorerTreeNodeView.axaml.cs | 55 +- .../TopicExplorerTreeNodeViewModel.cs | 9 + Source/Services/Mqtt/MqttClientService.cs | 556 +++++++-- .../Services/Mqtt/StreamConnectedEventArgs.cs | 15 + Source/mqttMultimeter.csproj | 22 +- docs/channel-handling.md | 566 +++++++++ docs/high-throughput-guide.md | 1067 +++++++++++++++++ 36 files changed, 3578 insertions(+), 343 deletions(-) create mode 100644 Source/Extensions/CollectionTrimExtensions.cs create mode 100644 Source/Logging/LogObserverExceptionHandler.cs create mode 100644 Source/Pages/Connection/BoundedChannelFullModeOption.cs create mode 100644 Source/Pages/Connection/MessageProcessingOptionsViewModel.cs create mode 100644 Source/Pages/Connection/MessageProcessingProfile.cs create mode 100644 Source/Services/Mqtt/StreamConnectedEventArgs.cs create mode 100644 docs/channel-handling.md create mode 100644 docs/high-throughput-guide.md diff --git a/Source/App.axaml b/Source/App.axaml index a6f44eb..62cc8e5 100644 --- a/Source/App.axaml +++ b/Source/App.axaml @@ -29,6 +29,7 @@ + diff --git a/Source/App.axaml.cs b/Source/App.axaml.cs index 269a064..330d0d0 100644 --- a/Source/App.axaml.cs +++ b/Source/App.axaml.cs @@ -1,11 +1,14 @@ using System; +using System.Threading.Tasks; using Avalonia; using Avalonia.Controls.ApplicationLifetimes; using Avalonia.Markup.Xaml; using Avalonia.Styling; +using Avalonia.Threading; using Microsoft.Extensions.DependencyInjection; -using mqttMultimeter.Common; +using Microsoft.Extensions.Logging; using mqttMultimeter.Controls; +using mqttMultimeter.Logging; using mqttMultimeter.Main; using mqttMultimeter.Pages.Connection; using mqttMultimeter.Pages.Inflight; @@ -20,6 +23,8 @@ using mqttMultimeter.Services.Mqtt; using mqttMultimeter.Services.State; using mqttMultimeter.Services.Updates; +using ReactiveUI; +using ViewLocator = mqttMultimeter.Common.ViewLocator; namespace mqttMultimeter; @@ -28,9 +33,31 @@ public sealed class App : Application static MainViewModel? _mainViewModel; readonly StateService _stateService; + + readonly ILoggerFactory _loggerFactory; + readonly ILogger _logger; public App() { + + _loggerFactory = LoggerFactory.Create(builder => + { +#if DEBUG + builder.SetMinimumLevel(LogLevel.Debug); + builder.AddDebug(); +#endif + }); + + _logger = _loggerFactory.CreateLogger(); + TaskScheduler.UnobservedTaskException += OnUnobservedTaskException; + Dispatcher.UIThread.UnhandledException += OnUnhandledException; + RxApp.DefaultExceptionHandler = new LogObserverExceptionHandler(_logger); + + this.AttachDeveloperTools(o => + { + o.AddMicrosoftLoggerObservable(_loggerFactory); + }); + var serviceProvider = new ServiceCollection() // Services .AddSingleton() @@ -48,6 +75,8 @@ public App() .AddSingleton() .AddSingleton() .AddSingleton() + .AddTransient(typeof(ILogger<>), typeof(Logger<>)) + .AddSingleton(_loggerFactory) .BuildServiceProvider(); var viewLocator = new ViewLocator(); @@ -56,8 +85,11 @@ public App() serviceProvider.GetRequiredService().EnableUpdateChecks(); _stateService = serviceProvider.GetRequiredService(); _mainViewModel = serviceProvider.GetService(); + + } + public override void Initialize() { AvaloniaXamlLoader.Load(this); @@ -113,4 +145,18 @@ public static void ShowException(Exception exception) _mainViewModel!.OverlayContent = errorBox; } + + + void OnUnhandledException(object sender, DispatcherUnhandledExceptionEventArgs e) + { + _logger.LogError(e.Exception, "Unhandled exception"); + ShowException(e.Exception); + } + + void OnUnobservedTaskException(object? sender, UnobservedTaskExceptionEventArgs e) + { + _logger.LogError(e.Exception, "Unobserved task exception"); + ShowException(e.Exception); + } + } \ No newline at end of file diff --git a/Source/Controls/AutoGrid.cs b/Source/Controls/AutoGrid.cs index ae60c1c..e6e6563 100644 --- a/Source/Controls/AutoGrid.cs +++ b/Source/Controls/AutoGrid.cs @@ -35,14 +35,16 @@ protected override void ChildrenChanged(object? sender, NotifyCollectionChangedE _cellArrangementPending = true; - Dispatcher.UIThread.Post(() => + Dispatcher.UIThread.Post( + state => { - _cellArrangementPending = false; + var grid = (AutoGrid)state!; + grid._cellArrangementPending = false; var currentRow = 0; var currentColumn = 0; - foreach (var child in Children) + foreach (var child in grid.Children) { if (child is Control control) { @@ -65,15 +67,16 @@ protected override void ChildrenChanged(object? sender, NotifyCollectionChangedE } } - if (RowDefinitions.Count - 1 != currentRow) + if (grid.RowDefinitions.Count - 1 != currentRow) { - RowDefinitions.Clear(); + grid.RowDefinitions.Clear(); for (var i = 0; i <= currentRow; i++) { - RowDefinitions.Add(new RowDefinition(GridLength.Auto)); + grid.RowDefinitions.Add(new RowDefinition(GridLength.Auto)); } } }, + this, DispatcherPriority.Render); } } \ No newline at end of file diff --git a/Source/Extensions/CollectionTrimExtensions.cs b/Source/Extensions/CollectionTrimExtensions.cs new file mode 100644 index 0000000..b1a00c0 --- /dev/null +++ b/Source/Extensions/CollectionTrimExtensions.cs @@ -0,0 +1,29 @@ +using System; +using System.Collections.Generic; +using DynamicData; + +namespace mqttMultimeter.Extensions; + +public static class CollectionTrimExtensions +{ + /// + /// Adds items and trims the oldest entries inside a single + /// changeset so the UI receives one update. + /// + public static void AddRangeAndTrim( + this SourceList source, + IList items, + int maxItems, + int trimBatchSize) where T : notnull + { + source.Edit(list => + { + list.AddRange(items); + + if (list.Count > maxItems) + { + list.RemoveRange(0, Math.Min(trimBatchSize, list.Count)); + } + }); + } +} diff --git a/Source/Logging/LogObserverExceptionHandler.cs b/Source/Logging/LogObserverExceptionHandler.cs new file mode 100644 index 0000000..df3c5ea --- /dev/null +++ b/Source/Logging/LogObserverExceptionHandler.cs @@ -0,0 +1,46 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Reactive.Concurrency; +using Microsoft.Extensions.Logging; +using ReactiveUI; + +namespace mqttMultimeter.Logging; + +public class LogObserverExceptionHandler(ILogger logger) : IObserver +{ + public void OnNext(Exception value) + { + if (Debugger.IsAttached) + Debugger.Break(); + + logger.LogError(value, "MyRxHandler Error"); + RxSchedulers.MainThreadScheduler.Schedule(() => + { + throw value; + }); + } + + public void OnError(Exception error) + { + if (Debugger.IsAttached) + Debugger.Break(); + + logger.LogError(error, "MyRxHandler OnError"); + + RxSchedulers.MainThreadScheduler.Schedule(() => + { + throw error; + }); + } + + public void OnCompleted() + { + if (Debugger.IsAttached) + Debugger.Break(); + RxSchedulers.MainThreadScheduler.Schedule(() => + { + throw new NotImplementedException(); + }); + } +} \ No newline at end of file diff --git a/Source/Main/MainView.axaml b/Source/Main/MainView.axaml index b36851b..f71ce62 100644 --- a/Source/Main/MainView.axaml +++ b/Source/Main/MainView.axaml @@ -9,7 +9,7 @@ - + + + @@ -113,27 +122,85 @@ Classes.status_bar_not_connected="{Binding !ConnectionPage.IsConnected}" Grid.Row="1"> - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + - + VerticalAlignment="Center" + Text="{CompiledBinding ConnectionPage.DisconnectedReason.Reason}" + IsVisible="{CompiledBinding !ConnectionPage.IsConnected}" /> + - + - - + + + + + + + diff --git a/Source/Main/MainView.axaml.cs b/Source/Main/MainView.axaml.cs index 2c68f7c..220a602 100644 --- a/Source/Main/MainView.axaml.cs +++ b/Source/Main/MainView.axaml.cs @@ -1,5 +1,6 @@ using System; using Avalonia.Controls; +using Avalonia.Controls.Primitives; using Avalonia.Input; using Avalonia.Markup.Xaml; @@ -54,6 +55,22 @@ void OnDataContextChanged(object? sender, EventArgs e) viewModel.ActivatePageRequested += OnActivatePageRequested; } + async void OnConnectionToggleClicked(object? sender, Avalonia.Interactivity.RoutedEventArgs e) + { + var viewModel = (MainViewModel?)DataContext; + if (viewModel == null) + { + return; + } + + await viewModel.ToggleConnectionAsync().ConfigureAwait(true); + + if (sender is ToggleButton toggleButton) + { + toggleButton.IsChecked = viewModel.ConnectionPage.IsConnected; + } + } + void OnUpdateAvailableNotificationPressed(object? _, PointerPressedEventArgs __) { ((MainViewModel)DataContext!).InfoPage.OpenReleasesUrl(); diff --git a/Source/Main/MainViewModel.cs b/Source/Main/MainViewModel.cs index e774f98..34e1f2a 100644 --- a/Source/Main/MainViewModel.cs +++ b/Source/Main/MainViewModel.cs @@ -1,4 +1,5 @@ using System; +using System.Threading.Tasks; using Avalonia.Threading; using mqttMultimeter.Common; using mqttMultimeter.Pages.Connection; @@ -10,6 +11,7 @@ using mqttMultimeter.Pages.Subscriptions; using mqttMultimeter.Pages.TopicExplorer; using mqttMultimeter.Services.Mqtt; +using MQTTnet; using ReactiveUI; namespace mqttMultimeter.Main; @@ -18,7 +20,13 @@ public sealed class MainViewModel : BaseViewModel { readonly MqttClientService _mqttClientService; - int _counter; + DispatcherTimer? _counterTimer; + long _receivedCounter; + long _notifiedCounter; + long _bufferedCounter; + long _droppedCounter; + bool _showDroppedCounter; + bool _hasDroppedMessages; object? _overlayContent; public MainViewModel(ConnectionPageViewModel connectionPage, @@ -45,19 +53,50 @@ public MainViewModel(ConnectionPageViewModel connectionPage, InflightPage.RepeatMessageRequested += item => PublishPage.RepeatMessage(item); topicExplorerPage.RepeatMessageRequested += item => PublishPage.RepeatMessage(item); - // Update the counter with a timer. There is no need to trigger a binding - // for each counter increment. - DispatcherTimer.Run(UpdateCounter, TimeSpan.FromSeconds(1)); + // Update counters at the same frequency as the message batch buffer. + // The observable is created on connect and disposed on disconnect. + _mqttClientService.MessageStreamConnected += OnCounterStreamConnected; + _mqttClientService.MessageStreamDisconnected += OnCounterStreamDisconnected; } public event EventHandler? ActivatePageRequested; public ConnectionPageViewModel ConnectionPage { get; } - public int Counter + public long ReceivedCounter { - get => _counter; - set => this.RaiseAndSetIfChanged(ref _counter, value); + get => _receivedCounter; + set => this.RaiseAndSetIfChanged(ref _receivedCounter, value); + } + + public long NotifiedCounter + { + get => _notifiedCounter; + set => this.RaiseAndSetIfChanged(ref _notifiedCounter, value); + } + + public long BufferedCounter + { + get => _bufferedCounter; + set => this.RaiseAndSetIfChanged(ref _bufferedCounter, value); + } + + public long DroppedCounter + { + get => _droppedCounter; + set => this.RaiseAndSetIfChanged(ref _droppedCounter, value); + } + + public bool ShowDroppedCounter + { + get => _showDroppedCounter; + set => this.RaiseAndSetIfChanged(ref _showDroppedCounter, value); + } + + public bool HasDroppedMessages + { + get => _hasDroppedMessages; + set => this.RaiseAndSetIfChanged(ref _hasDroppedMessages, value); } public InflightPageViewModel InflightPage { get; } @@ -86,9 +125,72 @@ TPage AttachEvents(TPage page) where TPage : BasePageViewModel return page; } - bool UpdateCounter() + void OnCounterStreamConnected(StreamConnectedEventArgs args) + { + if (_counterTimer is not null) + { + _counterTimer.Tick -= OnCounterTimerTick; + _counterTimer.Stop(); + } + + _counterTimer = new DispatcherTimer(DispatcherPriority.Render) + { + Interval = TimeSpan.FromMilliseconds(args.CounterUpdateMs) + }; + _counterTimer.Tick += OnCounterTimerTick; + _counterTimer.Start(); + } + + void OnCounterStreamDisconnected() + { + if (_counterTimer is not null) + { + _counterTimer.Tick -= OnCounterTimerTick; + _counterTimer.Stop(); + _counterTimer = null; + } + + // Final update to flush the last counter values. + UpdateCounter(); + } + + void OnCounterTimerTick(object? sender, EventArgs e) => UpdateCounter(); + + void UpdateCounter() + { + ReceivedCounter = _mqttClientService.ReceivedMessagesCount; + NotifiedCounter = _mqttClientService.NotifiedMessagesCount; + BufferedCounter = _mqttClientService.BufferedMessagesCount; + DroppedCounter = _mqttClientService.DroppedMessagesCount; + HasDroppedMessages = DroppedCounter > 0; + + var selectedItem = ConnectionPage.Items.SelectedItem; + if (selectedItem == null) + { + ShowDroppedCounter = false; + } + else + { + var options = selectedItem.MessageProcessingOptions; + ShowDroppedCounter = options.UseBoundedChannel && options.SelectedFullMode.Value != System.Threading.Channels.BoundedChannelFullMode.Wait; + } + } + + public async Task ToggleConnectionAsync() { - Counter = _mqttClientService.ReceivedMessagesCount; - return true; + var selectedItem = ConnectionPage.Items.SelectedItem; + if (selectedItem == null || ConnectionPage.IsConnecting) + { + return; + } + + if (ConnectionPage.IsConnected) + { + await selectedItem.Disconnect().ConfigureAwait(true); + } + else + { + await selectedItem.Connect().ConfigureAwait(true); + } } } \ No newline at end of file diff --git a/Source/Pages/Connection/BoundedChannelFullModeOption.cs b/Source/Pages/Connection/BoundedChannelFullModeOption.cs new file mode 100644 index 0000000..cad4d0b --- /dev/null +++ b/Source/Pages/Connection/BoundedChannelFullModeOption.cs @@ -0,0 +1,23 @@ +using System.Collections.Generic; +using System.Threading.Channels; + +namespace mqttMultimeter.Pages.Connection; + +/// +/// Data model for a bounded-channel full-mode option. +/// Replaces EnumViewModel<BoundedChannelFullMode> so the display +/// name lives next to the value without a generic wrapper. +/// +public sealed record BoundedChannelFullModeOption(string DisplayName, BoundedChannelFullMode Value) +{ + /// All available full-mode options. + public static IReadOnlyList All { get; } = + [ + new("Wait", BoundedChannelFullMode.Wait), + new("Drop newest", BoundedChannelFullMode.DropNewest), + new("Drop oldest", BoundedChannelFullMode.DropOldest), + new("Drop write", BoundedChannelFullMode.DropWrite) + ]; + + public override string ToString() => DisplayName; +} diff --git a/Source/Pages/Connection/ConnectionItemView.axaml b/Source/Pages/Connection/ConnectionItemView.axaml index c2423f6..1e52ce5 100644 --- a/Source/Pages/Connection/ConnectionItemView.axaml +++ b/Source/Pages/Connection/ConnectionItemView.axaml @@ -33,6 +33,7 @@ Classes="image_button" MinWidth="100" Margin="0,0,10,0" + IsEnabled="{Binding OwnerPage.CanConnect}" Command="{Binding Connect, Mode=OneTime}"> @@ -41,6 +42,7 @@