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

Generated by: LCOV version 1.10