diff --git a/Source/App.axaml b/Source/App.axaml index a6f44eb..6b9572e 100644 --- a/Source/App.axaml +++ b/Source/App.axaml @@ -29,6 +29,7 @@ + @@ -60,6 +61,20 @@ Value="{DynamicResource ControlContentThemeFontSize}" /> + + + + + @@ -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/MainViewModel.cs b/Source/Main/MainViewModel.cs index e774f98..8663bef 100644 --- a/Source/Main/MainViewModel.cs +++ b/Source/Main/MainViewModel.cs @@ -1,4 +1,7 @@ using System; +using System.Reactive; +using System.Reactive.Linq; +using System.Threading.Tasks; using Avalonia.Threading; using mqttMultimeter.Common; using mqttMultimeter.Pages.Connection; @@ -10,6 +13,7 @@ using mqttMultimeter.Pages.Subscriptions; using mqttMultimeter.Pages.TopicExplorer; using mqttMultimeter.Services.Mqtt; +using MQTTnet; using ReactiveUI; namespace mqttMultimeter.Main; @@ -18,7 +22,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 +55,54 @@ 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; + + ToggleConnectionCommand = ReactiveCommand.CreateFromTask(ToggleConnectionAsync); } public event EventHandler? ActivatePageRequested; + public ReactiveCommand ToggleConnectionCommand { get; } + public ConnectionPageViewModel ConnectionPage { get; } - public int Counter + public long ReceivedCounter + { + 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 => _counter; - set => this.RaiseAndSetIfChanged(ref _counter, value); + 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 +131,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 @@