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

using Microsoft.Azure.Cosmos.Table;
using Microsoft.Azure.Cosmos.Table.Queryable;
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.Helper.Common.CollectionHelper;
using Grpc.Extension.Common;
using TEAMModelOS.SDK.Extension;

namespace TEAMModelOS.SDK.DI
{
    public static class AzureServiceBusExtensions
    {

        /// <summary>
        /// 發送信息至對列或主題
        /// </summary>       
        /// <param name="pathName">QueueName or TopicName</param>
        /// <param name="message">訊息</param>
        /// <returns></returns>
        public static async Task SendMessageAsync(this ServiceBusClient client, string pathName, ServiceBusMessage message)
        {
            try
            {
                ServiceBusSender sender = client.CreateSender(pathName);
                await sender.SendMessageAsync(message);
            }
            catch
            {
                throw;
            }
        }

        /// <summary>
        /// 批量發送訊息至對列或主題，如果批量失敗返回False
        /// </summary>       
        /// <param name="pathName">QueueName or TopicName</param>
        /// <param name="messages">批量訊息</param>
        /// <returns></returns>
        public static async Task<bool> SendBatchMessageAsync(this ServiceBusClient client, string pathName, IList<ServiceBusMessage> messages)
        {
            try
            {
                ServiceBusSender sender = client.CreateSender(pathName);
                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="pathName">QueueName or TopicName</param>
        /// <param name="message">訊息</param>
        /// <returns>排程訊息的序列號。</returns>
        public static async Task<long> SendScheduleMessageAsync(this ServiceBusClient client, string pathName, ServiceBusMessage message, DateTimeOffset scheduleTime)
        {
            try
            {
                ServiceBusSender sender = client.CreateSender(pathName);
                return await sender.ScheduleMessageAsync(message, scheduleTime);
            }
            catch
            {
                throw;
            }
        }

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


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

            Console.WriteLine($"Sending message: {messageBody}");
            long SequenceNumber = await client.SendScheduleMessageAsync(TopicName, new ServiceBusMessage(dict.ToJsonString()), timer);
            return SequenceNumber;
        }
    }
}