C# Windows Azure Queue的操作

C# Windows Azure Queue的操作

namespace Axe.AzureStorage
{
    using System;
    using System.IO;
    using System.Runtime.Serialization.Formatters.Binary;
    using System.Threading;
    using System.Threading.Tasks;


    using Microsoft.WindowsAzure;
    using Microsoft.WindowsAzure.Storage;
    using Microsoft.WindowsAzure.Storage.Queue;


    public class WinAzureStorageAsync
    {
        private readonly CloudQueue queue;
        private readonly int timeoutSecond;
        private CloudQueueClient queueClient;
        public CloudQueueClient QueueClient
        {
            get
            {
                if (this.queueClient != null)
                    return this.queueClient;


                var storageAccount = CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
                this.queueClient = storageAccount.CreateCloudQueueClient();
                return this.queueClient;
            }
        }


        ////since each time fetch message is not a block operation
        ////so need to set a timeout & keep fetching , default is 3 seconds
        private const int SleepInterval = 100;


        public WinAzureStorageAsync(string queueName, int timeoutSecond = 3)
        {
            queueName = queueName.ToLower();
            this.queue = this.QueueClient.GetQueueReference(queueName);
            if (!this.QueueClient.GetQueueReference(queueName).Exists())
            {
                this.queue.CreateIfNotExists();
            }


            this.timeoutSecond = timeoutSecond;
        }


        public async Task GetMessage()
        {
            CloudQueueMessage message = null;


            var passed = 0;
            
            while (message == null && passed < this.timeoutSecond * 10 * SleepInterval)
            {
                message = await this.queue.GetMessageAsync();
                Thread.Sleep(SleepInterval);
                passed += SleepInterval;
            }
            if (message == null)
            {
                throw new TimeoutException("Get Message From Azure Queue Operation has been timeout");
            }
            await this.queue.DeleteMessageAsync(message);


            return message;
        }


        public async Task GetString()
        {
            var msg = await this.GetMessage();
            return msg.AsString;
        }


        public async Task GetBytes()
        {
            var msg = await this.GetMessage();
            return msg.AsBytes;
        }


        public T Get() where T : new()
        {
            var bytes = this.GetBytes();
            return this.BytesToT(bytes.Result);
        }


        public async Task Add(string message)
        {
            await this.queue.AddMessageAsync(new CloudQueueMessage(message));
        }


        public async Task Add(byte[] bytes)
        {
            await this.queue.AddMessageAsync(new CloudQueueMessage(bytes));
        }


        public void Add(T obj) where T : new()
        {
            var bytes = this.TToBytes(obj);
            this.Add(bytes);
        }


        /// 

        /// Note : this operation make takes around 40 seconds to complete, reference here:
        /// http://msdn.microsoft.com/library/azure/dd179387.aspx
        /// 
        /// 
        public async Task DeleteIfExists()
        {
            await this.queue.DeleteIfExistsAsync();
        }


        public async Task IsExist(string queueName)
        {
            queueName = queueName.ToLower();
            return await this.QueueClient.GetQueueReference(queueName).ExistsAsync();
        }


        public void ClearMessage()
        {
            this.queue.Clear();
        }


        private T BytesToT(byte[] bytes)
        {
            using (var ms = new MemoryStream())
            {
                ms.Write(bytes, 0, bytes.Length);
                var bf = new BinaryFormatter();
                ms.Position = 0;
                var x = bf.Deserialize(ms);
                return (T)x;
            }
        }


        private byte[] TToBytes(T obj)
        {
            var bf = new BinaryFormatter();
            using (var ms = new MemoryStream())
            {
                bf.Serialize(ms, obj);
                return ms.ToArray();
            }
        }
    }
}