<commit_before>﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="RedisMessageRouter.cs" company="Kephas Software SRL">
//   Copyright (c) Kephas Software SRL. All rights reserved.
//   Licensed under the MIT license. See LICENSE file in the project root for full license information.
// </copyright>
// <summary>
//   Implements the redis message router class.
// </summary>
// --------------------------------------------------------------------------------------------------------------------

namespace Kephas.Messaging.Redis.Routing
{
    using System;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    using Kephas.Application;
    using Kephas.Configuration;
    using Kephas.ExceptionHandling;
    using Kephas.Interaction;
    using Kephas.Logging;
    using Kephas.Messaging;
    using Kephas.Messaging.Distributed;
    using Kephas.Messaging.Distributed.Routing;
    using Kephas.Redis;
    using Kephas.Redis.Configuration;
    using Kephas.Redis.Interaction;
    using Kephas.Serialization;
    using Kephas.Services;
    using Kephas.Threading.Tasks;
    using StackExchange.Redis;

    /// <summary>
    /// The Redis message router.
    /// </summary>
    [ProcessingPriority(Priority.Low)]
    [MessageRouter(ReceiverMatch = ChannelType + ":.*", IsFallback = true)]
    public class RedisAppMessageRouter : InProcessAppMessageRouter
    {
        private readonly IRedisClient redisClient;
        private readonly ISerializationService serializationService;
        private readonly IConfiguration<RedisClientSettings> redisConfiguration;
        private readonly IEventHub eventHub;
        private ISubscriber subscriber;
        private ChannelMessageQueue messageQueue;
        private ChannelMessageQueue appMessageQueue;
        private ChannelMessageQueue appInstanceMessageQueue;
        private string redisRootChannelName;
        private bool isRedisChannelInitialized;
        private IEventSubscription redisInitializedSubscription;

        /// <summary>
        /// Initializes a new instance of the <see cref="RedisAppMessageRouter"/> class.
        /// </summary>
        /// <param name="contextFactory">The context factory.</param>
        /// <param name="appRuntime">The application runtime.</param>
        /// <param name="messageProcessor">The message processor.</param>
        /// <param name="redisClient">The redis client.</param>
        /// <param name="serializationService">The serialization service.</param>
        /// <param name="redisConfiguration">The redis configuration.</param>
        /// <param name="eventHub">The event hub.</param>
        public RedisAppMessageRouter(
            IContextFactory contextFactory,
            IAppRuntime appRuntime,
            IMessageProcessor messageProcessor,
            IRedisClient redisClient,
            ISerializationService serializationService,
            IConfiguration<RedisClientSettings> redisConfiguration,
            IEventHub eventHub)
            : base(contextFactory, appRuntime, messageProcessor)
        {
            this.redisClient = redisClient;
            this.serializationService = serializationService;
            this.redisConfiguration = redisConfiguration;
            this.eventHub = eventHub;
        }

        /// <summary>
        /// Initializes the router.
        /// </summary>
        /// <param name="context">An optional context for initialization.</param>
        /// <param name="cancellationToken">The cancellation token.</param>
        /// <returns>
        /// An asynchronous result.
        /// </returns>
        protected override async Task InitializeCoreAsync(IContext context, CancellationToken cancellationToken = default)
        {
            await base.InitializeCoreAsync(context, cancellationToken).PreserveThreadContext();

            await this.InitializeRedisChannelAsync(context, cancellationToken).PreserveThreadContext();
        }

        /// <summary>Initializes the Redis channel asynchronously.</summary>
        /// <param name="context">An optional context for initialization.</param>
        /// <param name="cancellationToken">The cancellation token.</param>
        /// <returns>An awaitable task.</returns>
        protected virtual async Task InitializeRedisChannelAsync(IContext context, CancellationToken cancellationToken)
        {
            async Task InitializeAsync(RedisClientInitializedSignal initSignal, CancellationToken token)
            {
                this.redisInitializedSubscription?.Dispose();
                this.redisInitializedSubscription = null;

                if (initSignal.Severity.IsError())
                {
                    this.Logger.Info($"Redis client initialization failed, cancelling initialization of the Redis channel.");
                    return;
                }

                this.Logger.Info($"Redis initialized, starting initialization of the Redis channel...");

                var redisNS = this.redisConfiguration.Settings.Namespace;
                this.redisRootChannelName = string.IsNullOrEmpty(redisNS) ? ChannelType : $"{redisNS}:{ChannelType}";

                var connection = this.redisClient.GetConnection();

                this.subscriber = connection.GetSubscriber();
                this.messageQueue = await this.subscriber.SubscribeAsync(this.redisRootChannelName).PreserveThreadContext();
                this.messageQueue.OnMessage(this.ReceiveMessageAsync);

                this.appMessageQueue = await this.subscriber.SubscribeAsync($"{this.redisRootChannelName}:{this.AppRuntime.GetAppId()}").PreserveThreadContext();
                this.appMessageQueue.OnMessage(this.ReceiveMessageAsync);

                this.appInstanceMessageQueue = await this.subscriber.SubscribeAsync($"{this.redisRootChannelName}:{this.AppRuntime.GetAppInstanceId()}").PreserveThreadContext();
                this.appInstanceMessageQueue.OnMessage(this.ReceiveMessageAsync);

                this.Logger.Info($"Completed initialization of the Redis channel.");

                this.isRedisChannelInitialized = true;
            }

            if (!this.redisClient.IsInitialized)
            {
                this.Logger.Info($"Redis client not initialized, postponing initialization of the Redis channel.");

                this.redisInitializedSubscription = this.eventHub.Subscribe<RedisClientInitializedSignal>((s, ctx, ct) => InitializeAsync(s, ct));

                return;
            }

            await InitializeAsync(new RedisClientInitializedSignal(), cancellationToken).PreserveThreadContext();
        }

        /// <summary>
        /// Calculates the root channel name.
        /// </summary>
        /// <returns>
        /// The calculated root channel name.
        /// </returns>
        protected override string ComputeRootChannelName() => $"Redis-{base.ComputeRootChannelName()}";

        /// <summary><commit_msg>RedisAppMessageBroker: Change the name of the InProcess root channel name.<commit_after>﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="RedisMessageRouter.cs" company="Kephas Software SRL">
//   Copyright (c) Kephas Software SRL. All rights reserved.
//   Licensed under the MIT license. See LICENSE file in the project root for full license information.
// </copyright>
// <summary>
//   Implements the redis message router class.
// </summary>
// --------------------------------------------------------------------------------------------------------------------

namespace Kephas.Messaging.Redis.Routing
{
    using System;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    using Kephas.Application;
    using Kephas.Configuration;
    using Kephas.ExceptionHandling;
    using Kephas.Interaction;
    using Kephas.Logging;
    using Kephas.Messaging;
    using Kephas.Messaging.Distributed;
    using Kephas.Messaging.Distributed.Routing;
    using Kephas.Redis;
    using Kephas.Redis.Configuration;
    using Kephas.Redis.Interaction;
    using Kephas.Serialization;
    using Kephas.Services;
    using Kephas.Threading.Tasks;
    using StackExchange.Redis;

    /// <summary>
    /// The Redis message router.
    /// </summary>
    [ProcessingPriority(Priority.Low)]
    [MessageRouter(ReceiverMatch = ChannelType + ":.*", IsFallback = true)]
    public class RedisAppMessageRouter : InProcessAppMessageRouter
    {
        private readonly IRedisClient redisClient;
        private readonly ISerializationService serializationService;
        private readonly IConfiguration<RedisClientSettings> redisConfiguration;
        private readonly IEventHub eventHub;
        private ISubscriber subscriber;
        private ChannelMessageQueue messageQueue;
        private ChannelMessageQueue appMessageQueue;
        private ChannelMessageQueue appInstanceMessageQueue;
        private string redisRootChannelName;
        private bool isRedisChannelInitialized;
        private IEventSubscription redisInitializedSubscription;

        /// <summary>
        /// Initializes a new instance of the <see cref="RedisAppMessageRouter"/> class.
        /// </summary>
        /// <param name="contextFactory">The context factory.</param>
        /// <param name="appRuntime">The application runtime.</param>
        /// <param name="messageProcessor">The message processor.</param>
        /// <param name="redisClient">The redis client.</param>
        /// <param name="serializationService">The serialization service.</param>
        /// <param name="redisConfiguration">The redis configuration.</param>
        /// <param name="eventHub">The event hub.</param>
        public RedisAppMessageRouter(
            IContextFactory contextFactory,
            IAppRuntime appRuntime,
            IMessageProcessor messageProcessor,
            IRedisClient redisClient,
            ISerializationService serializationService,
            IConfiguration<RedisClientSettings> redisConfiguration,
            IEventHub eventHub)
            : base(contextFactory, appRuntime, messageProcessor)
        {
            this.redisClient = redisClient;
            this.serializationService = serializationService;
            this.redisConfiguration = redisConfiguration;
            this.eventHub = eventHub;
        }

        /// <summary>
        /// Initializes the router.
        /// </summary>
        /// <param name="context">An optional context for initialization.</param>
        /// <param name="cancellationToken">The cancellation token.</param>
        /// <returns>
        /// An asynchronous result.
        /// </returns>
        protected override async Task InitializeCoreAsync(IContext context, CancellationToken cancellationToken = default)
        {
            await base.InitializeCoreAsync(context, cancellationToken).PreserveThreadContext();

            await this.InitializeRedisChannelAsync(context, cancellationToken).PreserveThreadContext();
        }

        /// <summary>Initializes the Redis channel asynchronously.</summary>
        /// <param name="context">An optional context for initialization.</param>
        /// <param name="cancellationToken">The cancellation token.</param>
        /// <returns>An awaitable task.</returns>
        protected virtual async Task InitializeRedisChannelAsync(IContext context, CancellationToken cancellationToken)
        {
            async Task InitializeAsync(RedisClientInitializedSignal initSignal, CancellationToken token)
            {
                this.redisInitializedSubscription?.Dispose();
                this.redisInitializedSubscription = null;

                if (initSignal.Severity.IsError())
                {
                    this.Logger.Info($"Redis client initialization failed, cancelling initialization of the Redis channel.");
                    return;
                }

                this.Logger.Info($"Redis initialized, starting initialization of the Redis channel...");

                var redisNS = this.redisConfiguration.Settings.Namespace;
                this.redisRootChannelName = string.IsNullOrEmpty(redisNS) ? ChannelType : $"{redisNS}:{ChannelType}";

                var connection = this.redisClient.GetConnection();

                this.subscriber = connection.GetSubscriber();
                this.messageQueue = await this.subscriber.SubscribeAsync(this.redisRootChannelName).PreserveThreadContext();
                this.messageQueue.OnMessage(this.ReceiveMessageAsync);

                this.appMessageQueue = await this.subscriber.SubscribeAsync($"{this.redisRootChannelName}:{this.AppRuntime.GetAppId()}").PreserveThreadContext();
                this.appMessageQueue.OnMessage(this.ReceiveMessageAsync);

                this.appInstanceMessageQueue = await this.subscriber.SubscribeAsync($"{this.redisRootChannelName}:{this.AppRuntime.GetAppInstanceId()}").PreserveThreadContext();
                this.appInstanceMessageQueue.OnMessage(this.ReceiveMessageAsync);

                this.Logger.Info($"Completed initialization of the Redis channel.");

                this.isRedisChannelInitialized = true;
            }

            if (!this.redisClient.IsInitialized)
            {
                this.Logger.Info($"Redis client not initialized, postponing initialization of the Redis channel.");

                this.redisInitializedSubscription = this.eventHub.Subscribe<RedisClientInitializedSignal>((s, ctx, ct) => InitializeAsync(s, ct));

                return;
            }

            await InitializeAsync(new RedisClientInitializedSignal(), cancellationToken).PreserveThreadContext();
        }

        /// <summary>
        /// Calculates the root channel name.
        /// </summary>
        /// <returns>
        /// The calculated root channel name.
        /// </returns>
        protected override string ComputeRootChannelName() => $"InProcess-{base.ComputeRootChannelName()}";

        /// <summary>
{'fraction_non_alphanumeric': 0.12621290459664283, 'fraction_numerical': 0.0, 'mean_word_length': 2.3547160845806605, 'pattern_counts': {'":': 2, '<': 97, '<?xml version=': 0, '>': 101, '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': '12898', 'n_tokens_mistral': 3326, 'n_tokens_neox': 3285, 'n_words': 734}