using System; using System.IO; using System.Threading.Tasks; namespace FluentTerminal.SystemTray { internal sealed class BufferedReader : IDisposable { private const int MaxTotalDelayMilliseconds = 100; private const int WaitPeriodMilliseconds = 30; private const int NearReadsPeriodMilliseconds = 20; private const int NearReadsBufferingTrigger = 3; private const int BufferSize = 204800; private readonly object _lock = new object(); private readonly Stream _stream; private readonly Action _callback; private readonly bool _enableBuffer; private bool _disposed; private bool _paused; private byte[] _buffer; private int _bufferIndex; private DateTime _lastRead; private DateTime _sendingDeadline; private DateTime _scheduledSend; private int _nearReadingsCount; private Task _sendingTask; internal BufferedReader(Stream stream, Action callback, bool enableBuffer) { _stream = stream; _callback = callback; _enableBuffer = enableBuffer; // ReSharper disable once AssignmentIsFullyDiscarded _ = Task.Factory.StartNew(ReadingLoop, TaskCreationOptions.LongRunning); } internal void SetPaused(bool value) { lock (_lock) _paused = value; } public void Dispose() { lock (_lock) _disposed = true; } private async Task ReadingLoop() { while (true) { // Allow CPU to jump between TerminalSessions' ReadingLoop Tasks await Task.Delay(1).ConfigureAwait(false); bool paused; lock (_lock) { if (_disposed) return; paused = _paused; } if (paused) { await Task.Delay(WaitPeriodMilliseconds).ConfigureAwait(false); continue; } var currentBuffer = new byte[BufferSize]; int read; try { read = await _stream.ReadAsync(currentBuffer, 0, currentBuffer.Length).ConfigureAwait(false); } catch { read = 0; } if (read < 1) { // Expected to happen only when terminal is closed. // Probably not recoverable, but we'll wait anyway 'till disposed. await Task.Delay(50).ConfigureAwait(false); continue; } if (!_enableBuffer) { // Buffering disabled. Just send. _buffer = currentBuffer; _bufferIndex = read; SendBuffer(); continue; } var now = DateTime.UtcNow; lock (_lock) { if (_buffer != null) { // We're already in buffered mode if (_bufferIndex + read > BufferSize) { // No room in the buffer. Have to flush it. SendBuffer(); _buffer = currentBuffer; _bufferIndex = 0; _scheduledSend = now.AddMilliseconds(WaitPeriodMilliseconds); _sendingDeadline = now.AddMilliseconds(MaxTotalDelayMilliseconds); } else { // Copy to existing buffer Buffer.BlockCopy(currentBuffer, 0, _buffer, _bufferIndex, read); _bufferIndex += read; _scheduledSend = now.AddMilliseconds(WaitPeriodMilliseconds); } if (now.Subtract(_lastRead).TotalMilliseconds < NearReadsPeriodMilliseconds) { // We should stop buffered mode SendBuffer(); } _lastRead = now; continue; } if (now.Subtract(_lastRead).TotalMilliseconds < NearReadsPeriodMilliseconds) { _nearReadingsCount++; } else { _nearReadingsCount = 0; } _lastRead = now; if (_nearReadingsCount >= NearReadsBufferingTrigger) { // We should enter buffered mode _buffer = currentBuffer; _bufferIndex = 0; _sendingDeadline = now.AddMilliseconds(MaxTotalDelayMilliseconds); _scheduledSend = now.AddMilliseconds(WaitPeriodMilliseconds); if (_sendingTask == null) _sendingTask = SendAsync(); _nearReadingsCount = 0; continue; } // Not in buffering mode. Just send. _buffer = currentBuffer; _bufferIndex = read; SendBuffer(); } } } private async Task SendAsync() { // Just to release the calling thread asap await Task.Delay(5).ConfigureAwait(false); while (true) { TimeSpan sleep; lock (_lock) { if (_buffer == null) { _sendingTask = null; return; } if (_paused) { sleep = TimeSpan.FromMilliseconds(WaitPeriodMilliseconds); } else { sleep = _scheduledSend < _sendingDeadline ? _scheduledSend.Subtract(DateTime.UtcNow) : _sendingDeadline.Subtract(DateTime.UtcNow); if (sleep.TotalMilliseconds < 5) { // Time to send SendBuffer(); _sendingTask = null; return; } } } await Task.Delay(sleep).ConfigureAwait(false); } } // Has to be called from a locked code! private void SendBuffer() { if (_bufferIndex == _buffer.Length) { _callback(_buffer); } else { var newBuffer = new byte[_bufferIndex]; Buffer.BlockCopy(_buffer, 0, newBuffer, 0, _bufferIndex); _callback(newBuffer); } _buffer = null; _bufferIndex = 0; } } }