Line data    Source code

       1             : #region Copyright
       2             : // // -----------------------------------------------------------------------
       3             : // // <copyright company="cdmdotnet Limited">
       4             : // //   Copyright cdmdotnet 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             :         internal class DefaultEventProcessor : IEventProcessor
      18             :         {
      19             :                 protected ILogger Logger { get; private set; }
      20             : 
      21             :                 protected Action<PartitionContext, EventData> ReceiverMessageHandler { get; private set; }
      22             : 
      23           0 :                 public DefaultEventProcessor(ILogger logger, Action<PartitionContext, EventData> receiverMessageHandler)
      24             :                 {
      25             :                         Logger = logger;
      26             :                         ReceiverMessageHandler = receiverMessageHandler;
      27             :                 }
      28             : 
      29             :                 #region Implementation of IEventProcessor
      30             : 
      31             :                 /// <summary>
      32             :                 /// Initializes the Event Hub processor instance. This method is called before any event data is passed to this processor instance.
      33             :                 /// </summary>
      34             :                 /// <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>
      35             :                 /// <returns>
      36             :                 /// The task that indicates that the Open operation is complete.
      37             :                 /// </returns>
      38           1 :                 public Task OpenAsync(PartitionContext context)
      39             :                 {
      40             :                         Logger.LogInfo("Open Async");
      41             :                         return Task.FromResult<object>(null);
      42             :                 }
      43             : 
      44             :                 /// <summary>
      45             :                 /// 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.
      46             :                 /// </summary>
      47             :                 /// <param name="context">Ownership information for the partition on which this processor instance works.</param>
      48             :                 /// <param name="messages">A batch of Event Hubs events.</param>
      49             :                 /// <returns>
      50             :                 /// 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.
      51             :                 /// </returns>
      52           1 :                 public Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
      53             :                 {
      54             :                         Task results = new TaskFactory().StartNew(() =>
      55             :                         {
      56             :                                 foreach (EventData eventData in messages)
      57             :                                         ReceiverMessageHandler(context, eventData);
      58             :                         });
      59             : 
      60             :                         return results;
      61             :                 }
      62             : 
      63             :                 /// <summary>
      64             :                 /// 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)"/>.
      65             :                 /// </summary>
      66             :                 /// <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>
      67             :                 /// <param name="reason">The reason for calling <see cref="M:Microsoft.ServiceBus.Messaging.IEventProcessor.CloseAsync(Microsoft.ServiceBus.Messaging.PartitionContext,Microsoft.ServiceBus.Messaging.CloseReason)"/>.</param>
      68             :                 /// <returns>
      69             :                 /// A task indicating that the Close operation is complete.
      70             :                 /// </returns>
      71           1 :                 public Task CloseAsync(PartitionContext context, CloseReason reason)
      72             :                 {
      73             :                         return Task.FromResult<object>(null);
      74             :                 }
      75             : 
      76             :                 #endregion
      77             :         }
      78             : }