using OfficeOpenXml.Packaging.Ionic.Zlib;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Compression;
using System.Linq;
using System.Text;
using System.Threading;

namespace AutoTurnOver.Utility
{
    public class RabbitMQHelper
    {
        private static ConnectionFactory connectionFactory = new ConnectionFactory
        {
            HostName = ConfigHelper.GetValue("RabbitMQ:Host"),
            UserName = ConfigHelper.GetValue("RabbitMQ:UserName"),
            Password = ConfigHelper.GetValue("RabbitMQ:Password"),
            Port = int.Parse(ConfigHelper.GetValue("RabbitMQ:Port")),
        };
        private static IConnection connection;
        private RabbitMQHelper() { }

        private static void CreateConn()
        {
            Console.WriteLine("HostName:" + connectionFactory.HostName);
            Console.WriteLine("UserName:" + connectionFactory.UserName);
            Console.WriteLine("Password:" + connectionFactory.Password);
            connection = connectionFactory.CreateConnection();
        }
        /// <summary>
        /// 发送多条消息-指定Mq
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="queName"></param>
        /// <param name="msg"></param>
        public static bool SendMessagesToTargetMq<T>(string queName, ConnectionFactory connectionFactory, List<T> msgs) where T : class
        {
            if (msgs == null && !msgs.Any())
            {
                return false;
            }
            try
            {
                var newConnection = connectionFactory.CreateConnection();
                using (var channel = newConnection.CreateModel())
                {
                    //声明一个队列
                    channel.QueueDeclare(queName, true, false, false, null);

                    var basicProperties = channel.CreateBasicProperties();
                    //1：非持久化 2：可持久化
                    basicProperties.DeliveryMode = 2;
                    var address = new PublicationAddress(ExchangeType.Direct, "", queName);
                    msgs.ForEach((msg) =>
                    {
                        var payload = Encoding.UTF8.GetBytes(msg.ToJson());
                        channel.BasicPublish(address, basicProperties, payload);
                    });
                }
                newConnection.Close();
                newConnection.Dispose();
                return true;
            }
            catch (Exception ex)
            {
                var gewa = ex;
                return false;
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="queName"></param>
        /// <param name="msg"></param>
        public static bool SendMsg<T>(string exchangeName, string queName, T msg, bool doCompress = false) where T : class
        {
            if (msg == null)
            {
                return false;
            }
            try
            {
                if (connection == null || !connection.IsOpen)
                {
                    CreateConn();
                }
                using (var channel = connection.CreateModel())
                {
                    //声明一个队列
                    channel.QueueDeclare(queName, true, false, false, null);
                    if (!string.IsNullOrEmpty(exchangeName))
                    {
                        //声明交换机
                        channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, true);
                        //绑定队列，交换机，路由键
                        channel.QueueBind(queName, exchangeName, queName);
                    }

                    var basicProperties = channel.CreateBasicProperties();
                    //1：非持久化 2：可持久化
                    basicProperties.DeliveryMode = 2;
                    var inputBytes = Encoding.UTF8.GetBytes(msg.ToJson());
                    var payload = doCompress ? CompressZlipBytes(inputBytes) : inputBytes;

                    var address = new PublicationAddress(ExchangeType.Direct, exchangeName, queName);
                    channel.BasicPublish(address, basicProperties, payload);
                }
                return true;
            }
            catch (Exception)
            {

                return false;
            }

        }

        /// <summary>
        /// 发送多条消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="queName"></param>
        /// <param name="msg"></param>
        public static bool SendMessages<T>(string exchangeName, string queName, List<T> msgs, bool doCompress = false) where T : class
        {
            if (msgs == null && !msgs.Any())
            {
                return false;
            }
            try
            {
                if (connection == null || !connection.IsOpen)
                {
                    CreateConn();
                }
                using (var channel = connection.CreateModel())
                {
                    //声明一个队列
                    channel.QueueDeclare(queName, true, false, false, null);
                    if (!string.IsNullOrEmpty(exchangeName))
                    {
                        //声明交换机
                        channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, true);
                        //绑定队列，交换机，路由键
                        channel.QueueBind(queName, exchangeName, queName);
                    }

                    var basicProperties = channel.CreateBasicProperties();
                    //1：非持久化 2：可持久化
                    basicProperties.DeliveryMode = 2;
                    var address = new PublicationAddress(ExchangeType.Direct, exchangeName, queName);
                    msgs.ForEach((msg) =>
                    {
                        var inputBytes = Encoding.UTF8.GetBytes(msg.ToJson());
                        var payload = doCompress ? CompressZlipBytes(inputBytes) : inputBytes;
                        channel.BasicPublish(address, basicProperties, payload);
                    });
                }
                return true;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        /// <summary>
        /// 消息加入队列中
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="queneName"></param>
        /// <param name="msg"></param>
        /// <returns></returns>
        public static bool EnqueneMsg<T>(string queneName, T msg, bool doCompress = false) where T : class
        {
            if (msg == null)
            {
                return false;
            }
            for (int i = 0; i < 3; i++)
            {
                var pushMsgResult = SendMsg("", queneName, msg, doCompress);
                if (pushMsgResult) return true;
                Thread.Sleep(100);
            }
            return false;
        }
        public static bool EnqueneMessages<T>(string queneName, List<T> msgs, bool doCompress = false) where T : class
        {
            if (msgs == null && !msgs.Any())
            {
                return false;
            }
            for (int i = 0; i < 3; i++)
            {
                var pushMsgResult = SendMessages("", queneName, msgs, doCompress);
                if (pushMsgResult) return true;
                Thread.Sleep(100);
            }
            return false;
        }
        public static bool GetMessageCount(string queName, out uint count)
        {
            count = 0;
            try
            {

                if (connection == null || !connection.IsOpen)
                {
                    CreateConn();
                }
                using (var channel = connection.CreateModel())
                {
                    //声明一个队列
                    channel.QueueDeclare(queName, true, false, false, null);
                    count = channel.MessageCount(queName);
                }
                return true;
            }
            catch (Exception ex)
            {
                var gewa = ex;
                return false;
            }

        }
        public static IModel GetChannel()
        {
            if (connection == null || !connection.IsOpen)
            {
                CreateConn();
            }
            var channel = connection.CreateModel();
            return channel;
        }
        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="queName"></param>
        /// <param name="received"></param>
        public static void Receive<T>(string exchangeName, string queName, IModel channel, Action<T> received, bool isZip) where T : class
        {
            try
            {

                {
                    //声明一个队列
                    channel.QueueDeclare(queName, true, false, false, null);
                    if (!string.IsNullOrEmpty(exchangeName))
                    {
                        //声明交换机
                        channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, true);
                        //绑定队列，交换机，路由键
                        channel.QueueBind(queName, exchangeName, queName);
                    }

                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                    //事件基本消费者
                    var consumer = new EventingBasicConsumer(channel);
                    //接收到消息事件
                    consumer.Received += (ch, ea) =>
                    {
                        var temp = isZip ? DeconpressZlip(ea.Body.ToArray()) : ea.Body.ToArray();
                        string message = Encoding.UTF8.GetString(temp);
                        var msg = message.ToObj<T>();
                        DateTime time = DateTime.Now;
                        received(msg);
                        var timeEnd = DateTime.Now - time;
                        //channel.DefaultConsumer.HandleBasicCancelOk(consumer.ConsumerTag);
                        if (channel.IsClosed)
                        {
                            return;
                        }
                        Console.WriteLine($"{DateTime.Now} 任务执行完成，用时 {timeEnd.TotalSeconds:0.00}s {queName} 队列剩余任务数量： {channel.MessageCount(queName)}");
                        //确认该消息已被消费
                        channel.BasicAck(ea.DeliveryTag, false);
                    };
                    //启动消费者 设置为手动应答消息
                    channel.BasicConsume(queName, false, consumer);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            Thread.Sleep(60);
        }



        /// <summary>
        /// 释放链接
        /// </summary>
        public void Dispose()
        {
            //channel.Dispose();
            connection.Close();
        }

        /// <summary>
        /// 关闭连接
        /// </summary>
        public void ReceiveClose()
        {
            //channel.BasicCancel(consumer.ConsumerTag);
            // channel.Close();
            //Dispose();
        }

        /// <summary>
        /// 解压zlip压缩二进制
        /// </summary>
        /// <param name="bytes"></param>
        /// <returns></returns>
        public static byte[] DeconpressZlip(byte[] bytes)
        {
            using (var compressStream = new MemoryStream(bytes))
            {
                using (var zipStream = new ZlibStream(compressStream, OfficeOpenXml.Packaging.Ionic.Zlib.CompressionMode.Decompress))
                {
                    using (var resultStream = new MemoryStream())
                    {
                        zipStream.CopyTo(resultStream);
                        return resultStream.ToArray();
                    }
                }
            }



        }

        //zlip压缩字节
        //1.创建压缩的数据流 
        //2.设定compressStream为存放被压缩的文件流,并设定为压缩模式
        //3.将需要压缩的字节写到被压缩的文件流
        public static byte[] CompressZlipBytes(byte[] bytes)
        {
            using (MemoryStream compressStream = new MemoryStream())
            {
                using (var zipStream = new ZlibStream(compressStream, OfficeOpenXml.Packaging.Ionic.Zlib.CompressionMode.Compress))
                {
                    zipStream.Write(bytes, 0, bytes.Length);
                    return compressStream.ToArray();
                }
            }

        }
    }
}
