<commit_before>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Lime.Protocol;
using Lime.Protocol.Listeners;
using Lime.Protocol.Util;
using Takenet.MessagingHub.Client.Connection;
using Takenet.MessagingHub.Client.Sender;

namespace Takenet.MessagingHub.Client.Listener
{
    internal sealed class MessagingHubListener : IMessagingHubListener
    {
        private readonly IMessagingHubConnection _connection;
        private readonly IMessagingHubSender _sender;
        private ChannelListener _channelListener;
        private CancellationTokenSource _cts;
        private MessageReceivedHandler _messageHandler;
        private NotificationReceivedHandler _notificationHandler;

        public MessagingHubListener(IMessagingHubConnection connection, IMessagingHubSender sender = null)
        {
            _connection = connection;
            _sender = sender ?? new MessagingHubSender(connection);
            EnvelopeRegistrar = new EnvelopeListenerRegistrar(this);
        }
        
        public bool Listening { get; private set; }

        internal EnvelopeListenerRegistrar EnvelopeRegistrar { get; }

        public void AddMessageReceiver(IMessageReceiver messageReceiver, Predicate<Message> messageFilter, int priority = 0)
        {
            EnvelopeRegistrar.AddMessageReceiver(() => messageReceiver, messageFilter, priority);
        }

        public void AddNotificationReceiver(INotificationReceiver notificationReceiver, Predicate<Notification> notificationFilter, int priority = 0)
        {
            EnvelopeRegistrar.AddNotificationReceiver(() => notificationReceiver, notificationFilter, priority);
        }

        public Task StartAsync(CancellationToken cancellationToken = default(CancellationToken))
        {
            StartEnvelopeListeners();
            Listening = true;
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken = default(CancellationToken))
        {
            StopEnvelopeListeners();
            Listening = false;
            return Task.CompletedTask;
        }

        private void StartEnvelopeListeners()
        {
            _cts = new CancellationTokenSource();
            _messageHandler = new MessageReceivedHandler(_sender, EnvelopeRegistrar, _cts);
            _notificationHandler = new NotificationReceivedHandler(_sender, EnvelopeRegistrar, _cts);
            
            _channelListener = new ChannelListener(
                m => _messageHandler.HandleAsync(m, _cts.Token),
                n => _notificationHandler.HandleAsync(n, _cts.Token),
                c => TaskUtil.TrueCompletedTask);
            _channelListener.Start(_connection.OnDemandClientChannel);<commit_msg>Removed unecessary fields<commit_after>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Lime.Protocol;
using Lime.Protocol.Listeners;
using Lime.Protocol.Util;
using Takenet.MessagingHub.Client.Connection;
using Takenet.MessagingHub.Client.Sender;

namespace Takenet.MessagingHub.Client.Listener
{
    internal sealed class MessagingHubListener : IMessagingHubListener
    {
        private readonly IMessagingHubConnection _connection;
        private readonly IMessagingHubSender _sender;
        private ChannelListener _channelListener;
        private CancellationTokenSource _cts;

        public MessagingHubListener(IMessagingHubConnection connection, IMessagingHubSender sender = null)
        {
            _connection = connection;
            _sender = sender ?? new MessagingHubSender(connection);
            EnvelopeRegistrar = new EnvelopeListenerRegistrar(this);
        }
        
        public bool Listening { get; private set; }

        internal EnvelopeListenerRegistrar EnvelopeRegistrar { get; }

        public void AddMessageReceiver(IMessageReceiver messageReceiver, Predicate<Message> messageFilter, int priority = 0)
        {
            EnvelopeRegistrar.AddMessageReceiver(() => messageReceiver, messageFilter, priority);
        }

        public void AddNotificationReceiver(INotificationReceiver notificationReceiver, Predicate<Notification> notificationFilter, int priority = 0)
        {
            EnvelopeRegistrar.AddNotificationReceiver(() => notificationReceiver, notificationFilter, priority);
        }

        public Task StartAsync(CancellationToken cancellationToken = default(CancellationToken))
        {
            StartEnvelopeListeners();
            Listening = true;
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken = default(CancellationToken))
        {
            StopEnvelopeListeners();
            Listening = false;
            return Task.CompletedTask;
        }

        private void StartEnvelopeListeners()
        {
            _cts = new CancellationTokenSource();
            var messageHandler = new MessageReceivedHandler(_sender, EnvelopeRegistrar, _cts);
            var notificationHandler = new NotificationReceivedHandler(_sender, EnvelopeRegistrar, _cts);
            
            _channelListener = new ChannelListener(
                m => messageHandler.HandleAsync(m, _cts.Token),
                n => notificationHandler.HandleAsync(n, _cts.Token),
                c => TaskUtil.TrueCompletedTask);
            _channelListener.Start(_connection.OnDemandClientChannel);
{'fraction_non_alphanumeric': 0.06474287828338883, 'fraction_numerical': 0.0007399186089530152, 'mean_word_length': 2.9963045084996303, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, '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': '4043909', 'n_tokens_mistral': 1265, 'n_tokens_neox': 1261, 'n_words': 281}