<commit_before>﻿using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using homeControl.ClientServerShared;

namespace homeControl.ClientApi.Server
{
    internal sealed class MessageWriterPipeline
    {
        private readonly IClientWriter _writer;
        private readonly IClientMessageSerializer _serializer;
        private readonly ITargetBlock<IClientMessage> _pipeline;
        private readonly CancellationToken _ct;

        public MessageWriterPipeline(IClientWriter writer, IClientMessageSerializer serializer, CancellationToken ct)
        {
            Guard.DebugAssertArgumentNotNull(writer, nameof(writer));
            Guard.DebugAssertArgumentNotNull(serializer, nameof(serializer));

            _writer = writer;
            _serializer = serializer;
            _ct = ct;
            _pipeline = BuildPipeline();
        }

        private ITargetBlock<IClientMessage> BuildPipeline()
        {
            var bufferOptions = new DataflowBlockOptions
            {
                EnsureOrdered = false,
                CancellationToken = _ct,
            };
            var executionOptions = new ExecutionDataflowBlockOptions
            {
                CancellationToken = _ct,
                EnsureOrdered = true,
                MaxDegreeOfParallelism = 1,
                SingleProducerConstrained = true
            };

            var input = new BufferBlock<IClientMessage>(bufferOptions);
            var serializerBlock = new TransformBlock<IClientMessage, byte[]>(m => PerformSerialization(m), executionOptions);
            var weighterBlock = new TransformBlock<byte[], byte[]>(m => AddLength(m), executionOptions);
            var writerBlock = new ActionBlock<byte[]>(WriteMessage, executionOptions);

            input.LinkTo(serializerBlock);
            serializerBlock.LinkTo(weighterBlock);
            weighterBlock.LinkTo(writerBlock);

            return input;
        }

        private Task WriteMessage(byte[] data)
        {
            Guard.DebugAssertArgumentNotNull(data, nameof(data));

            return _writer.WriteAsync(data, _ct);
        }

        private byte[] AddLength(byte[] data)
        {
            Guard.DebugAssertArgumentNotNull(data, nameof(data));

            var length = data.Length;
            return BitConverter.GetBytes(length).Concat(data).ToArray();
        }

        private byte[] PerformSerialization(IClientMessage message)
        {
            Guard.DebugAssertArgumentNotNull(message, nameof(message));

            return _serializer.Serialize(message);
        }

        public void PostMessage(IClientMessage message)
        {
            Guard.DebugAssertArgumentNotNull(message, nameof(message));

            _pipeline.Post(message);
        }
    }
}
<commit_msg>MessageWriterPipeline: propagate completion<commit_after>﻿using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using homeControl.ClientServerShared;

namespace homeControl.ClientApi.Server
{
    internal sealed class MessageWriterPipeline
    {
        private readonly IClientWriter _writer;
        private readonly IClientMessageSerializer _serializer;
        private readonly ITargetBlock<IClientMessage> _pipeline;
        private readonly CancellationToken _ct;

        public MessageWriterPipeline(IClientWriter writer, IClientMessageSerializer serializer, CancellationToken ct)
        {
            Guard.DebugAssertArgumentNotNull(writer, nameof(writer));
            Guard.DebugAssertArgumentNotNull(serializer, nameof(serializer));

            _writer = writer;
            _serializer = serializer;
            _ct = ct;
            _pipeline = BuildPipeline();
        }

        private ITargetBlock<IClientMessage> BuildPipeline()
        {
            var bufferOptions = new DataflowBlockOptions
            {
                EnsureOrdered = false,
                CancellationToken = _ct,
            };
            var executionOptions = new ExecutionDataflowBlockOptions
            {
                CancellationToken = _ct,
                EnsureOrdered = true,
                MaxDegreeOfParallelism = 1,
                SingleProducerConstrained = true
            };

            var input = new BufferBlock<IClientMessage>(bufferOptions);
            var serializerBlock = new TransformBlock<IClientMessage, byte[]>(m => PerformSerialization(m), executionOptions);
            var weighterBlock = new TransformBlock<byte[], byte[]>(m => AddLength(m), executionOptions);
            var writerBlock = new ActionBlock<byte[]>(WriteMessage, executionOptions);

            var linkOptions = new DataflowLinkOptions { PropagateCompletion = true };
            input.LinkTo(serializerBlock, linkOptions);
            serializerBlock.LinkTo(weighterBlock, linkOptions);
            weighterBlock.LinkTo(writerBlock, linkOptions);

            return input;
        }

        private Task WriteMessage(byte[] data)
        {
            Guard.DebugAssertArgumentNotNull(data, nameof(data));

            return _writer.WriteAsync(data, _ct);
        }

        private byte[] AddLength(byte[] data)
        {
            Guard.DebugAssertArgumentNotNull(data, nameof(data));

            var length = data.Length;
            return BitConverter.GetBytes(length).Concat(data).ToArray();
        }

        private byte[] PerformSerialization(IClientMessage message)
        {
            Guard.DebugAssertArgumentNotNull(message, nameof(message));

            return _serializer.Serialize(message);
        }

        public void PostMessage(IClientMessage message)
        {
            Guard.DebugAssertArgumentNotNull(message, nameof(message));

            _pipeline.SendAsync(message);
        }
    }
}

{'fraction_non_alphanumeric': 0.07704133925521012, 'fraction_numerical': 0.00034164673727365904, 'mean_word_length': 2.384393063583815, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2272906', 'n_tokens_mistral': 1408, 'n_tokens_neox': 1416, 'n_words': 289}