Documentation Coverage Report
Current view: top level - Azure/Cqrs.Azure.EventHub - AzureQueuedEventBusReceiver.cs Hit Total Coverage
Version: 2.2 Artefacts: 4 4 100.0 %
Date: 2017-09-22

          Line data    Source code
       1             : #region Copyright
       2             : // // -----------------------------------------------------------------------
       3             : // // <copyright company="Chinchilla Software Limited">
       4             : // //   Copyright Chinchilla Software Limited. All rights reserved.
       5             : // // </copyright>
       6             : // // -----------------------------------------------------------------------
       7             : #endregion
       8             : 
       9             : using System;
      10             : using System.Collections.Concurrent;
      11             : using System.Collections.Generic;
      12             : using System.Text;
      13             : using System.Threading;
      14             : using cdmdotnet.Logging;
      15             : using Cqrs.Authentication;
      16             : using Cqrs.Configuration;
      17             : using Cqrs.Events;
      18             : using Microsoft.ServiceBus.Messaging;
      19             : using EventData = Microsoft.ServiceBus.Messaging.EventData;
      20             : using SpinWait = Cqrs.Infrastructure.SpinWait;
      21             : 
      22             : namespace Cqrs.Azure.ServiceBus
      23             : {
      24             :         /// <summary>
      25             :         /// A concurrent implementation of <see cref="AzureEventBusReceiver{TAuthenticationToken}"/> that resides in memory.
      26             :         /// </summary>
      27             :         /// <typeparam name="TAuthenticationToken">The <see cref="Type"/> of the authentication token.</typeparam>
      28             :         public class AzureQueuedEventBusReceiver<TAuthenticationToken> : AzureEventBusReceiver<TAuthenticationToken>
      29             :         {
      30             :                 /// <summary>
      31             :                 /// Tracks all queues.
      32             :                 /// </summary>
      33             :                 protected static ConcurrentDictionary<string, ConcurrentQueue<IEvent<TAuthenticationToken>>> QueueTracker { get; private set; }
      34             : 
      35             :                 /// <summary>
      36             :                 /// Gets the <see cref="ReaderWriterLockSlim"/>.
      37             :                 /// </summary>
      38             :                 protected ReaderWriterLockSlim QueueTrackerLock { get; private set; }
      39             : 
      40             :                 /// <summary>
      41             :                 /// Receives a <see cref="BrokeredMessage"/> from the event bus, identifies a key and queues it accordingly.
      42             :                 /// </summary>
      43           2 :                 public AzureQueuedEventBusReceiver(IConfigurationManager configurationManager, IMessageSerialiser<TAuthenticationToken> messageSerialiser, IAuthenticationTokenHelper<TAuthenticationToken> authenticationTokenHelper, ICorrelationIdHelper correlationIdHelper, ILogger logger, IAzureBusHelper<TAuthenticationToken> azureBusHelper)
      44             :                         : base(configurationManager, messageSerialiser, authenticationTokenHelper, correlationIdHelper, logger, azureBusHelper)
      45             :                 {
      46             :                         QueueTracker = new ConcurrentDictionary<string, ConcurrentQueue<IEvent<TAuthenticationToken>>>();
      47             :                         QueueTrackerLock = new ReaderWriterLockSlim();
      48             :                 }
      49             : 
      50             :                 /// <summary>
      51             :                 /// Receives a <see cref="BrokeredMessage"/> from the event bus, identifies a key and queues it accordingly.
      52             :                 /// </summary>
      53           2 :                 protected override void ReceiveEvent(PartitionContext context, EventData eventData)
      54             :                 {
      55             :                         // Do a manual 10 try attempt with back-off
      56             :                         for (int i = 0; i < 10; i++)
      57             :                         {
      58             :                                 try
      59             :                                 {
      60             :                                         Logger.LogDebug(string.Format("An event message arrived with the partition key '{0}', sequence number '{1}' and offset '{2}'.", eventData.PartitionKey, eventData.SequenceNumber, eventData.Offset));
      61             :                                         string messageBody = Encoding.UTF8.GetString(eventData.GetBytes());
      62             :                                         IEvent<TAuthenticationToken> @event = MessageSerialiser.DeserialiseEvent(messageBody);
      63             : 
      64             :                                         CorrelationIdHelper.SetCorrelationId(@event.CorrelationId);
      65             :                                         Logger.LogInfo(string.Format("An event message arrived with the partition key '{0}', sequence number '{1}' and offset '{2}' was of type {3}.", eventData.PartitionKey, eventData.SequenceNumber, eventData.Offset, @event.GetType().FullName));
      66             : 
      67             :                                         Type eventType = @event.GetType();
      68             : 
      69             :                                         string targetQueueName = eventType.FullName;
      70             : 
      71             :                                         try
      72             :                                         {
      73             :                                                 object rsn = eventType.GetProperty("Rsn").GetValue(@event, null);
      74             :                                                 targetQueueName = string.Format("{0}.{1}", targetQueueName, rsn);
      75             :                                         }
      76             :                                         catch
      77             :                                         {
      78             :                                                 Logger.LogDebug(string.Format("An event message arrived with the partition key '{0}', sequence number '{1}' and offset '{2}' was of type {3} but with no Rsn property.", eventData.PartitionKey, eventData.SequenceNumber, eventData.Offset, eventType));
      79             :                                                 // Do nothing if there is no rsn. Just use @event type name
      80             :                                         }
      81             : 
      82             :                                         CreateQueueAndAttachListenerIfNotExist(targetQueueName);
      83             :                                         EnqueueEvent(targetQueueName, @event);
      84             : 
      85             :                                         // remove the original message from the incoming queue
      86             :                                         context.CheckpointAsync(eventData);
      87             : 
      88             :                                         Logger.LogDebug(string.Format("An event message arrived and was processed with the partition key '{0}', sequence number '{1}' and offset '{2}'.", eventData.PartitionKey, eventData.SequenceNumber, eventData.Offset));
      89             :                                         return;
      90             :                                 }
      91             :                                 catch (Exception exception)
      92             :                                 {
      93             :                                         // Indicates a problem, unlock message in queue
      94             :                                         Logger.LogError(string.Format("An event message arrived with the partition key '{0}', sequence number '{1}' and offset '{2}' but failed to be process.", eventData.PartitionKey, eventData.SequenceNumber, eventData.Offset), exception: exception);
      95             : 
      96             :                                         switch (i)
      97             :                                         {
      98             :                                                 case 0:
      99             :                                                 case 1:
     100             :                                                         // 10 seconds
     101             :                                                         Thread.Sleep(10 * 1000);
     102             :                                                         break;
     103             :                                                 case 2:
     104             :                                                 case 3:
     105             :                                                         // 30 seconds
     106             :                                                         Thread.Sleep(30 * 1000);
     107             :                                                         break;
     108             :                                                 case 4:
     109             :                                                 case 5:
     110             :                                                 case 6:
     111             :                                                         // 1 minute
     112             :                                                         Thread.Sleep(60 * 1000);
     113             :                                                         break;
     114             :                                                 case 7:
     115             :                                                 case 8:
     116             :                                                 case 9:
     117             :                                                         // 3 minutes
     118             :                                                         Thread.Sleep(3 * 60 * 1000);
     119             :                                                         break;
     120             :                                         }
     121             :                                 }
     122             :                         }
     123             :                         // Eventually just accept it
     124             :                         context.CheckpointAsync(eventData);
     125             :                 }
     126             : 
     127             :                 /// <summary>
     128             :                 /// Adds the provided <paramref name="event"/> to the <see cref="QueueTracker"/> of the queue <paramref name="targetQueueName"/>.
     129             :                 /// </summary>
     130             :                 private void EnqueueEvent(string targetQueueName, IEvent<TAuthenticationToken> @event)
     131             :                 {
     132             :                         var queue = QueueTracker.GetOrAdd(targetQueueName, new ConcurrentQueue<IEvent<TAuthenticationToken>>());
     133             :                         queue.Enqueue(@event);
     134             :                 }
     135             : 
     136             :                 /// <summary>
     137             :                 /// Creates the queue of the name <paramref name="queueName"/> if it does not already exist,
     138             :                 /// the queue is attached to <see cref="DequeuAndProcessEvent"/> using a <see cref="Thread"/>.
     139             :                 /// </summary>
     140             :                 /// <param name="queueName">The name of the queue to check and create.</param>
     141           2 :                 protected void CreateQueueAndAttachListenerIfNotExist(string queueName)
     142             :                 {
     143             :                         if (!QueueTracker.ContainsKey(queueName))
     144             :                         {
     145             :                                 QueueTrackerLock.EnterWriteLock();
     146             :                                 try
     147             :                                 {
     148             :                                         if (!QueueTracker.ContainsKey(queueName))
     149             :                                         {
     150             :                                                 QueueTracker.TryAdd(queueName, new ConcurrentQueue<IEvent<TAuthenticationToken>>());
     151             :                                                 new Thread(() =>
     152             :                                                 {
     153             :                                                         Thread.CurrentThread.Name = queueName;
     154             :                                                         DequeuAndProcessEvent(queueName);
     155             :                                                 }).Start();
     156             :                                         }
     157             :                                 }
     158             :                                 catch (Exception exception)
     159             :                                 {
     160             :                                         Logger.LogError(string.Format("Processing a request to start a thread for the queue '{0}' failed.", queueName), exception: exception);
     161             :                                 }
     162             :                                 finally
     163             :                                 {
     164             :                                         QueueTrackerLock.ExitWriteLock();
     165             :                                 }
     166             :                         }
     167             :                 }
     168             : 
     169             :                 /// <summary>
     170             :                 /// Takes an <see cref="IEvent{TAuthenticationToken}"/> off the queue of <paramref name="queueName"/>
     171             :                 /// and calls <see cref="ReceiveEvent"/>. Repeats in a loop until the queue is empty.
     172             :                 /// </summary>
     173             :                 /// <param name="queueName">The name of the queue process.</param>
     174           2 :                 protected void DequeuAndProcessEvent(string queueName)
     175             :                 {
     176             :                         SpinWait.SpinUntil
     177             :                         (
     178             :                                 () =>
     179             :                                 {
     180             :                                         try
     181             :                                         {
     182             :                                                 ConcurrentQueue<IEvent<TAuthenticationToken>> queue;
     183             :                                                 if (QueueTracker.TryGetValue(queueName, out queue))
     184             :                                                 {
     185             :                                                         while (!queue.IsEmpty)
     186             :                                                         {
     187             :                                                                 IEvent<TAuthenticationToken> @event;
     188             :                                                                 if (queue.TryDequeue(out @event))
     189             :                                                                 {
     190             :                                                                         try
     191             :                                                                         {
     192             :                                                                                 CorrelationIdHelper.SetCorrelationId(@event.CorrelationId);
     193             :                                                                         }
     194             :                                                                         catch (Exception exception)
     195             :                                                                         {
     196             :                                                                                 Logger.LogError(string.Format("Trying to set the CorrelationId from the event type {1} for a request for the queue '{0}' failed.", queueName, @event.GetType()), exception: exception);
     197             :                                                                         }
     198             :                                                                         try
     199             :                                                                         {
     200             :                                                                                 AuthenticationTokenHelper.SetAuthenticationToken(@event.AuthenticationToken);
     201             :                                                                         }
     202             :                                                                         catch (Exception exception)
     203             :                                                                         {
     204             :                                                                                 Logger.LogError(string.Format("Trying to set the AuthenticationToken from the event type {1} for a request for the queue '{0}' failed.", queueName, @event.GetType()), exception: exception);
     205             :                                                                         }
     206             :                                                                         try
     207             :                                                                         {
     208             :                                                                                 ReceiveEvent(@event);
     209             :                                                                         }
     210             :                                                                         catch (Exception exception)
     211             :                                                                         {
     212             :                                                                                 Logger.LogError(string.Format("Processing the event type {1} for a request for the queue '{0}' failed.", queueName, @event.GetType()), exception: exception);
     213             :                                                                                 queue.Enqueue(@event);
     214             :                                                                         }
     215             :                                                                 }
     216             :                                                                 else
     217             :                                                                         Logger.LogDebug(string.Format("Trying to dequeue a event from the queue '{0}' failed.", queueName));
     218             :                                                         }
     219             :                                                 }
     220             :                                                 else
     221             :                                                         Logger.LogDebug(string.Format("Trying to find the queue '{0}' failed.", queueName));
     222             :                                                 Thread.Sleep(100);
     223             :                                         }
     224             :                                         catch (Exception exception)
     225             :                                         {
     226             :                                                 Logger.LogError(string.Format("Dequeuing and processing a request for the queue '{0}' failed.", queueName), exception: exception);
     227             :                                         }
     228             : 
     229             :                                         // Always return false to keep this spinning.
     230             :                                         return false;
     231             :                                 },
     232             :                                 sleepInMilliseconds: 1000
     233             :                         );
     234             :                 }
     235             : 
     236             :                 /// <summary>
     237             :                 /// The number of queues currently known.
     238             :                 /// </summary>
     239             :                 public int QueueCount
     240             :                 {
     241             :                         get
     242             :                         {
     243             :                                 QueueTrackerLock.EnterReadLock();
     244             :                                 try
     245             :                                 {
     246             :                                         return QueueTracker.Count;
     247             :                                 }
     248             :                                 finally
     249             :                                 {
     250             :                                         QueueTrackerLock.ExitReadLock();
     251             :                                 }
     252             :                         }
     253             :                 }
     254             : 
     255             :                 /// <summary>
     256             :                 /// The name of all currently known queues.
     257             :                 /// </summary>
     258             :                 public ICollection<string> QueueNames
     259             :                 {
     260             :                         get
     261             :                         {
     262             :                                 QueueTrackerLock.EnterReadLock();
     263             :                                 try
     264             :                                 {
     265             :                                         return QueueTracker.Keys;
     266             :                                 }
     267             :                                 finally
     268             :                                 {
     269             :                                         QueueTrackerLock.ExitReadLock();
     270             :                                 }
     271             :                         }
     272             :                 }
     273             :         }
     274             : }

Generated by: LCOV version 1.10