TEAMModelOS/TEAMModelOS.SDK/DI/AzureServiceBus/AzureServiceBusExtensions.cs

using System;
using System.Collections;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Linq;
using Azure.Messaging.ServiceBus;
using TEAMModelOS.SDK;
using TEAMModelOS.SDK.Extension;
using System.Collections.Concurrent;

namespace TEAMModelOS.SDK.DI
{
    public static class AzureServiceBusExtensions
    {
        private static ConcurrentDictionary<string, ServiceBusSender> ServiceBusSenders { get; } = new ConcurrentDictionary<string, ServiceBusSender>();
        
        /// <summary>
        /// 發送信息至對列或主題
        /// </summary>       
        /// <param name="name">QueueName or TopicName</param>
        /// <param name="message">訊息</param>
        /// <returns></returns>
        public static async Task SendMessageAsync(this ServiceBusClient client, string name, ServiceBusMessage message)
        {
            try
            {
                ServiceBusSender sender = ServiceBusSenders.GetOrAdd(name, x => client.CreateSender(name));                
                await sender.SendMessageAsync(message);                
            }
            catch
            {
                throw;
            }
        }

        /// <summary>
        /// 批量發送訊息至對列或主題，如果批量失敗返回False
        /// </summary>       
        /// <param name="name">QueueName or TopicName</param>
        /// <param name="messages">批量訊息</param>
        /// <returns></returns>
        public static async Task<bool> SendBatchMessageAsync(this ServiceBusClient client, string name, IList<ServiceBusMessage> messages)
        {
            try
            {
                ServiceBusSender sender = ServiceBusSenders.GetOrAdd(name, x => client.CreateSender(name));
                ServiceBusMessageBatch messageBatch = await sender.CreateMessageBatchAsync();
                foreach (var msg in messages)
                {
                    if (!messageBatch.TryAddMessage(msg)) return false;
                }
                await sender.SendMessagesAsync(messageBatch);                
                return true;
            }
            catch
            {
                return false;
            }
        }
        
        /// <summary>
        /// 發送信息至對列或主題(指定時間排程)
        /// </summary>       
        /// <param name="name">QueueName or TopicName</param>
        /// <param name="message">訊息</param>
        /// <returns>排程訊息的序列號。</returns>
        public static async Task<long> SendScheduleMessageAsync(this ServiceBusClient client, string name, ServiceBusMessage message, DateTimeOffset scheduleTime)
        {
            try
            {
                ServiceBusSender sender = ServiceBusSenders.GetOrAdd(name, x => client.CreateSender(name));
                long num= await sender.ScheduleMessageAsync(message, scheduleTime);                
                return num;
            }
            catch
            {
                throw;
            }
        }

        /// <summary>
        /// 批量發送訊息至對列或主題(指定時間排程)，如果批量失敗返回False
        /// </summary>       
        /// <param name="name">QueueName or TopicName</param>
        /// <param name="messages">批量訊息</param>
        /// <returns>排程訊息的序列號</returns>
        public static async Task<IReadOnlyList<long>> SendScheduleMessagesAsync(this ServiceBusClient client, string name, IList<ServiceBusMessage> messages, DateTimeOffset scheduleTime)
        {
            try
            {
                ServiceBusSender sender = ServiceBusSenders.GetOrAdd(name, x => client.CreateSender(name));
                List<ServiceBusMessage> msgs = new() { };
                foreach (var msg in messages)
                {
                    msgs.Add(msg);
                }
                var nums= await sender.ScheduleMessagesAsync(msgs, scheduleTime);                
                return nums;
            }
            catch
            {
                return null;
            }
        }


        public static async Task<long> SendLeamMessage<T>(this ServiceBusClient client, string name, string id, string pk, long startTime)
        {
            //微調代碼
            var timer = DateTimeOffset.FromUnixTimeMilliseconds(startTime);
            /*if (type.Equals("start"))
            {
                if (timer.CompareTo(DateTimeOffset.UtcNow) < 0)
                {
                    progress = "going";
                }
            }
            else if(type.Equals("end"))
            {
                if (timer.CompareTo(DateTimeOffset.UtcNow) < 0)
                {
                    progress = "finish";
                }
            }*/
            
            
            //设定开始时间
            Dictionary<string, object> dict = new() {
                    { "name",typeof(T).Name},
                    { "id",id},
                    { "code",pk}
                };
            //var msgId = "1";
            string messageBody = $"Message {dict}";

            long SequenceNumber = await client.SendScheduleMessageAsync(name, new ServiceBusMessage(dict.ToJsonString()), timer);

            return SequenceNumber;
        }

        public static async Task CancelMessageAsync(this ServiceBusClient client, string name, long number)
        {
            try
            {
                ServiceBusSender sender = ServiceBusSenders.GetOrAdd(name, x => client.CreateSender(name));
                await sender.CancelScheduledMessageAsync(number);                
            }
            catch
            {
                throw;
            }
        }
    }
}