<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using ProtoBuf;
using Shoutr.Contracts;
using Shoutr.Reactive;
using Shoutr.Serialization;

namespace Shoutr
{
    public class Listener : IListener
    {
        public async Task Listen(int port = Defaults.Port,
            CancellationToken cancellationToken = default)
        {
            var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            var token = cancellationTokenSource.Token;

            var packetBufferObservable = new Subject<byte[]>();
            var observableScheduler =
                System.Reactive.Concurrency.Scheduler.Default; //new TaskPoolScheduler(new TaskFactory(token));
            var protoMessageObservable = packetBufferObservable
                .ObserveOn(observableScheduler)
                .Select(bytes =>
                {
                    ProtoMessage message;
                    using (var memoryStream = new MemoryStream(bytes))
                    {
                        var protoMessage = Serializer.Deserialize<ProtoMessage>(memoryStream);
                        message = protoMessage;
                    }

                    //System.Console.WriteLine(
                    //    $"id:{new Guid(message.BroadcastId)} {JsonConvert.SerializeObject(message)}");
                    return message;
                });

            var headerCache = new ConcurrentDictionary<Guid, Header>();
            var payloadCache = new ConcurrentDictionary<Guid, List<ProtoMessage>>();

            var fileWriteRequestSubject = new Subject<FileWriteWrapper>();

            var headerObservable = protoMessageObservable
                .Where(message => !string.IsNullOrWhiteSpace(message.FileName) && message.PayloadCount.HasValue);

            var payloadObservable = protoMessageObservable
                .Where(message => message.PayloadIndex.HasValue && message.Payload != null);

            string GetFileName(ProtoMessage header)
            {
                return $"{System.DateTime.Now:HHmmssffff}.{header.FileName}";
            }

            Header ConvertToHeader(ProtoMessage header)
            {
                return new Header
                {
                    BroadcastId = header.GetBroadcastId(), FileName = GetFileName(header),
                    PayloadCount = header.PayloadCount.Value, PayloadMaxBytes = header.PayloadMaxSize.Value
                };
            }

            headerObservable
                .ObserveOn(observableScheduler)
                .Subscribe(header => headerCache.AddOrUpdate(header.GetBroadcastId(),
                    bcid =>
                    {
                        if (payloadCache.TryRemove(bcid, out var payloads))
                        {
                            var p = payloads.ToArray(); //todo:figure out why this bombed - List changed exception
                            payloads.Clear();
                            foreach (var payload in p)
                            {
                                fileWriteRequestSubject.OnNext(new FileWriteWrapper()
                                    {Header = ConvertToHeader(header), Payload = payload});
                            }
                        }

                        return ConvertToHeader(header);
                    },
                    (bcid, m) => m));


            payloadObservable
                .ObserveOn(observableScheduler)
                .Subscribe(payload =>
                {
                    if (headerCache.TryGetValue(payload.GetBroadcastId(), out var header))
                    {
                        fileWriteRequestSubject.OnNext(new FileWriteWrapper() {Header = header, Payload = payload});
                    }
                    else
                    {
                        payloadCache.AddOrUpdate(payload.GetBroadcastId(),
                            bcid =>
                            {
                                var list = new List<ProtoMessage>();
                                list.Add(payload);
                                return list;
                            },
                            (bcid, list) =>
                            {
                                list.Add(payload);
                                return list;
                            });
                    }
                });

            var writeCompleteObservable = fileWriteRequestSubject
                //.ObserveOn(observableScheduler)
                .Select(writeRequest =>
                {
                    return Observable.FromAsync(async () =>
                    {
                        var file = new FileInfo(writeRequest.Header.FileName);
                        using (var stream = file.OpenWrite())
                        {
                            var writeIndex = writeRequest.Payload.PayloadIndex.Value *
                                             writeRequest.Header.PayloadMaxBytes;
                            stream.Seek(writeIndex, SeekOrigin.Begin);
                            await stream.WriteAsync(writeRequest.Payload.Payload, token).ConfigureAwait(false);
                        }

                        return writeRequest.Header;
                    });
                }).Merge(1);

            var fileWriteObservable = writeCompleteObservable
                .ObserveOn(observableScheduler)
                .GroupBy(w => w.BroadcastId);

            const int fileCompleteTimeout = 10;
            var fileTimeout = TimeSpan.FromSeconds(fileCompleteTimeout);
            var fileStoppedObservable = CreateTimeoutObservable(fileTimeout, fileWriteObservable, observableScheduler);

            var fileStoppedSub = fileStoppedObservable.Subscribe();
            token.Register(() => fileStoppedSub.Dispose());

            UdpClient receiver = new UdpClient(port);
            //receiver.EnableBroadcast = true;
            IPEndPoint sender = new IPEndPoint(IPAddress.Any, 0); //new IPEndPoint(IPAddress.Parse("192.168.1.255"), 0);
            while (!token.IsCancellationRequested)
            {
                var received = await receiver.ReceiveAsync().ConfigureAwait(false);
                System.Console.WriteLine($"Buff bytes {received.Buffer.Length}");
                packetBufferObservable.OnNext(received.Buffer);
            }
        }

        private IObservable<string> CreateTimeoutObservable(TimeSpan completeTimeout,
            IObservable<IGroupedObservable<Guid, Header>> observable,
            IScheduler scheduler)
        {
            var fileStoppedObservable = observable
                .SelectMany(fileWriteObservable =>
                {
                    return Observable.FromAsync(async () =>
                    {
                        var first = await fileWriteObservable.FirstOrDefaultAsync();
                        if (first == null)
                        {
                            return Observable.Empty<Header>();
                        }

                        return fileWriteObservable.WhenStopped((Header) first, completeTimeout, scheduler);
                    });
                })
                .Merge()
                .Select(h => h.FileName);

            return fileStoppedObservable;
        }

        public event EventHandler<IBroadcastResult> BroadcastEnded;

        protected virtual void OnBroadcastEnded(IBroadcastResult e)
        {
            BroadcastEnded?.Invoke(this, e);
        }
    }
}<commit_msg>raise event on file end<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using ProtoBuf;
using Shoutr.Contracts;
using Shoutr.Reactive;
using Shoutr.Serialization;

namespace Shoutr
{
    public class Listener : IListener
    {
        public async Task Listen(int port = Defaults.Port,
            CancellationToken cancellationToken = default)
        {
            var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            var token = cancellationTokenSource.Token;

            var packetBufferObservable = new Subject<byte[]>();
            var observableScheduler =
                System.Reactive.Concurrency.Scheduler.Default; //new TaskPoolScheduler(new TaskFactory(token));
            var protoMessageObservable = packetBufferObservable
                .ObserveOn(observableScheduler)
                .Select(bytes =>
                {
                    ProtoMessage message;
                    using (var memoryStream = new MemoryStream(bytes))
                    {
                        var protoMessage = Serializer.Deserialize<ProtoMessage>(memoryStream);
                        message = protoMessage;
                    }

                    //System.Console.WriteLine(
                    //    $"id:{new Guid(message.BroadcastId)} {JsonConvert.SerializeObject(message)}");
                    return message;
                });

            var headerCache = new ConcurrentDictionary<Guid, Header>();
            var payloadCache = new ConcurrentDictionary<Guid, List<ProtoMessage>>();

            var fileWriteRequestSubject = new Subject<FileWriteWrapper>();

            var headerObservable = protoMessageObservable
                .Where(message => !string.IsNullOrWhiteSpace(message.FileName) && message.PayloadCount.HasValue);

            var payloadObservable = protoMessageObservable
                .Where(message => message.PayloadIndex.HasValue && message.Payload != null);

            string GetFileName(ProtoMessage header)
            {
                return $"{System.DateTime.Now:HHmmssffff}.{header.FileName}";
            }

            Header ConvertToHeader(ProtoMessage header)
            {
                return new Header
                {
                    BroadcastId = header.GetBroadcastId(), FileName = GetFileName(header),
                    PayloadCount = header.PayloadCount.Value, PayloadMaxBytes = header.PayloadMaxSize.Value
                };
            }

            headerObservable
                .ObserveOn(observableScheduler)
                .Subscribe(header => headerCache.AddOrUpdate(header.GetBroadcastId(),
                    bcid =>
                    {
                        if (payloadCache.TryRemove(bcid, out var payloads))
                        {
                            var p = payloads.ToArray(); //todo:figure out why this bombed - List changed exception
                            payloads.Clear();
                            foreach (var payload in p)
                            {
                                fileWriteRequestSubject.OnNext(new FileWriteWrapper()
                                    {Header = ConvertToHeader(header), Payload = payload});
                            }
                        }

                        return ConvertToHeader(header);
                    },
                    (bcid, m) => m));


            payloadObservable
                .ObserveOn(observableScheduler)
                .Subscribe(payload =>
                {
                    if (headerCache.TryGetValue(payload.GetBroadcastId(), out var header))
                    {
                        fileWriteRequestSubject.OnNext(new FileWriteWrapper() {Header = header, Payload = payload});
                    }
                    else
                    {
                        payloadCache.AddOrUpdate(payload.GetBroadcastId(),
                            bcid =>
                            {
                                var list = new List<ProtoMessage>();
                                list.Add(payload);
                                return list;
                            },
                            (bcid, list) =>
                            {
                                list.Add(payload);
                                return list;
                            });
                    }
                });

            var writeCompleteObservable = fileWriteRequestSubject
                //.ObserveOn(observableScheduler)
                .Select(writeRequest =>
                {
                    return Observable.FromAsync(async () =>
                    {
                        var file = new FileInfo(writeRequest.Header.FileName);
                        using (var stream = file.OpenWrite())
                        {
                            var writeIndex = writeRequest.Payload.PayloadIndex.Value *
                                             writeRequest.Header.PayloadMaxBytes;
                            stream.Seek(writeIndex, SeekOrigin.Begin);
                            await stream.WriteAsync(writeRequest.Payload.Payload, token).ConfigureAwait(false);
                        }

                        return writeRequest.Header;
                    });
                }).Merge(1);

            var fileWriteObservable = writeCompleteObservable
                .ObserveOn(observableScheduler)
                .GroupBy(w => w.BroadcastId);

            const int fileCompleteTimeout = 10;
            var fileTimeout = TimeSpan.FromSeconds(fileCompleteTimeout);
            var fileStoppedObservable = CreateTimeoutObservable(fileTimeout, fileWriteObservable, observableScheduler);

            var fileStoppedSub = fileStoppedObservable.Subscribe(header =>
            {
                OnBroadcastEnded(new BroadcastResult(header.BroadcastId, header.FileName));
            });
            token.Register(() => fileStoppedSub.Dispose());

            UdpClient receiver = new UdpClient(port);
            //receiver.EnableBroadcast = true;
            IPEndPoint sender = new IPEndPoint(IPAddress.Any, 0); //new IPEndPoint(IPAddress.Parse("192.168.1.255"), 0);
            while (!token.IsCancellationRequested)
            {
                var received = await receiver.ReceiveAsync().ConfigureAwait(false);
                Console.WriteLine($"Buff bytes {received.Buffer.Length}");
                packetBufferObservable.OnNext(received.Buffer);
            }
        }

        private IObservable<Header> CreateTimeoutObservable(TimeSpan completeTimeout,
            IObservable<IGroupedObservable<Guid, Header>> observable,
            IScheduler scheduler)
        {
            var fileStoppedObservable = observable
                .SelectMany(fileWriteObservable =>
                {
                    return Observable.FromAsync(async () =>
                    {
                        var first = await fileWriteObservable.FirstOrDefaultAsync();
                        if (first == null)
                        {
                            return Observable.Empty<Header>();
                        }

                        return fileWriteObservable.WhenStopped((Header) first, completeTimeout, scheduler);
                    });
                })
                .Merge();

            return fileStoppedObservable;
        }

        public event EventHandler<IBroadcastResult> BroadcastEnded;

        protected virtual void OnBroadcastEnded(IBroadcastResult e)
        {
            BroadcastEnded?.Invoke(this, e);
        }
    }
}
{'fraction_non_alphanumeric': 0.07850790197709387, 'fraction_numerical': 0.001919508605796916, 'mean_word_length': 1.3946683009039376, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3554208', 'n_tokens_mistral': 3567, 'n_tokens_neox': 3339, 'n_words': 637}