<commit_before>﻿using Lime.Messaging.Resources;
using Lime.Protocol;
using Lime.Protocol.Client;
using Lime.Protocol.Network;
using Lime.Protocol.Security;
using Lime.Transport.Tcp;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Takenet.MessagingHub.Client
{
    public class MessagingHubClient : IMessagingHubClient
    {
        static readonly MediaType defaultReceiverMediaType = new MediaType("null", "null");
        static readonly string defaultDomainName = "msging.net";

        public IMessageSender MessageSender { get; private set; }

        readonly Uri endpoint;
        readonly IDictionary<MediaType, IList<IMessageReceiver>> receivers;

        string domainName;
        string login;
        string password;
        string accessKey;
        IClientChannel clientChannel;

        CancellationTokenSource cancellationTokenSource;
        Task backgroundExecution;
        Task messageReceiver;

        MessagingHubClient()
        {
            receivers = new Dictionary<MediaType, IList<IMessageReceiver>>();
        }

        public MessagingHubClient(string hostname = null, string domainName = null) : this()
        {
            this.domainName = domainName ?? defaultDomainName;
            hostname = hostname ?? defaultDomainName;
            this.endpoint = new Uri($"net.tcp://{hostname}:55321");
        }

        public MessagingHubClient UsingAccount(string login, string password)
        {
            this.login = login;
            this.password = password;
            return this;
        }

        public MessagingHubClient UsingAccessKey(string login, string key)
        {
            this.login = login;
            this.accessKey = key;
            return this;
        }

        #region PublicMethods

        /// <summary>
        /// Add a message receiver listener to handle received messages
        /// </summary>
        /// <param name="receiver">Listener</param>
        /// <param name="forMimeType">MediaType used as a filter of messages received by listener. When not informed, only receives messages which no 'typed' receiver is registered</param>
        /// <returns></returns>
        public MessagingHubClient AddMessageReceiver(IMessageReceiver receiver, MediaType forMimeType = null)
        {
            AddReceiver(forMimeType, receiver);
            return this;
        }

        /// <summary>
        /// Connect and receives messages from Lime server
        /// </summary>
        /// <returns>Task representing the running state of the client (when this tasks finishes, the connection has been terminated)</returns>
        public async Task<Task> StartAsync()
        {
            var authentication = GetAuthenticationScheme();

            clientChannel = await CreateAndOpenAsync().ConfigureAwait(false);

            var session = await EstablishSession(authentication).ConfigureAwait(false);

            if (session.State != SessionState.Established)
            {
                throw new Exception($"Could not connect: {session.Reason.Description} (code: {session.Reason.Code})");
            }

            MessageSender = new MessageSenderWrapper(clientChannel);
            await clientChannel.SetResourceAsync(
                LimeUri.Parse(UriTemplates.PRESENCE),
                new Presence { RoutingRule = RoutingRule.Identity },
                CancellationToken.None)
                .ConfigureAwait(false);

            cancellationTokenSource = new CancellationTokenSource();
            messageReceiver = Task.Run(() => ProcessIncomingMessages(cancellationTokenSource.Token));
            //Add all background tasks here. Ex: notification receiver
            backgroundExecution = Task.WhenAll(messageReceiver);

            return backgroundExecution;
        }

        /// <summary>
        /// Close connecetion and stop to receive messages from Lime server 
        /// </summary>
        /// <returns>
        public async Task StopAsync()
        {

            if (clientChannel?.State == SessionState.Established)
            {
                await clientChannel.SendFinishingSessionAsync().ConfigureAwait(false);
            }
            else
            {
                await clientChannel.Transport.CloseAsync(CancellationToken.None).ConfigureAwait(false);
            }

            if (cancellationTokenSource != null && !cancellationTokenSource.IsCancellationRequested)
            {
                cancellationTokenSource.Cancel();
                await backgroundExecution;
                cancellationTokenSource.Dispose();
            }
        }

        #endregion PublicMethods

        #region InternalMethods

        void AddReceiver(MediaType mediaType, IMessageReceiver receiver)
        {
            var mediaTypeToSave = mediaType ?? MediaTypes.Any;

            IList<IMessageReceiver> mediaTypeReceivers;
            if (!receivers.TryGetValue(mediaTypeToSave, out mediaTypeReceivers))
            {
                mediaTypeReceivers = new List<IMessageReceiver>();
                receivers.Add(mediaTypeToSave, mediaTypeReceivers);
            }

            mediaTypeReceivers.Add(receiver);
        }

        async Task ProcessIncomingMessages(CancellationToken cancellationToken)
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var message = await clientChannel.ReceiveMessageAsync(cancellationToken).ConfigureAwait(false);

                //When cancelation 
                if(message == null)
                {
                    continue;
                }

                IList<IMessageReceiver> mimeTypeReceivers = null;
                if (receivers.TryGetValue(message.Type, out mimeTypeReceivers) ||
                    receivers.TryGetValue(MediaTypes.Any, out mimeTypeReceivers) ||
                    receivers.TryGetValue(defaultReceiverMediaType, out mimeTypeReceivers))
                {
                    try
                    {
                        await Task.WhenAll(
                                mimeTypeReceivers.Select(r => CallMessageReceiver(r, message))).ConfigureAwait(false);
                    }
                    catch { }
                }
            }
        }

        Task CallMessageReceiver(IMessageReceiver receiver, Message message)
        {
            if (receiver is MessageReceiverBase)
            {
                ((MessageReceiverBase)receiver).Sender = MessageSender;<commit_msg>Removed default media type from client<commit_after>﻿using Lime.Messaging.Resources;
using Lime.Protocol;
using Lime.Protocol.Client;
using Lime.Protocol.Network;
using Lime.Protocol.Security;
using Lime.Transport.Tcp;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Takenet.MessagingHub.Client
{
    public class MessagingHubClient : IMessagingHubClient
    {
        static readonly string defaultDomainName = "msging.net";

        public IMessageSender MessageSender { get; private set; }

        readonly Uri endpoint;
        readonly IDictionary<MediaType, IList<IMessageReceiver>> receivers;

        string domainName;
        string login;
        string password;
        string accessKey;
        IClientChannel clientChannel;

        CancellationTokenSource cancellationTokenSource;
        Task backgroundExecution;
        Task messageReceiver;

        MessagingHubClient()
        {
            receivers = new Dictionary<MediaType, IList<IMessageReceiver>>();
        }

        public MessagingHubClient(string hostname = null, string domainName = null) : this()
        {
            this.domainName = domainName ?? defaultDomainName;
            hostname = hostname ?? defaultDomainName;
            this.endpoint = new Uri($"net.tcp://{hostname}:55321");
        }

        public MessagingHubClient UsingAccount(string login, string password)
        {
            this.login = login;
            this.password = password;
            return this;
        }

        public MessagingHubClient UsingAccessKey(string login, string key)
        {
            this.login = login;
            this.accessKey = key;
            return this;
        }

        #region PublicMethods

        /// <summary>
        /// Add a message receiver listener to handle received messages
        /// </summary>
        /// <param name="receiver">Listener</param>
        /// <param name="forMimeType">MediaType used as a filter of messages received by listener. When not informed, only receives messages which no 'typed' receiver is registered</param>
        /// <returns></returns>
        public MessagingHubClient AddMessageReceiver(IMessageReceiver receiver, MediaType forMimeType = null)
        {
            AddReceiver(forMimeType, receiver);
            return this;
        }

        /// <summary>
        /// Connect and receives messages from Lime server
        /// </summary>
        /// <returns>Task representing the running state of the client (when this tasks finishes, the connection has been terminated)</returns>
        public async Task<Task> StartAsync()
        {
            var authentication = GetAuthenticationScheme();

            clientChannel = await CreateAndOpenAsync().ConfigureAwait(false);

            var session = await EstablishSession(authentication).ConfigureAwait(false);

            if (session.State != SessionState.Established)
            {
                throw new Exception($"Could not connect: {session.Reason.Description} (code: {session.Reason.Code})");
            }

            MessageSender = new MessageSenderWrapper(clientChannel);
            await clientChannel.SetResourceAsync(
                LimeUri.Parse(UriTemplates.PRESENCE),
                new Presence { RoutingRule = RoutingRule.Identity },
                CancellationToken.None)
                .ConfigureAwait(false);

            cancellationTokenSource = new CancellationTokenSource();
            messageReceiver = Task.Run(() => ProcessIncomingMessages(cancellationTokenSource.Token));
            //Add all background tasks here. Ex: notification receiver
            backgroundExecution = Task.WhenAll(messageReceiver);

            return backgroundExecution;
        }

        /// <summary>
        /// Close connecetion and stop to receive messages from Lime server 
        /// </summary>
        /// <returns>
        public async Task StopAsync()
        {

            if (clientChannel?.State == SessionState.Established)
            {
                await clientChannel.SendFinishingSessionAsync().ConfigureAwait(false);
            }
            else
            {
                await clientChannel.Transport.CloseAsync(CancellationToken.None).ConfigureAwait(false);
            }

            if (cancellationTokenSource != null && !cancellationTokenSource.IsCancellationRequested)
            {
                cancellationTokenSource.Cancel();
                await backgroundExecution;
                cancellationTokenSource.Dispose();
            }
        }

        #endregion PublicMethods

        #region InternalMethods

        void AddReceiver(MediaType mediaType, IMessageReceiver receiver)
        {
            var mediaTypeToSave = mediaType ?? MediaTypes.Any;

            IList<IMessageReceiver> mediaTypeReceivers;
            if (!receivers.TryGetValue(mediaTypeToSave, out mediaTypeReceivers))
            {
                mediaTypeReceivers = new List<IMessageReceiver>();
                receivers.Add(mediaTypeToSave, mediaTypeReceivers);
            }

            mediaTypeReceivers.Add(receiver);
        }

        async Task ProcessIncomingMessages(CancellationToken cancellationToken)
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var message = await clientChannel.ReceiveMessageAsync(cancellationToken).ConfigureAwait(false);

                //When cancelation 
                if(message == null)
                {
                    continue;
                }

                IList<IMessageReceiver> mimeTypeReceivers = null;
                if (receivers.TryGetValue(message.Type, out mimeTypeReceivers) ||
                    receivers.TryGetValue(MediaTypes.Any, out mimeTypeReceivers))
                {
                    try
                    {
                        await Task.WhenAll(
                                mimeTypeReceivers.Select(r => CallMessageReceiver(r, message))).ConfigureAwait(false);
                    }
                    catch { }
                }
            }
        }

        Task CallMessageReceiver(IMessageReceiver receiver, Message message)
        {
            if (receiver is MessageReceiverBase)
            {
                ((MessageReceiverBase)receiver).Sender = MessageSender;
{'fraction_non_alphanumeric': 0.07910436197412257, 'fraction_numerical': 0.0007747733787867049, 'mean_word_length': 1.9886547811993518, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 53, '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': '2976403', 'n_tokens_mistral': 3075, 'n_tokens_neox': 3001, 'n_words': 694}