Reactive Cloud Queue

AzureQueue_T.cs

/* Reactive Cloud Queue */



namespace AzureTrader.Messaging
{
    
    using System;
    using System.Collections;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.IO;
    using System.Linq;
    using System.Reactive;
    using System.Reactive.Concurrency;
    using System.Reactive.Linq;
    using System.Reactive.Subjects;
    using System.Reactive.Threading.Tasks;
    using System.Runtime.Serialization.Json;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;

    using Microsoft.WindowsAzure;
    using Microsoft.WindowsAzure.StorageClient;

    using ReactiveDemo.Messaging;
    using Extensions;


    public class AzureQueue<T> : ISubject<T, ILockedMessage<T>>
    {
        private class AzureQueueLockedMessage : ILockedMessage<T>
        {
            private CloudQueueMessage message;

            private readonly AzureQueue<T> parent;

            private Lazy<T> value;

            public AzureQueueLockedMessage(AzureQueue<T> parent, DataContractJsonSerializer serializer, CloudQueueMessage message)
            {
                this.message = message;
                this.parent = parent;
                this.value = new Lazy<T>(() => (T)serializer.ReadObject(new MemoryStream(message.AsBytes)));
            }

            public void Complete()
            {
                this.parent.messagesForDeletion.OnNext(this.message);
            }

            public void Defer()
            {
                // Not implemented for Azure Queues
            }

            public void Unlock()
            {
                // Do Nothing, no API to release peek lock
            }

            public string LockToken
            {
                get
                {
                    return message.Id + "|" + message.PopReceipt;
                }
            }

            public T Value
            {
                get
                {
                    return value.Value;
                }
            }
        }

        public AzureQueue(string queueAddress)
        {
            CloudStorageAccount.SetConfigurationSettingPublisher(
                (configName, configSettingPublisher) =>
                    {
                        var connectionString = DataConnectionSettings.ConnectionString;
                        configSettingPublisher(connectionString);
                    });

            var cloudStorageAccount = CloudStorageAccount.FromConfigurationSetting("DataConnectionString");
            var cloudQueueClient = cloudStorageAccount.CreateCloudQueueClient();

            this.cloudQueue = cloudQueueClient.GetQueueReference(queueAddress.ToLower());
            this.cloudQueue.CreateIfNotExist();

            // Receive messages and de-multiplex
            var messagesFromQueue = Observable.FromAsyncPattern<int, TimeSpan, IEnumerable<CloudQueueMessage>>(
                this.cloudQueue.BeginGetMessages, 
                this.cloudQueue.EndGetMessages);

            var receiveDataContractSerializer = new DataContractJsonSerializer(typeof(T));
            this.sendingDataContractSerializer = new DataContractJsonSerializer(typeof(T));
            
            int backoutDelay = 1;
            this.messages =  (from messages in Observable.Defer(() => messagesFromQueue(32, TimeSpan.FromSeconds(10))).Do(m => TraceMessages(this.cloudQueue.Name, m))
                              // Mutate the backout delay, multiplying by 8 every time there are no messages (up to a 4 second delay)
                              let thisDelay = GetUpdatedDelay(messages.Count(), ref backoutDelay)
                              // Add a delay to the end of the messages
                              from message in messages.ToObservable()
                                                     .Concat(Observable.Empty<CloudQueueMessage>()
                                                      .Delay(TimeSpan.FromMilliseconds(backoutDelay), Globals.Scheduler))
                              select message)
                              .Repeat()
                              .ObserveOn(Globals.Scheduler)
                              .Select(m => new AzureQueueLockedMessage(this, receiveDataContractSerializer, m));

            var asyncDelete = Observable.FromAsyncPattern<CloudQueueMessage>(this.cloudQueue.BeginDeleteMessage, this.cloudQueue.EndDeleteMessage);

            this.messageDeletionTask = this.messagesForDeletion
                .TakeUntil(this.stopSignalled)
                .Select(v => asyncDelete(v).CatchWithTrace(Globals.TraceSource)) // If it fails, just ignore the failure
                //// Send up to 5 deletes at a time
                .Merge(5)
                //// Store as a task, to allow waiting on completion
                .DefaultIfEmpty()
                .ToTask();
        }

        private int GetUpdatedDelay(int messageCount, ref int backoutDelay)
        {
            if (messageCount == 0)
            {
                backoutDelay = Math.Min(backoutDelay*4, 4096);
            }
            else if (messageCount > 16)
            {
                backoutDelay = 1;
            }
            else
            {
                if (backoutDelay > 1)
                {
                    backoutDelay = backoutDelay/4;
                }
            }
            return backoutDelay;
        }

        private void TraceMessages(string name, IEnumerable<CloudQueueMessage> cloudQueueMessages)
        {
            var count = cloudQueueMessages.Count();
            Debug.WriteLine("{0} received {1} messages", name, count);
        }

        public void Reset()
        {
            this.cloudQueue.Clear();
        }

        ISubject<CloudQueueMessage, CloudQueueMessage> messagesForDeletion = Subject.Synchronize(new Subject<CloudQueueMessage>(), Globals.Scheduler);

        private Task<Unit> messageDeletionTask;

        private ReplaySubject<bool> stopSignalled = new ReplaySubject<bool>(1);

        private CloudQueue cloudQueue;

        private DataContractJsonSerializer sendingDataContractSerializer;

        private IObservable<AzureQueueLockedMessage> messages;

        public string Name
        {
            get
            {
                return this.cloudQueue.Name;
            }
        }

        public void Dispose()
        {
            this.stopSignalled.OnNext(true);
            this.stopSignalled.OnCompleted();
            this.messageDeletionTask.Wait();
        }

        #region Implementation of IObserver<in T>

        public void OnNext(T value)
        {
            var ms = new MemoryStream();
            this.sendingDataContractSerializer.WriteObject(ms, value);
            ms.Close();

            var m = new CloudQueueMessage(Encoding.UTF8.GetString(ms.ToArray()));
            this.cloudQueue.AddMessage(m);
        }

        public void OnError(Exception error)
        {
            Trace.TraceError("Error received for queue " + this.cloudQueue.Name+ ": " + error);
            this.Dispose();
        }

        public void OnCompleted()
        {
            this.Dispose();
        }

        #endregion

        #region Implementation of IObservable<out T>

        public IDisposable Subscribe(IObserver<ILockedMessage<T>> observer)
        {
            return this.messages.Subscribe(observer);
        }

        #endregion

        public void Clear()
        {
            this.cloudQueue.Clear();
        }

    }
}