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.Generic;
      11             : using System.Threading.Tasks;
      12             : using cdmdotnet.Logging;
      13             : using Microsoft.ServiceBus.Messaging;
      14             : 
      15             : namespace Cqrs.Azure.ServiceBus
      16             : {
      17             :         /// <summary>
      18             :         /// A default implementation of <see cref="IEventProcessor"/> suitable for most situations and conditions.
      19             :         /// </summary>
      20             :         internal class DefaultEventProcessor : IEventProcessor
      21             :         {
      22             :                 protected ILogger Logger { get; private set; }
      23             : 
      24             :                 protected Action<PartitionContext, EventData> ReceiverMessageHandler { get; private set; }
      25             : 
      26             :                 /// <summary>
      27             :                 /// Initializes a new instance of the <see cref="DefaultEventProcessor"/> class.
      28             :                 /// </summary>
      29           1 :                 public DefaultEventProcessor(ILogger logger, Action<PartitionContext, EventData> receiverMessageHandler)
      30             :                 {
      31             :                         Logger = logger;
      32             :                         ReceiverMessageHandler = receiverMessageHandler;
      33             :                 }
      34             : 
      35             :                 #region Implementation of IEventProcessor
      36             : 
      37             :                 /// <summary>
      38             :                 /// Initializes the Event Hub processor instance. This method is called before any event data is passed to this processor instance.
      39             :                 /// </summary>
      40             :                 /// <param name="context">Ownership information for the partition on which this processor instance works. Any attempt to call <see cref="M:Microsoft.ServiceBus.Messaging.PartitionContext.CheckpointAsync"/> will fail during the Open operation.</param>
      41             :                 /// <returns>
      42             :                 /// The task that indicates that the Open operation is complete.
      43             :                 /// </returns>
      44           1 :                 public virtual Task OpenAsync(PartitionContext context)
      45             :                 {
      46             :                         Logger.LogInfo("Open Async");
      47             :                         return Task.FromResult<object>(null);
      48             :                 }
      49             : 
      50             :                 /// <summary>
      51             :                 /// Asynchronously processes the specified context and messages. This method is called when there are new messages in the Event Hubs stream. Make sure to checkpoint only when you are finished processing all the events in each batch.
      52             :                 /// </summary>
      53             :                 /// <param name="context">Ownership information for the partition on which this processor instance works.</param>
      54             :                 /// <param name="messages">A batch of Event Hubs events.</param>
      55             :                 /// <returns>
      56             :                 /// The task that indicates that <see cref="M:Microsoft.ServiceBus.Messaging.IEventProcessor.ProcessEventsAsync(Microsoft.ServiceBus.Messaging.PartitionContext,System.Collections.Generic.IEnumerable{Microsoft.ServiceBus.Messaging.EventData})"/> is complete.
      57             :                 /// </returns>
      58           1 :                 public virtual Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
      59             :                 {
      60             :                         Task results = new TaskFactory().StartNew(() =>
      61             :                         {
      62             :                                 foreach (EventData eventData in messages)
      63             :                                         ReceiverMessageHandler(context, eventData);
      64             :                         });
      65             : 
      66             :                         return results;
      67             :                 }
      68             : 
      69             :                 /// <summary>
      70             :                 /// Called when the ownership of partition moves to a different node for load-balancing purpose, or when the host is shutting down. Called in response to <see cref="M:Microsoft.ServiceBus.Messaging.EventHubConsumerGroup.UnregisterProcessorAsync(Microsoft.ServiceBus.Messaging.Lease,Microsoft.ServiceBus.Messaging.CloseReason)"/>.
      71             :                 /// </summary>
      72             :                 /// <param name="context">Partition ownership information for the partition on which this processor instance works. You can call <see cref="M:Microsoft.ServiceBus.Messaging.PartitionContext.CheckpointAsync"/> to checkpoint progress in the processing of messages from Event Hub streams.</param>
      73             :                 /// <param name="reason">The reason for calling <see cref="M:Microsoft.ServiceBus.Messaging.IEventProcessor.CloseAsync(Microsoft.ServiceBus.Messaging.PartitionContext,Microsoft.ServiceBus.Messaging.CloseReason)"/>.</param>
      74             :                 /// <returns>
      75             :                 /// A task indicating that the Close operation is complete.
      76             :                 /// </returns>
      77           1 :                 public virtual Task CloseAsync(PartitionContext context, CloseReason reason)
      78             :                 {
      79             :                         return Task.FromResult<object>(null);
      80             :                 }
      81             : 
      82             :                 #endregion
      83             :         }
      84             : }