using Capnp.FrameTracing; using Microsoft.Extensions.Logging; using System; using System.Collections.Generic; using System.Diagnostics; using System.IO; using System.Net.Sockets; using System.Runtime.InteropServices; using System.Text; using System.Threading; namespace Capnp { /// /// The FramePump handles sending and receiving Cap'n Proto messages over a stream. It exposes a Send method for writing frames to the stream, and an /// event handler for processing received frames. It does not fork any new thread by itself, but instead exposes a synchronous blocking Run method that /// implements the receive loop. Invoke this method in the thread context of your choice. /// public class FramePump: IDisposable { ILogger Logger { get; } = Logging.CreateLogger(); int _disposing; readonly Stream _stream; readonly BinaryWriter _writer; readonly object _writeLock = new object(); readonly List _tracers = new List(); /// /// Constructs a new instance for given stream. /// /// The stream for message I/O. /// If you intend to receive messages, the stream must support reading (CanRead). /// If you intend to send messages, the stream must support writing (CanWrite). /// /// is null. public FramePump(Stream stream) { _stream = stream ?? throw new ArgumentNullException(nameof(stream)); if (stream.CanWrite) _writer = new BinaryWriter(_stream); } /// /// Disposes this instance and the underlying stream. This will also cause the Run method to return. /// public void Dispose() { if (0 == Interlocked.Exchange(ref _disposing, 1)) { foreach (var tracer in _tracers) { tracer.Dispose(); } _writer?.Dispose(); _stream.Dispose(); } } /// /// Event handler for frame reception. /// public event Action FrameReceived; /// /// Sends a message over the stream. /// /// Message to be sent /// The underlying stream does not support writing. /// The message does not provide at least one segment, or one of its segments is empty. /// An I/O error occurs. /// This instance or stream is diposed. public void Send(WireFrame frame) { if (_writer == null) throw new InvalidOperationException("Stream is not writable"); if (frame.Segments == null) throw new ArgumentException("Do not pass default(WireFrame)"); if (frame.Segments.Count == 0) throw new ArgumentException("Expected at least one segment"); foreach (var segment in frame.Segments) { if (segment.Length == 0) throw new ArgumentException("Segment must not have zero length"); } lock (_writeLock) { foreach (var tracer in _tracers) { tracer.TraceFrame(FrameDirection.Tx, frame); } _writer.Write(frame.Segments.Count - 1); foreach (var segment in frame.Segments) { _writer.Write(segment.Length); } if ((frame.Segments.Count & 1) == 0) { // Padding _writer.Write(0); } foreach (var segment in frame.Segments) { #if NETSTANDARD2_0 var bytes = MemoryMarshal.Cast(segment.Span).ToArray(); #else var bytes = MemoryMarshal.Cast(segment.Span); #endif _writer.Write(bytes); } } } long _waitingForData; /// /// Whether the pump is currently waiting for data to receive. /// public bool IsWaitingForData { get => Interlocked.Read(ref _waitingForData) != 0; private set => Interlocked.Exchange(ref _waitingForData, value ? 1 : 0); } /// /// Synchronously runs the frame reception loop. Will only return after calling Dispose() or upon error condition. /// The method does not propagate EndOfStreamException or ObjectDisposedException to the caller, since these conditions are considered /// to be part of normal operation. It does pass exceptions which arise due to I/O errors or invalid data. /// /// The underlying stream does not support reading or is already closed. /// Encountered Invalid Framing Data /// Received a message with too many or too big segments, probably dues to invalid data. /// An I/O error occurs. public void Run() { try { using (var reader = new BinaryReader(_stream, Encoding.Default)) { while (true) { IsWaitingForData = true; var frame = reader.ReadWireFrame(); IsWaitingForData = false; foreach (var tracer in _tracers) { tracer.TraceFrame(FrameDirection.Rx, frame); } FrameReceived?.Invoke(frame); } } } catch (EndOfStreamException) { Logger.LogWarning("Encountered End of Stream"); } catch (InvalidDataException e) { Logger.LogWarning(e.Message); } catch (ObjectDisposedException) { } catch (Exception exception) { // When disposing, all kinds of errors might happen here, // not worth logging. if (_disposing == 0) { Logger.LogWarning(exception.Message); } } finally { IsWaitingForData = false; } } public void AttachTracer(IFrameTracer tracer) { _tracers.Add(tracer); } } }