replaced buffer handling thread with timer in CommunicationProtocol,

therefore implemented RecurringTimer,
implemented ICloneable for ExtendedByte,
implemented second stage queue to buffer data in TerminalViewModel,
updating data in UI in a separate task,
fixed warnings
master
Jonas Arnold 3 years ago
parent bc33e30083
commit df681d6189
  1. 51
      Common/Helpers/RecurringTimer.cs
  2. 5
      MultiTerm.Core/ViewModel/MultiFormatDataViewModel.cs
  3. 73
      MultiTerm.Core/ViewModel/TerminalViewModel.cs
  4. 92
      MultiTerm.Protocols/CommunicationProtocol.cs
  5. 9
      MultiTerm.Protocols/Model/ExtendedByte.cs

@ -0,0 +1,51 @@
namespace Common.Helpers;
/// <summary>
/// A <see cref="System.Timers.Timer"/> that recurrs after a given interval.
/// The callback is non-reentrant, unline a standard <see cref="System.Timers.Timer"/>.
/// </summary>
public class RecurringTimer : IDisposable
{
private readonly System.Timers.Timer _timer;
/// <summary>
/// Instanciates the recurring timer, with the <paramref name="intervalMs"/> given.
/// </summary>
/// <param name="intervalMs">interval to act on callback, in milliseconds</param>
/// <param name="callbackAction">action to perform when the timer elapses</param>
public RecurringTimer(int intervalMs, Action callbackAction)
{
_timer = new System.Timers.Timer()
{
AutoReset = false,
Interval = intervalMs
};
_timer.Elapsed += delegate
{
callbackAction(); // perform callback action
_timer.Start(); // manual restart after action finished
};
}
/// <summary>
/// Starts the timing by setting <see cref='System.Timers.Timer.Enabled'/> to <see langword='true'/>.
/// </summary>
public void Start()
{
_timer.Start();
}
/// <summary>
/// Stops the timing by setting <see cref='System.Timers.Timer.Enabled'/> to <see langword='false'/>.
/// </summary>
public void Stop()
{
_timer.Stop();
}
public void Dispose()
{
_timer?.Dispose();
}
}

@ -16,8 +16,7 @@ public partial class MultiFormatDataViewModel : ObservableObject
private int dataCharacterCount = 0; private int dataCharacterCount = 0;
private List<byte>? listOfPreviousCharacters = null; private List<byte>? listOfPreviousCharacters = null;
#region ICommunicationDataViewModel Implementation #region Observable Properties
[ObservableProperty] [ObservableProperty]
private ObservableCollection<ByteDataViewModel> data = new(); private ObservableCollection<ByteDataViewModel> data = new();
@ -40,7 +39,7 @@ public partial class MultiFormatDataViewModel : ObservableObject
/// <summary> /// <summary>
/// Newline Sequence string inserted on every newline in the <see cref="DataAsString"/>. /// Newline Sequence string inserted on every newline in the <see cref="DataAsString"/>.
/// </summary> /// </summary>
public static string NewlineSequence = Environment.NewLine; public static readonly string NewlineSequence = Environment.NewLine;
public MultiFormatDataViewModel(IContext context) public MultiFormatDataViewModel(IContext context)
{ {

@ -1,5 +1,4 @@
using Common; using Common.AppSettings;
using Common.AppSettings;
using Common.Helpers; using Common.Helpers;
using Common.Messaging; using Common.Messaging;
using CommunityToolkit.Mvvm.ComponentModel; using CommunityToolkit.Mvvm.ComponentModel;
@ -9,6 +8,7 @@ using MultiTerm.Core.Types;
using MultiTerm.Protocols; using MultiTerm.Protocols;
using MultiTerm.Protocols.Model; using MultiTerm.Protocols.Model;
using MultiTerm.Protocols.Types; using MultiTerm.Protocols.Types;
using System.Collections.Concurrent;
namespace MultiTerm.Core.ViewModel; namespace MultiTerm.Core.ViewModel;
@ -19,6 +19,11 @@ public abstract partial class TerminalViewModel : ObservableObject, ITerminalVie
private readonly IAppSettingsProvider appSettings; private readonly IAppSettingsProvider appSettings;
private readonly IMessenger messenger; private readonly IMessenger messenger;
private readonly RecurringTimer displayNewDataTimer; // Timer for data updates
private const int displayNewDataTimerIntervalMs = 20; // after x milliseconds new data is fed forward to the UI
private readonly ConcurrentQueue<IEnumerable<ExtendedByte>> receivedDataQueue = new(); // Queue to buffer received data from communication protocol
private readonly ConcurrentQueue<IEnumerable<ExtendedByte>> sentDataQueue = new(); // Queue to buffer sent data (as reported from communication protocol)
public abstract TerminalViewType ViewType { get; } public abstract TerminalViewType ViewType { get; }
private ProtocolType protocolType; private ProtocolType protocolType;
@ -96,10 +101,14 @@ public abstract partial class TerminalViewModel : ObservableObject, ITerminalVie
{ {
this.appSettings = appSettings; this.appSettings = appSettings;
this.messenger = messenger; this.messenger = messenger;
this.displayNewDataTimer = new RecurringTimer(displayNewDataTimerIntervalMs, delegate() { this.UpdateDisplayedData(); });
this.displayNewDataTimer.Start();
} }
~TerminalViewModel() ~TerminalViewModel()
{ {
this.displayNewDataTimer?.Stop();
if (this.CommunicationProtocol != null) if (this.CommunicationProtocol != null)
{ {
// disconnect communication protocol // disconnect communication protocol
@ -128,47 +137,61 @@ public abstract partial class TerminalViewModel : ObservableObject, ITerminalVie
this.CommunicationProtocol.ConnectionStateChangedEvent += CommunicationProtocol_ConnectionStateChangedEvent; this.CommunicationProtocol.ConnectionStateChangedEvent += CommunicationProtocol_ConnectionStateChangedEvent;
} }
// initializes default newline separators, updates them directly inside ReceivedData and SentData objects // initializes default newline separators, updates them directly
this.InitializeNewlineSeparatorsFromAppSettings(); this.InitializeNewlineSeparatorsFromAppSettings();
}
private void CommunicationProtocol_ConnectionStateChangedEvent(object? sender, ProtocolConnectionState e)
{
// update internal state
this.CommunicationProtocolConnectionState = e;
this.CommunicationProtocolLongIdentifier = this.CommunicationProtocol!.LongInstanceIdentifier;
// update newline settings in data objects // If ViewModel still displays connected after unintentional disconnect => force to disconnected state
// TODO!!! // allows user to change settings
//this.ReceivedData.NewlineSeparator = this.DataDisplayNewlineSeparatorType; if (e == ProtocolConnectionState.UnintentionallyDisconnected && this.ProtocolSettings!.DisplaysConnected == true)
//this.SentData.NewlineSeparator = this.DataDisplayNewlineSeparatorType; {
this.ProtocolSettings.ForceConnectedState(false);
}
} }
#region Communication Protocol Data Handling
/// <summary>
/// Add newly received data to display.
/// </summary>
public abstract void DisplayNewReceivedData(IEnumerable<ExtendedByte> newData); public abstract void DisplayNewReceivedData(IEnumerable<ExtendedByte> newData);
/// <summary>
/// Add newly sent data to display.
/// </summary>
public abstract void DisplayNewSentData(IEnumerable<ExtendedByte> newData);
private void CommunicationProtocol_ReceivedDataEvent(object? sender, ReceivedDataEventArgs e) private void CommunicationProtocol_ReceivedDataEvent(object? sender, ReceivedDataEventArgs e)
{ {
// handover data this.receivedDataQueue.Enqueue(e.ReceivedData);
this.DisplayNewReceivedData(e.ReceivedData);
} }
public abstract void DisplayNewSentData(IEnumerable<ExtendedByte> newData);
private void CommunicationProtocol_SentDataEvent(object? sender, SentDataEventArgs e) private void CommunicationProtocol_SentDataEvent(object? sender, SentDataEventArgs e)
{ {
// handover data this.sentDataQueue.Enqueue(e.SentData);
this.DisplayNewSentData(e.SentData);
} }
private void CommunicationProtocol_ConnectionStateChangedEvent(object? sender, ProtocolConnectionState e) private async void UpdateDisplayedData()
{ {
// update internal state // display new data in seperate tasks
this.CommunicationProtocolConnectionState = e; if(this.receivedDataQueue.TryDequeue(out var newReceivedData))
this.CommunicationProtocolLongIdentifier = this.CommunicationProtocol!.LongInstanceIdentifier;
// If ViewModel still displays connected after unintentional disconnect => force to disconnected state
// allows user to change settings
if (e == ProtocolConnectionState.UnintentionallyDisconnected && this.ProtocolSettings!.DisplaysConnected == true)
{ {
this.ProtocolSettings.ForceConnectedState(false); await Task.Run(() => this.DisplayNewReceivedData(newReceivedData));
}
if(this.sentDataQueue.TryDequeue(out var newSentData))
{
await Task.Run(() => this.DisplayNewSentData(newSentData));
} }
} }
#endregion #endregion
#endregion
/// <summary> /// <summary>
/// Sends the given data to this instances <see cref="CommunicationProtocol"/>. /// Sends the given data to this instances <see cref="CommunicationProtocol"/>.
/// Appends configured <see cref="SendNewlineSeparatorType"/> to the end of string. /// Appends configured <see cref="SendNewlineSeparatorType"/> to the end of string.
@ -294,10 +317,16 @@ public abstract partial class TerminalViewModel : ObservableObject, ITerminalVie
catch { } catch { }
} }
/// <summary>
/// Act when DisplayData NewlineSeparator Changed.
/// For example rearrange data according to new newline separator.
/// </summary>
/// <param name="newType"></param>
protected abstract void ActOnDataDisplayNewlineSeparatorTypeChanged(NewlineSeparatorType newType); protected abstract void ActOnDataDisplayNewlineSeparatorTypeChanged(NewlineSeparatorType newType);
partial void OnDataDisplayNewlineSeparatorTypeChanged(NewlineSeparatorType value) partial void OnDataDisplayNewlineSeparatorTypeChanged(NewlineSeparatorType value)
{ {
// inform derivatives
this.ActOnDataDisplayNewlineSeparatorTypeChanged(value); this.ActOnDataDisplayNewlineSeparatorTypeChanged(value);
} }
#endregion #endregion

@ -14,7 +14,9 @@ public abstract class CommunicationProtocol : ICommunicationProtocol
protected readonly IMessenger messenger; protected readonly IMessenger messenger;
private CancellationTokenSource cancellationTokenSource; private CancellationTokenSource cancellationTokenSource;
private Thread? readingThread; private Thread? readingThread;
private Thread? bufferHandlingThread; private RecurringTimer? bufferHandlingTimer;
private const int bufferHandlingTimerIntervalMs = 20; // After x milliseconds the timer shall be called
private const int eventBlockSize = 50; // Block size for Received/Sent Data events
private ConcurrentQueue<ExtendedByte>? receivedDataQueue; private ConcurrentQueue<ExtendedByte>? receivedDataQueue;
private ConcurrentQueue<ExtendedByte>? sentDataQueue; private ConcurrentQueue<ExtendedByte>? sentDataQueue;
@ -63,7 +65,7 @@ public abstract class CommunicationProtocol : ICommunicationProtocol
// perform all steps in a task, to prevent deadlock when this method is called from a thread that is joined. // perform all steps in a task, to prevent deadlock when this method is called from a thread that is joined.
Task.Run(() => Task.Run(() =>
{ {
this.CancelThreads(); this.CancelOngoingOperations();
this.logger.LogError($"'{nameof(OnUnintentionallyDisconnected)}()' called.", nameof(CommunicationProtocol)); this.logger.LogError($"'{nameof(OnUnintentionallyDisconnected)}()' called.", nameof(CommunicationProtocol));
// update state // update state
@ -143,8 +145,8 @@ public abstract class CommunicationProtocol : ICommunicationProtocol
this.receivedDataQueue = new ConcurrentQueue<ExtendedByte>(); this.receivedDataQueue = new ConcurrentQueue<ExtendedByte>();
this.sentDataQueue = new ConcurrentQueue<ExtendedByte>(); this.sentDataQueue = new ConcurrentQueue<ExtendedByte>();
// start interal buffer handling thread // start interal buffer handling thread
this.bufferHandlingThread = new Thread(() => this.HandleDataQueues(this.cancellationTokenSource.Token)); this.bufferHandlingTimer = new(bufferHandlingTimerIntervalMs, delegate() { this.HandleDataQueues(); });
this.bufferHandlingThread.Start(); this.bufferHandlingTimer.Start();
// start internal reading thread // start internal reading thread
this.readingThread = new Thread(() => this.InternalRead(this.cancellationTokenSource.Token)); this.readingThread = new Thread(() => this.InternalRead(this.cancellationTokenSource.Token));
this.readingThread.Start(); this.readingThread.Start();
@ -166,69 +168,85 @@ public abstract class CommunicationProtocol : ICommunicationProtocol
protected abstract void InternalDisconnect(); protected abstract void InternalDisconnect();
public void Disconnect() public void Disconnect()
{ {
this.CancelThreads(); this.CancelOngoingOperations();
this.InternalDisconnect(); this.InternalDisconnect();
this.State = ProtocolConnectionState.NotConnected; this.State = ProtocolConnectionState.NotConnected;
} }
/// <summary> /// <summary>
/// Sets cancellation token and makes sure both threads do not run anymore. /// Sets cancellation token and makes sure ongoing operations do not run anymore.
/// </summary> /// </summary>
private void CancelThreads() private void CancelOngoingOperations()
{ {
this.cancellationTokenSource.Cancel(); this.cancellationTokenSource.Cancel();
// join both threads if they exist // stop thread and timer
this.readingThread?.Join(); this.readingThread?.Join();
this.bufferHandlingThread?.Join(); this.bufferHandlingTimer?.Stop();
} }
/// <summary> /// <summary>
/// Internal handling of internal data queues. /// Internal handling of internal data queues.
/// Arranges characters into blocks and raises <see cref="ReceivedDataEvent"/> and <see cref="SentDataEvent"/>.
/// </summary> /// </summary>
/// <param name="ct">cancellation token</param> private void HandleDataQueues()
private void HandleDataQueues(CancellationToken ct)
{ {
while(ct.IsCancellationRequested == false) bool receivedDataAvailable = this.receivedDataQueue != null && this.receivedDataQueue.TryPeek(out _);
bool sentDataAvailable = this.sentDataQueue != null && this.sentDataQueue.TryPeek(out _);
List<ExtendedByte> internalList = new();
List<ExtendedByte> duplicateInternalList()
{ {
bool receivedDataAvailable = this.receivedDataQueue != null && this.receivedDataQueue.TryPeek(out _); // shallow copy of data
bool sentDataAvailable = this.sentDataQueue != null && this.sentDataQueue.TryPeek(out _); var duplicatedList = internalList.Select(item => (ExtendedByte)item.Clone()).ToList();
// clear internal list
internalList.Clear();
return duplicatedList;
}
// collect received data // collect received data
if (receivedDataAvailable) if (receivedDataAvailable)
{
// while data is available
while (this.receivedDataQueue!.TryDequeue(out ExtendedByte? newReceivedByte))
{ {
// something is in the queue => block thread and try dequeue // fill it into the internal list
ExtendedByte? newReceivedData; internalList.Add(newReceivedByte);
while (this.receivedDataQueue!.TryDequeue(out newReceivedData) == false) if(internalList.Count > eventBlockSize)
{ {
Thread.Sleep(1); // raise event with block of data
this.ReceivedDataEvent?.Invoke(this, new ReceivedDataEventArgs(duplicateInternalList()));
} }
// raise event
this.ReceivedDataEvent?.Invoke(this, new ReceivedDataEventArgs(new ExtendedByte[] { newReceivedData }));
} }
// no more new data but still some data in the list => send it
if (internalList.Count > 0)
{
this.ReceivedDataEvent?.Invoke(this, new ReceivedDataEventArgs(duplicateInternalList()));
}
// reset list for sent data
internalList = new();
}
// collect sent data // collect sent data
if (sentDataAvailable) if (sentDataAvailable)
{
// while data is available
while (this.sentDataQueue!.TryDequeue(out ExtendedByte? newSentByte))
{ {
// something is in the queue => block thread and try dequeue // fill it into the internal list
ExtendedByte? newSentData; internalList.Add(newSentByte);
while (this.sentDataQueue!.TryDequeue(out newSentData) == false) if (internalList.Count > eventBlockSize)
{ {
Thread.Sleep(1); // raise event with block of data
this.SentDataEvent?.Invoke(this, new SentDataEventArgs(duplicateInternalList()));
} }
// raise event
this.SentDataEvent?.Invoke(this, new SentDataEventArgs(new ExtendedByte[] { newSentData }));
} }
// no more new data but still some data in the list => send it
// generally wait some time to slow down thread, if no data available if (internalList.Count > 0)
if (!receivedDataAvailable && !sentDataAvailable)
{ {
Thread.Sleep(5); this.SentDataEvent?.Invoke(this, new SentDataEventArgs(duplicateInternalList()));
} }
// always sleep 1ms to keep UI rolling
Thread.Sleep(1);
} }
} }

@ -7,7 +7,7 @@ namespace MultiTerm.Protocols.Model;
/// A time can be stored in combination with this <see cref="byte"/> using the <see cref="Time"/> property. E.g. to represent arrived or sent time. /// A time can be stored in combination with this <see cref="byte"/> using the <see cref="Time"/> property. E.g. to represent arrived or sent time.
/// Several methods to display the <see cref="byte"/> in other formats are provided. /// Several methods to display the <see cref="byte"/> in other formats are provided.
/// </summary> /// </summary>
public partial class ExtendedByte public partial class ExtendedByte : ICloneable
{ {
/// <summary> /// <summary>
/// Data in the form of a byte. /// Data in the form of a byte.
@ -127,4 +127,11 @@ public partial class ExtendedByte
return resultString; return resultString;
} }
#region ICloneable
public object Clone()
{
return this.MemberwiseClone();
}
#endregion
} }

Loading…
Cancel
Save