using System;
using System.Text;
using System.Threading;



namespace SFI.COMM.LIB.Redis
{
    public delegate void RedisSubEventHandler(object sender, RedisSubEventArgs e);

    public class RedisSubEventArgs : EventArgs
    {
        public string kind;
        public string pattern;
        public string channel;
        public object message;  // byte [] or integer depending on kind; or override On...
    }

    public class RedisSub : RedisClient
    {
        Thread worker = null;
        bool continueWorking = false;

        public RedisSub(string host, int port, string password) : base(host, port, password)
        {
        }

        public RedisSub(string host, string password) : base(host, password)
        {
        }

        public RedisSub(string password) : base(password)
        {
        }

        #region Redis commands
        public bool Subscribe(params string[] channels)
        {
            return SendSubCommand("SUBSCRIBE", channels);
        }

        public bool PSubscribe(params string[] patterns)
        {
            return SendSubCommand("PSUBSCRIBE", patterns);
        }

        public bool Unsubscribe(params string[] channels)
        {
            return SendSubCommand("UNSUBSCRIBE", channels);
        }

        public bool PUnsubscribe(params string[] patterns)
        {
            return SendSubCommand("PUNSUBSCRIBE", patterns);
        }

        public bool Publish()
        {
            var patterns = new string[] { @"foo", "\"Test10\"" };
            return SendSubCommand("PUBLISH", patterns);
        }
        #endregion

        #region Event handlers
        public event RedisSubEventHandler MessageReceived;
        public event RedisSubEventHandler SubscribeReceived;
        public event RedisSubEventHandler UnsubscribeReceived;

        protected virtual void OnMessageReceived(RedisSubEventArgs e)
        {
            try
            {
                if (MessageReceived != null)
                    MessageReceived(this, e);
            }
            catch (Exception ex)
            {

            }
            
        }

        protected virtual void OnSubscribeReceived(RedisSubEventArgs e)
        {
            try
            {
                if (SubscribeReceived != null)
                    SubscribeReceived(this, e);
            }
            catch (Exception ex)
            {

            }

            
        }

        protected virtual void OnUnsubscribeReceived(RedisSubEventArgs e)
        {
            try
            {
                if (UnsubscribeReceived != null)
                    UnsubscribeReceived(this, e);
            }
            catch (Exception ex)
            {

            }

            
        }
        #endregion

        /// <summary>
        /// Send a (un)subscribe command and start/stop listening.
        /// </summary>
        public bool SendSubCommand(string cmd, params string[] args)
        {
            if (SendCommand(cmd, args))
            {
                // command was sent; make sure we are listening
                if (!continueWorking)
                {
                    Log("C", "# start listening");
                    continueWorking = true;
                    worker = new Thread(SubWorker);
                    worker.Start();
                }
            }
            else
            {
                // send failed; stop listening
                if (continueWorking)
                {
                    Log("C", "# stop listening");
                    continueWorking = false;
                    worker.Join();
                    worker = null;
                }
            }
            return continueWorking;
        }

        public bool SendSubCommandFree(string cmd, params string[] args)
        {
            SendCommand(cmd, args);
            try
            {
                object[] data = ReadMixedArray();
            }
            catch (Exception ex)
            {
                Console.WriteLine("free command error : " + ex.Message);
            }
            return true;
        }


        /// <summary>
        /// Worker method to handle incoming messages from Redis.
        /// </summary>
        void SubWorker()
        {
            object[] data = null;
            while (continueWorking)
            {
                try
                {
                    data = ReadMixedArray();
                }
                catch (Exception ex)
                {
                    if (continueWorking)
                    { 
                        //throw ex;
                    }
                    else
                    {
                        Log("C", "# break listening");
                        break;
                    }
                }


                try
                {
                    if (data != null && data.Length >= 3)
                    {
                        RedisSubEventArgs e = new RedisSubEventArgs();
                        e.message = data[data.Length - 1];
                        e.kind = Encoding.UTF8.GetString(data[0] as byte[]);
                        switch (e.kind)
                        {
                            case "message":
                                if (MessageReceived == null)
                                    break;
                                e.channel = Encoding.UTF8.GetString(data[1] as byte[]);
                                e.message = Encoding.UTF8.GetString(data[data.Length - 1] as byte[]);
                                OnMessageReceived(e);
                                break;
                            case "subscribe":
                                if (SubscribeReceived == null) break;
                                e.channel = Encoding.UTF8.GetString(data[1] as byte[]);
                                OnSubscribeReceived(e);
                                break;
                            case "unsubscribe":
                                if (UnsubscribeReceived == null) break;
                                e.channel = Encoding.UTF8.GetString(data[1] as byte[]);
                                OnUnsubscribeReceived(e);
                                break;
                            case "pmessage":
                                if (MessageReceived == null) break;
                                if (data.Length != 4)
                                    throw new Exception("Received invalid pmessage with " +
                                        data.Length + " elements");
                                e.pattern = Encoding.UTF8.GetString(data[1] as byte[]);
                                e.channel = Encoding.UTF8.GetString(data[2] as byte[]);
                                OnMessageReceived(e);
                                break;
                            case "psubscribe":
                                if (SubscribeReceived == null) break;
                                e.pattern = Encoding.UTF8.GetString(data[1] as byte[]);
                                OnSubscribeReceived(e);
                                break;
                            case "punsubscribe":
                                if (UnsubscribeReceived == null) break;
                                e.pattern = Encoding.UTF8.GetString(data[1] as byte[]);
                                OnUnsubscribeReceived(e);
                                break;
                            default:
                                throw new Exception("Received message of unsupported kind " + e.kind);
                        }
                    }
                }
                catch (Exception ex)
                {
                    //Log("hata", ex.ToString());
                }


                    //throw new Exception("Received unexpected message with " +
                        //data.Length + " elements");
                 


            }
        }

        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                continueWorking = false;
                base.Dispose(true); // close socket
                if (worker != null)
                    worker.Join();
                worker = null;
            }
            else
                base.Dispose(false);
        }
    }
}
