/********************************
 *
 * 作者:吴隆健
 * 创建时间: 2019/6/24 14:22:26
 * 
 ********************************/
#if RABBITMQ
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace MrWu.RabbitMQ {
    /// <summary>
    /// rabbit管理
    /// </summary>
    public class RabbitManager {

        /// <summary>
        /// 配置
        /// </summary>
        public RabbitConfig config {
            get;
            private set;
        }

        /// <summary>
        /// 发送失败
        /// </summary>
        public event MessageHandle SendResult;

        /// <summary>
        /// 未路由成功事件
        /// </summary>
        public event MessageHandle notRoutingOk;

        /// <summary>
        /// 构造
        /// </summary>
        public RabbitManager() {
        }

        /// <summary>
        /// 未确认的数量
        /// </summary>
        public int NoAckCount {
            get {
                return waitAck.Count;
            }
        }

        /// <summary>
        /// 启动rabbit管理
        /// </summary>
        /// <param name="config"></param>
        public void Start(RabbitConfig config = null) {
            if (connection != null) {
                Debug.Debug.Error("已启动过!");
                return;
            }
            if (config != null)
                this.config = config;
            if (this.config == null)
                throw new ArgumentNullException("config is null");

            Connection();
        }

        /// <summary>
        /// 链接
        /// </summary>
        public IConnection connection {
            get;
            private set;
        }

        /// <summary>
        /// 生产者信道
        /// </summary>
        private IModel producer {
            get;
            set;
        }

        /// <summary>
        /// 获取信息的通道
        /// </summary>
        private IModel infoChannel {
            get;
            set;
        }

        /// <summary>
        /// 等待发送确认的消息
        /// </summary>
        private ConcurrentDictionary<ulong, Message> waitAck { get; } = new ConcurrentDictionary<ulong, Message>();

        /// <summary>
        /// 生产者锁
        /// </summary>
        private readonly object lockobj = new object();

        /// <summary>
        /// 链接rabbit
        /// </summary>
        private void Connection() {
            ConnectionFactory factory = new ConnectionFactory();
            factory.UserName = config.username;
            factory.Password = config.password;
            factory.VirtualHost = config.vhost;
            factory.HostName = config.host;
            factory.Port = config.port;
            factory.AutomaticRecoveryEnabled = true;

            if (connection != null) {
                connection.ConnectionBlocked -= ConnectionBlockedEventHandler;
                connection.ConnectionUnblocked -= ConnectionUnblockedEventHandler;
            }


            connection = factory.CreateConnection();
            connection.ConnectionBlocked += ConnectionBlockedEventHandler;
            connection.ConnectionUnblocked += ConnectionUnblockedEventHandler;

            Debug.Debug.Log("链接创建生产者!");
            CreateProducer();
            DeclareExchange(config.exchanges);
            DeclareQueue(config.queues);
            Bind(config.binds);
        }

        /// <summary>
        /// 创建生产者
        /// </summary>
        private void CreateProducer() {
            try {

                Debug.Debug.Log("创建生产者:" + Thread.CurrentThread.ManagedThreadId);

                producer = connection.CreateModel();
                producer.ConfirmSelect();
                producer.ModelShutdown += ModelShutdown;
                producer.BasicReturn += MessageReturn;
                producer.BasicAcks += MessageAck;
                producer.BasicNacks += MessageNAck;
            } catch (Exception e) {
                Debug.Debug.Error("创建生产者出错!" + e);
            }
        }

        /// <summary>
        /// 消息阻塞!!!
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="args"></param>
        private void ConnectionBlockedEventHandler(IConnection sender, ConnectionBlockedEventArgs args) {
            Debug.Debug.Fatal("RabbitMQ 消息阻塞,链接被禁用!" + args.Reason);
        }

        /// <summary>
        /// 消息阻塞被解除!
        /// </summary>
        /// <param name="sender"></param>
        private void ConnectionUnblockedEventHandler(IConnection sender) {
            Debug.Debug.Fatal("RabbitMQ 消息解除阻塞!");
        }

        /// <summary>
        /// 链接禁用!
        /// </summary>
        public void ConnectionBlocked() {
            if (connection == null)
                return;

            connection.HandleConnectionBlocked("auto blocked");
        }

        /// <summary>
        /// 解除链接禁用
        /// </summary>
        public void ConnectionUnBlocked() {
            if (connection == null)
                return;
            connection.HandleConnectionUnblocked();
        }

        /// <summary>
        /// 创建信息通道
        /// </summary>
        private void CreateInfoChannel() {
            try {
                infoChannel = connection.CreateModel();
                infoChannel.ModelShutdown += ModelShutdown;
            } catch (Exception e) {
                Debug.Debug.Error("创建信息获取通道出错!" + e.ToString());
            }
        }

        /// <summary>
        /// 消息发送成功
        /// </summary>
        /// <param name="model"></param>
        /// <param name="args"></param>
        private void MessageAck(IModel model, BasicAckEventArgs args) {
            DoAck(model, args.DeliveryTag, args.Multiple, true);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="model"></param>
        /// <param name="deliverytag"></param>
        /// <param name="multiple"></param>
        /// <param name="success"></param>
        private void DoAck(IModel model, ulong deliverytag, bool multiple, bool success) {

            void DoAck(Message _msg, bool _success) {
                _msg.success = success;
                if (_msg.SendResult != null)
                    _msg.SendResult(_msg);
                else
                    SendResult?.Invoke(_msg);

                if (_msg.isAutoFree)
                    MessagePool.PutMessage(ref _msg);
            }

            Message msg;

            //同步确认的消息也会到这里来!!!
            if (!multiple) {
                if (waitAck.TryRemove(deliverytag, out msg)) {//拿到
                    DoAck(msg, success);
                }
            } else {
                List<ulong> keys = new List<ulong>();
                foreach (var item in waitAck) {
                    if (item.Value.deliveryTag <= deliverytag)
                        keys.Add(item.Value.deliveryTag);
                }
                foreach (var item in keys) {
                    if (waitAck.TryRemove(item, out msg)) {
                        DoAck(msg, success);
                    }
                }
            }
        }



        /// <summary>
        /// 消息未被发送
        /// </summary>
        /// <param name="model"></param>
        /// <param name="args"></param>
        private void MessageNAck(IModel model, BasicNackEventArgs args) {
            DoAck(model, args.DeliveryTag, args.Multiple, false);
        }

        /// <summary>
        /// 信道断开
        /// </summary>
        /// <param name="model"></param>
        /// <param name="reason"></param>
        private void ModelShutdown(IModel model, ShutdownEventArgs reason) {

            Debug.Debug.Warning("信道断开事件:" + Thread.CurrentThread.ManagedThreadId);

            if (producer != null && producer.IsClosed) {
                Debug.Debug.Info("剩余确认数量:" + waitAck.Count);
                // producer.BasicNack 
                producer.ModelShutdown -= ModelShutdown;
                producer.BasicReturn -= MessageReturn;
                producer.BasicAcks -= MessageAck;
                producer.BasicNacks -= MessageNAck;

                waitAck.Clear();

                producer = null;
            } else if (infoChannel != null && infoChannel.IsClosed)
                infoChannel = null;
            Debug.Debug.Error("信道断开:" + reason);
        }


        /// <summary>
        /// 消息被返回
        /// </summary>
        /// <param name="model"></param>
        /// <param name="args"></param>
        private void MessageReturn(IModel model, BasicReturnEventArgs args) {

            if (args.BasicProperties != null) {
                string msgid = args.BasicProperties.MessageId;
                Message msg = MessagePool.GetMessage(msgid);//  new Message(msgid);
                msg.noAck = true;
                msg.body = args.Body;
                msg.channel = model;
                msg.exchange = args.Exchange;
                msg.routingkey = args.RoutingKey;
                msg.userData = args.BasicProperties.Headers;

                notRoutingOk?.Invoke(msg);
            } else {
                Debug.Debug.Error("怎么可能BasicProperties 不存在");
            }

            Debug.Debug.Error("消息被撤回,ReplyText:" + args.Exchange + "," + args.RoutingKey +
                "," + args.ReplyText + ",ReplyCode:" + args.ReplyCode + ",body:" + Encoding.UTF8.GetString(args.Body));
        }

        /// <summary>
        /// 声明路由
        /// </summary>
        public void DeclareExchange(ExchangeInfo[] infos) {
            if (infos != null) {
                int len = infos.Length;
                for (int i = 0; i < len; i++) {
                    ExchangeInfo exchange = infos[i];
                    IDictionary<string, object> properties = new Dictionary<string, object>();
                    if (!string.IsNullOrEmpty(exchange.alternate_exchange)) {
                        properties.Add("alternate-exchange", exchange.alternate_exchange);
                    }

                    if (exchange.delExists)
                        DelExchange(exchange.name);

                    //Debug.Debug.Log("进入锁44444"+Thread.CurrentThread.ManagedThreadId);
                    lock (lockobj) {
                        producer.ExchangeDeclare(exchange.name, exchange.type, exchange.durable, exchange.autodelete, properties);
                    }
                    //Debug.Debug.Log("退出锁4444");
                }
            }
        }

        /// <summary>
        /// 声明队列
        /// </summary>
        public QueueDeclareOk DeclareQueue() {
            QueueDeclareOk result = null;
            //Debug.Debug.Log("进入锁3333"+Thread.CurrentThread.ManagedThreadId);
            lock (lockobj) {
                try {
                    result = producer.QueueDeclare();
                } catch {
                    return null;
                }
            }
            //Debug.Debug.Log("退出锁3333");
            return result;
        }

        /// <summary>
        ///声明队列
        /// </summary>
        public void DeclareQueue(QueueInfo[] infos) {
            if (infos != null) {
                int len = infos.Length;
                for (int i = 0; i < len; i++) {
                    QueueInfo queue = infos[i];
                    IDictionary<string, object> properties = new Dictionary<string, object>();
                    if (!string.IsNullOrEmpty(queue.x_dead_letter_exchange)) {
                        properties.Add("x-dead-letter-exchange", queue.x_dead_letter_exchange);
                    }
                    if (!string.IsNullOrEmpty(queue.x_dead_routing_key)) {
                        properties.Add("x-dead-routing-key", queue.x_dead_routing_key);
                    }
                    if (queue.x_expires > 0) {
                        properties.Add("x-expires", queue.x_expires);
                    }
                    if (queue.x_max_length > 0) {
                        properties.Add("x-max-length", queue.x_max_length);
                    }
                    if (queue.x_max_length_bytes > 0) {
                        properties.Add("x-max-length-bytes", queue.x_max_length_bytes);
                    }
                    if (queue.x_max_priority > 0) {
                        properties.Add("x-max-priority", queue.x_max_priority);
                    }
                    if (queue.x_message_ttl > 0) {
                        properties.Add("x-message-ttl", queue.x_message_ttl);
                    }
                    if (!string.IsNullOrEmpty(queue.x_overflow)) {
                        properties.Add("x-overflow", queue.x_overflow);
                    }
                    if (!string.IsNullOrEmpty(queue.x_queue_mode)) {
                        properties.Add("x-queue-mode", queue.x_queue_mode);
                    }
                    if (queue.delExists)
                        DelQueue(queue.name);

                    //Debug.Debug.Log("进入锁7454354354"+Thread.CurrentThread.ManagedThreadId);
                    lock (lockobj) {
                        producer.QueueDeclare(queue.name, queue.durable, queue.exclusive, queue.autodelete, properties);
                        if (queue.startClear)
                            producer.QueuePurge(queue.name);
                    }
                    //Debug.Debug.Log("退出锁7454354354"+Thread.CurrentThread.ManagedThreadId);
                }
            }
        }

        /// <summary>
        /// 绑定
        /// </summary>
        public void Bind(BindInfo[] infos) {
            if (infos != null) {
                int len = infos.Length;
                for (int i = 0; i < len; i++) {
                    BindInfo bind = infos[i];
                    switch (bind.bindType) {
                        case "queue":
                            //Debug.Debug.Log("进入锁asdasfasf"+Thread.CurrentThread.ManagedThreadId);
                            lock (lockobj) {
                                producer.QueueBind(bind.destination, bind.source, bind.routingkey);
                            }
                            //Debug.Debug.Log("退出锁asdasfasf"+Thread.CurrentThread.ManagedThreadId);
                            break;
                        case "exchange":
                            //Debug.Debug.Log("进入锁asdsafasfasd"+Thread.CurrentThread.ManagedThreadId);
                            lock (lockobj) {
                                producer.ExchangeBind(bind.destination, bind.source, bind.routingkey);
                            }
                            //Debug.Debug.Log("退出锁asdsafasfasd"+Thread.CurrentThread.ManagedThreadId);
                            break;
                        default:
                            throw new ArgumentException("BindType value is not queue/exchange");
                    }
                }
            }
        }

        /// <summary>
        /// 获取队列信息
        /// </summary>
        /// <param name="name">名称</param>
        /// <returns>队列信息</returns>
        public QueueDeclareOk GetQueueInfo(string name) {
            try {
                IModel channel = connection.CreateModel();
                var result = channel.QueueDeclarePassive(name);
                return result;
            } catch {
                return null;
            }
        }

        /// <summary>
        /// 交换机存在不存在
        /// </summary>
        /// <param name="name">交换机名称</param>
        /// <returns>true 表示存在 false 可能是网络故障，或者不存在</returns>
        public bool ExchangeExists(string name) {
            try {
                IModel channel = connection.CreateModel();
                channel.ExchangeDeclarePassive(name);
                return true;
            } catch {
                return false;
            }
        }

        private readonly ConcurrentBag<RpcClient> rpcs = new ConcurrentBag<RpcClient>();

        /// <summary>
        /// 创建RPC客户端
        /// </summary>
        /// <returns></returns>
        public RpcClient CreateRpcClient(ushort prefetchCount = 3) {
            var rpc = new RpcClient(config, prefetchCount);
            rpcs.Add(rpc);
            return rpc;
        }

        /// <summary>
        /// 创建信道
        /// </summary>
        /// <returns></returns>
        internal IModel CreateIModel() {
            return connection.CreateModel();
        }

        /// <summary>
        /// 获取消息属性
        /// </summary>
        /// <returns></returns>
        public IBasicProperties GetBasicProperties() {
            try {

                IBasicProperties result = null;
                //Debug.Debug.Log("进入锁6666" + Thread.CurrentThread.ManagedThreadId);
                lock (lockobj) {
                    //Debug.Debug.Log("生产者:" + (producer == null));
                    if (producer == null)
                        CreateProducer();
                    if (producer == null)
                        result = null;
                    else
                        result = producer.CreateBasicProperties();
                }
                return result;
            } catch (Exception e) {
                Debug.Debug.Error("获取消息属性出错:" + e.ToString());
                return null;
            } finally {
                //Debug.Debug.Log("退出锁6666");
            }
        }

        /// <summary>
        /// 无返回值的发送,按配置来决定是异步还是同步确认
        /// </summary>
        public void BasicPublish(Message msg, bool isAutoFree) {
            BasicPublish(msg, config.AsynConfirm, isAutoFree);
        }

        /// <summary>
        ///  有返回值的发送_必须是同步，马上需要知道结果
        /// </summary>
        /// <returns></returns>
        public bool BasicPublish_Result(Message msg, bool isAutoFree) {
            return BasicPublish(msg, false, isAutoFree);
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="msg">消息</param>
        /// <param name="isAsyn">isAsyn 是否异步</param>
        /// <param name="isAutoFree">是否自动释放</param>
        private bool BasicPublish(Message msg, bool isAsyn, bool isAutoFree) {
            if (msg == null || msg.body == null || msg.body.Length == 0) {
                Debug.Debug.Error("发送数据为空!" + new System.Diagnostics.StackTrace().ToString());
                return false;
            }

            msg.isAutoFree = isAutoFree;
            bool isNotice = false;
            bool result = false;
            try {
                //Debug.Debug.Log("进入锁1111" + Thread.CurrentThread.ManagedThreadId);
                lock (lockobj) {
                    //Debug.Debug.Log("处理发送!" + (producer == null));
                    if (producer == null) {
                        CreateProducer();
                    }
                    //Debug.Debug.Log("111");
                    if (producer == null) {
                        isNotice = true;
                        msg.success = false;
                    } else {
                        msg.channel = producer;

                        msg.noAck = true;
                        msg.deliveryTag = producer.NextPublishSeqNo;
                        if (msg.propertis == null)
                            msg.propertis = producer.CreateBasicProperties();
                        msg.propertis.MessageId = msg.id;
                        msg.propertis.Headers = msg.userData;
                        //持久化
                        msg.propertis.SetPersistent(msg.durable);
                        //Debug.Debug.Info($"读取到过期时间:{msg.propertis.Expiration}");
                        if (!msg.durable)
                        {
                            msg.propertis.Expiration = "3000";
                        }

                        //Debug.Debug.Log("发送消息:" + msg.bodystr + "," + msg.exchange + "," + msg.propertis + "," + msg.mandatory);

                        producer.BasicPublish(msg.exchange, msg.routingkey, msg.mandatory, false, msg.propertis, msg.body);
                        if (!isAsyn) {//同步
                                      //Debug.Debug.Log("同步等待!");
                            msg.success = producer.WaitForConfirms();
                            result = msg.success;
                            if (!msg.success) {
                                Console.WriteLine(msg.success);
                            }
                            isNotice = true;

                        } else { //异步 //添加到列表
                            if (producer.IsClosed) {
                                isNotice = true;
                                msg.success = false;
                            } else {
                                //Debug.Debug.Log("异步发送:" + msg.deliveryTag);

                                //Debug.Debug.Log("得到数据:" + waitAck.Count);
                                waitAck.GetOrAdd(msg.deliveryTag, msg);

                                //Debug.Debug.Log("得到数据:" + waitAck.Count);

                            }
                        }
                    }
                }
            } catch (Exception e) {
                Debug.Debug.Warning("发送消息失败:" + e.ToString() + Environment.NewLine + msg.ToString());
                isNotice = true;
                msg.success = false;
            }

            if (isNotice) {
                if (msg.SendResult != null)
                    msg.SendResult(msg);
                else
                    SendResult?.Invoke(msg);

                if (msg.isAutoFree)
                    MessagePool.PutMessage(ref msg);
            }

            return result;
        }

        /// <summary>
        /// 删除队列
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="ifUnused">是否没有用户才删除</param>
        /// <param name="ifEmpty">是否是空队列才删除</param>
        /// <returns>uint 0表示删除失败 大于等于1表示删除成功 队列中消息数量+1</returns>
        public uint DelQueue(string queue, bool ifUnused = false, bool ifEmpty = false) {
            IModel model = connection.CreateModel();
            uint result = 0;
            try {
                result = model.QueueDelete(queue, ifUnused, ifEmpty);
                result += 1;
            } catch (Exception e) {
                Debug.Debug.Warning("删除队列异常:" + e.ToString());
            }
            return result;
        }

        /// <summary>
        /// 删除交换机
        /// </summary>
        /// <param name="exchange">交换机</param>
        /// <param name="ifUnused">是否没用户才删除</param>
        /// <returns>true 表示删除成功 false 队列不存在</returns>
        public bool DelExchange(string exchange, bool ifUnused = false) {
            IModel model = connection.CreateModel();
            try {
                model.ExchangeDelete(exchange, ifUnused);
            } catch (Exception e) {
                Debug.Debug.Warning("删除交换机异常:" + e.ToString());
                return false;
            }
            return true;
        }

        /// <summary>
        /// 获取消息
        /// </summary>
        /// <param name="queue">队列</param>
        /// <param name="noAck">true 表示自动确认 false 表示手动确认</param>
        /// <returns></returns>
        public Message GetMessage(string queue, bool noAck = true) {
            //Debug.Debug.Log("进入锁2222" + Thread.CurrentThread.ManagedThreadId);
            lock (lockobj) {
                if (infoChannel == null)
                    CreateInfoChannel();
                if (infoChannel == null)
                    return null;
            }
            //Debug.Debug.Log("退出锁2222");
            IModel key = infoChannel;
            Message msg = null;
            lock (key) {
                try {
                    BasicGetResult bgr = key.BasicGet(queue, noAck);

                    if (bgr != null) {
                        string msgid = null;
                        IDictionary<string, object> userdata = null;
                        if (bgr.BasicProperties == null) {
                            Debug.Debug.Error("这个消息竟然没有属性!");
                        } else {
                            msgid = bgr.BasicProperties.MessageId;
                            userdata = bgr.BasicProperties.Headers;
                        }
                        msg = MessagePool.GetMessage(msgid);// new Message(msgid);
                        msg.noAck = noAck;
                        msg.exchange = bgr.Exchange;
                        msg.routingkey = bgr.RoutingKey;
                        msg.propertis = bgr.BasicProperties;
                        msg.body = bgr.Body;
                        msg.deliveryTag = bgr.DeliveryTag;
                        msg.channel = key;
                        msg.redelivered = bgr.Redelivered;
                        msg.messageCount = bgr.MessageCount;
                        msg.userData = userdata;
                    }
                } catch (Exception e) {
                    Debug.Debug.Error("获取消息出错:" + e.ToString());
                    return null;
                }
            }
            return msg;
        }

        /// <summary>
        /// 添加消费者
        /// </summary>
        /// <param name="consumer">消费者</param>
        public void AddConsumer(Consumer consumer) {
            IModel model = connection.CreateModel();
            consumer.channel = model;
            if (string.IsNullOrEmpty(consumer.consumertag))
                consumer.consumertag = Guid.NewGuid().ToString();

            consumer.consumertag = model.BasicConsume(consumer.queue, consumer.noAck, consumer.consumertag,
                consumer.noLoacl, consumer.exclusive, consumer.arguments, new BasicConsumer(consumer));

            if (consumer.prefetchCount > 0) {
                model.BasicQos(0, consumer.prefetchCount, true);
            }

        }

        /// <summary>
        /// 移出消费者
        /// </summary>
        /// <param name="consumer"></param>
        public void RemoveConsumer(Consumer consumer) {
            if (consumer == null || consumer.channel == null || consumer.channel.IsClosed)
                return;
            consumer.channel.BasicCancel(consumer.consumertag);
        }

        /// <summary>
        /// 关闭
        /// </summary>
        public void Close() {
            if (connection != null) {
                connection.Dispose();
                connection = null;
            }
            foreach (var item in rpcs) {
                item.Close();
            }
        }


        /// <summary>
        /// 消费者类
        /// </summary>
        private class BasicConsumer : DefaultBasicConsumer {

            public Consumer consumer;

            public BasicConsumer(Consumer consumer) {
                this.consumer = consumer;
            }

            public override void HandleBasicCancel(string consumerTag) {
                base.HandleBasicCancel(consumerTag);
            }

            public override void HandleBasicConsumeOk(string consumerTag) {
                base.HandleBasicConsumeOk(consumerTag);
            }

            public override void HandleBasicCancelOk(string consumerTag) {
                base.HandleBasicCancelOk(consumerTag);
            }

            public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body) {
                string msgid = null;
                IDictionary<string, object> userdata = null;
                if (properties == null) {
                    Debug.Debug.Error("消费者接收消息异常,没有收到消息id");
                } else {
                    msgid = properties.MessageId;
                    userdata = properties.Headers;
                }

                Message message = MessagePool.GetMessage(msgid); // new Message(msgid);
                message.exchange = exchange;
                message.routingkey = routingKey;
                message.noAck = consumer.noAck;
                message.propertis = properties;
                message.redelivered = redelivered;
                message.deliveryTag = deliveryTag;
                message.channel = consumer.channel;
                message.body = body;
                message.userData = userdata;

                //去处理消息
                consumer.Receive?.Invoke(message);
                //base.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, properties, body);
            }

            public override void HandleModelShutdown(IModel model, ShutdownEventArgs reason) {
                base.HandleModelShutdown(model, reason);
            }
        }


    }


}
#endif