Merge dev to master (#244)

* Fix EventHubsConnectionStringBuilder ctor crash (#219)

When supplying SharedAccessSignature as a parameter

* Fixing deceptive description of InitialOffsetProvider (#224)

* Aad token provider (#225)

* Adding AAD token support.

* improve lock

* Adding MSI provider.

* Some small changes

* Adddresing comments and couple small fixes.

* Remove string resource

* Remove test secrets

* Renaming paramater endpoint to endpointAddress

* Renaming paramater endpoint to endpointAddress

* Remove unused AsyncLock

* Remove redundant config name

* Remove using

* Remove usings

* Merge fix

* API name change

* Xmldoc type fix.

* Remove SecurityToken internal constructor

* Build error fix

* Cleanup code

* Pass temporary audience instead of empty string to SecurityToken.

* Make token type required

* Refactoring SharedAccessSignatureToken for cleanup

* Function renaming to carry UTC notion.

* Remove procted accesors from SecurityToken class properties.

* Adding receiver identifier support (#226)

* Fixing EventDataBatch with partition-key related issues. (#198)

* Fixing DataBatch partition-key usage

* Fixing EventDataBatch with partition-key related issues.

* Incorporating comments from latest team discussion.

* Removing a leftover from debug session.

* Adding maxMessageSize to CreateBatch API on clients.

* Aligning API with Java client.

* Fixing build error.

* Adding EH client changes to support sequence number receiver. (#230)

* EventData.SystemProperties to inherit Dictionary (#206)

* Adding BatchOptions

* Avoid application properties to overwrite IOT set properties.

* Remove batchoptions, not related to this change

* Avoid IOT reserved names to be used in the message propery bag.

* SystemProperties to inherit System.Dictionary.

* Return system property from dictionary instead of assigning them during message conversion.

* Throw if system property is missing.

* Fix merge errors

* Merge fix

* Remove unnecessary partitionid/partitionkey check

* Close receivers

* Adjusting class comment of EventDataBatch class (#213)

* Adjusting class comment of EventDataBatch class

Adjusting top comment, as SendBatch and SendBatchAsync methods don't exist on EventHubClient anymore, the methods are called Send and SendAsync and take either a single EventData or an IEnumerable<EventData>.
In addition the comment makes it not clear why a dev should use this class. Why not just create an IEnumerable<EventData> on your own? A dev should use this class because it takes care of the size limit, that's the main reason and it should be in the header.

* Update EventDataBatch.cs

Fixing the comment to make it XML-compliant.

* Update EventDataBatch.cs

Link EventHubClient in XML comment

* Add default batch size for receive handler. (#235)

* Add default batch size for receive handler.

* Rely on default MaxBatchSize in the unit tests

* XMLdoc correction

* Address James's comment on PR

* Fixed a typographical error (#239)

* invokeWhenNoEvents flag for Pump (#238)

* Adding invokeWhenNoEvents to pump and wiring up EH null callback on this.

* Fix default pump invoke on null behavior

* Stop-CancellationToken for PartitionPump (#187)

* Temporarily hiding AAD token provider API

Author: serkantkaraca

readme.md

@@ -43,7 +43,7 @@ consumers that filter and/or transform event streams and then forward them on to
 
 [https://github.com/Azure/azure-event-hubs/tree/master/samples](https://github.com/Azure/azure-event-hubs/tree/master/samples)
 
-### How to build client liblaries on Visual Studio 2017? 
+### How to build client libraries on Visual Studio 2017? 
 
 Make sure you have .NET Core Cross-Platform Development SDK installed. If it is missing, you can install the SDK by running Visual Studio Installer. See https://www.microsoft.com/net/core#windowscmd for VS 2017 and SDK installation.
 

src/Microsoft.Azure.EventHubs.Processor/Checkpoint.cs

@@ -13,7 +13,7 @@ public class Checkpoint
         /// </summary>
         /// <param name="partitionId">The partition ID for the checkpoint</param>
         public Checkpoint(string partitionId)
-            : this(partitionId, PartitionReceiver.StartOfStream, 0)
+            : this(partitionId, EventPosition.FromStart().Offset, 0)
         {
         }
 

src/Microsoft.Azure.EventHubs.Processor/EventHubPartitionPump.cs

@@ -55,7 +55,9 @@ protected override async Task OnOpenAsync()
                 // meaning it is safe to set the handler and start calling IEventProcessor.OnEvents.
                 // Set the status to running before setting the client handler, so the IEventProcessor.OnEvents can never race and see status != running.
                 this.PumpStatus = PartitionPumpStatus.Running;
-                this.partitionReceiver.SetReceiveHandler(this.partitionReceiveHandler);
+                this.partitionReceiver.SetReceiveHandler(
+                    this.partitionReceiveHandler,
+                    this.Host.EventProcessorOptions.InvokeProcessorAfterReceiveTimeout);
             }
 
             if (this.PumpStatus == PartitionPumpStatus.OpenFailed)
@@ -69,9 +71,10 @@ protected override async Task OnOpenAsync()
         async Task OpenClientsAsync() // throws EventHubsException, IOException, InterruptedException, ExecutionException
         {
             // Create new clients
-            object startAt = await this.PartitionContext.GetInitialOffsetAsync().ConfigureAwait(false);
+            EventPosition eventPosition = await this.PartitionContext.GetInitialOffsetAsync().ConfigureAwait(false);
             long epoch = this.Lease.Epoch;
-            ProcessorEventSource.Log.PartitionPumpCreateClientsStart(this.Host.Id, this.PartitionContext.PartitionId, epoch, startAt?.ToString());
+            ProcessorEventSource.Log.PartitionPumpCreateClientsStart(this.Host.Id, this.PartitionContext.PartitionId, epoch,
+                $"Offset:{eventPosition.Offset}, SequenceNumber:{eventPosition.SequenceNumber}, DateTime:{eventPosition.EnqueuedTimeUtc}");
             this.eventHubClient = EventHubClient.CreateFromConnectionString(this.Host.EventHubConnectionString);
 
             var receiverOptions = new ReceiverOptions()
@@ -81,24 +84,12 @@ async Task OpenClientsAsync() // throws EventHubsException, IOException, Interru
             };
 
             // Create new receiver and set options
-            if (startAt is string)
-            {
-                this.partitionReceiver = this.eventHubClient.CreateEpochReceiver(
-                    this.PartitionContext.ConsumerGroupName,
-                    this.PartitionContext.PartitionId,
-                    (string)startAt,
-                    epoch,
-                    receiverOptions);
-            }
-            else if (startAt is DateTime)
-            {
-                this.partitionReceiver = this.eventHubClient.CreateEpochReceiver(
-                    this.PartitionContext.ConsumerGroupName, 
-                    this.PartitionContext.PartitionId, 
-                    (DateTime)startAt, 
-                    epoch,
-                    receiverOptions);
-            }
+            this.partitionReceiver = this.eventHubClient.CreateEpochReceiver(
+                this.PartitionContext.ConsumerGroupName,
+                this.PartitionContext.PartitionId,
+                eventPosition,
+                epoch,
+                receiverOptions);
 
             this.partitionReceiver.PrefetchCount = this.Host.EventProcessorOptions.PrefetchCount;
 
@@ -146,7 +137,7 @@ public PartitionReceiveHandler(EventHubPartitionPump eventHubPartitionPump)
                 this.MaxBatchSize = eventHubPartitionPump.Host.EventProcessorOptions.MaxBatchSize;
             }
 
-            public int MaxBatchSize { get; }
+            public int MaxBatchSize { get; set; }
 
             public Task ProcessEventsAsync(IEnumerable<EventData> events)
             {

src/Microsoft.Azure.EventHubs.Processor/EventProcessorOptions.cs

@@ -38,7 +38,7 @@ public EventProcessorOptions()
             this.MaxBatchSize = 10;
             this.PrefetchCount = 300;
             this.ReceiveTimeout = TimeSpan.FromMinutes(1);
-            this.InitialOffsetProvider = partitionId => PartitionReceiver.StartOfStream;
+            this.InitialOffsetProvider = partitionId => EventPosition.FromStart();
         }
 
         /// <summary>
@@ -79,12 +79,11 @@ public bool EnableReceiverRuntimeMetric
         public int PrefetchCount { get; set; }
 
         /// <summary>
-        /// Get or sets the current function used to determine the initial offset at which to start receiving
-        /// events for a partition.
-        /// <para>A null return indicates that it is using the internal provider, which uses the last checkpointed
-        /// offset value (if present) or StartOfSTream (if not).</para>
+        /// Gets or sets a delegate which is used to get the initial position for a given partition to create <see cref="PartitionReceiver"/>.
+        /// Delegate is invoked by passing in PartitionId and then user can return <see cref="PartitionReceiver"/> for receiving messages.
+        /// This is only used when <see cref="Lease.Offset"/> is not provided and receiver is being created for the very first time.
         /// </summary>
-        public Func<string, object> InitialOffsetProvider { get; set; }
+        public Func<string, EventPosition> InitialOffsetProvider { get; set; }
 
         /// <summary>
         /// Returns whether the EventProcessorHost will call IEventProcessor.OnEvents(null) when a receive

src/Microsoft.Azure.EventHubs.Processor/Microsoft.Azure.EventHubs.Processor.csproj

@@ -29,7 +29,6 @@
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
-    <DefineConstants>$(DefineConstants);UAP10_0</DefineConstants>
     <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
     <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
     <TargetPlatformVersion>10.0.14393.0</TargetPlatformVersion>

src/Microsoft.Azure.EventHubs.Processor/PartitionContext.cs

@@ -4,6 +4,7 @@
 namespace Microsoft.Azure.EventHubs.Processor
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
 
     /// <summary>
@@ -13,18 +14,24 @@ public class PartitionContext
     {
         readonly EventProcessorHost host;
 
-        internal PartitionContext(EventProcessorHost host, string partitionId, string eventHubPath, string consumerGroupName)
+        internal PartitionContext(EventProcessorHost host, string partitionId, string eventHubPath, string consumerGroupName, CancellationToken cancellationToken)
         {
             this.host = host;
             this.PartitionId = partitionId;
             this.EventHubPath = eventHubPath;
             this.ConsumerGroupName = consumerGroupName;
+            this.CancellationToken = cancellationToken;
             this.ThisLock = new object();
-            this.Offset = PartitionReceiver.StartOfStream;
+            this.Offset = EventPosition.FromStart().Offset;
             this.SequenceNumber = 0;
             this.RuntimeInformation = new ReceiverRuntimeInformation(partitionId);
         }
 
+        /// <summary>
+        /// Gets triggered when the partition gets closed.
+        /// </summary>
+        public CancellationToken CancellationToken { get; }
+
         /// <summary>
         /// Gets the name of the consumer group.
         /// </summary>
@@ -84,43 +91,28 @@ internal void SetOffsetAndSequenceNumber(EventData eventData)
             }
         }
 
-        internal async Task<object> GetInitialOffsetAsync() // throws InterruptedException, ExecutionException
+        internal async Task<EventPosition> GetInitialOffsetAsync() // throws InterruptedException, ExecutionException
         {
             Checkpoint startingCheckpoint = await this.host.CheckpointManager.GetCheckpointAsync(this.PartitionId).ConfigureAwait(false);
-            object startAt;
+            EventPosition eventPosition;
 
             if (startingCheckpoint == null)
             {
                 // No checkpoint was ever stored. Use the initialOffsetProvider instead.
-                Func<string, object> initialOffsetProvider = this.host.EventProcessorOptions.InitialOffsetProvider;
+                Func<string, EventPosition> initialOffsetProvider = this.host.EventProcessorOptions.InitialOffsetProvider;
                 ProcessorEventSource.Log.PartitionPumpInfo(this.host.Id, this.PartitionId, "Calling user-provided initial offset provider");
-                startAt = initialOffsetProvider(this.PartitionId);
-
-                if (startAt is string)
-                {
-                    this.Offset = (string)startAt;
-                    this.SequenceNumber = 0; // TODO we use sequenceNumber to check for regression of offset, 0 could be a problem until it gets updated from an event
-                    ProcessorEventSource.Log.PartitionPumpInfo(this.host.Id, this.PartitionId, $"Initial offset/sequenceNumber provided: {this.Offset}/{this.SequenceNumber}");
-                }
-                else if (startAt is DateTime)
-                {
-                    // can't set offset/sequenceNumber
-                    ProcessorEventSource.Log.PartitionPumpInfo(this.host.Id, this.PartitionId, $"Initial timestamp provided: {(DateTime)startAt}");
-                }
-                else
-                {
-                    throw new ArgumentException("Unexpected object type returned by user-provided initialOffsetProvider");
-                }
+                eventPosition = initialOffsetProvider(this.PartitionId);
+                ProcessorEventSource.Log.PartitionPumpInfo(this.host.Id, this.PartitionId, $"Initial Position Provider. Offset:{eventPosition.Offset}, SequenceNumber:{eventPosition.SequenceNumber}, DateTime:{eventPosition.EnqueuedTimeUtc}");
             }
             else
             {
                 this.Offset = startingCheckpoint.Offset;
                 this.SequenceNumber = startingCheckpoint.SequenceNumber;
                 ProcessorEventSource.Log.PartitionPumpInfo(this.host.Id, this.PartitionId, $"Retrieved starting offset/sequenceNumber: {this.Offset}/{this.SequenceNumber}");
-                startAt = this.Offset;
+                eventPosition = EventPosition.FromOffset(this.Offset);
             }
 
-            return startAt;
+            return eventPosition;
         }
 
         /// <summary>
@@ -154,7 +146,7 @@ public Task CheckpointAsync(EventData eventData)
             {
                 throw new ArgumentNullException("eventData");
             }
-            
+
             // We have never seen this sequence number yet
             if (eventData.SystemProperties.SequenceNumber > this.SequenceNumber)
             {

src/Microsoft.Azure.EventHubs.Processor/PartitionPump.cs

@@ -6,10 +6,13 @@ namespace Microsoft.Azure.EventHubs.Processor
     using System;
     using System.Collections.Generic;
     using System.Linq;
+    using System.Threading;
     using System.Threading.Tasks;
 
     abstract class PartitionPump
-    {   
+    {
+        CancellationTokenSource cancellationTokenSource;
+
         protected PartitionPump(EventProcessorHost host, Lease lease)
         {
             this.Host = host;
@@ -37,7 +40,9 @@ public async Task OpenAsync()
         {
             this.PumpStatus = PartitionPumpStatus.Opening;
 
-            this.PartitionContext = new PartitionContext(this.Host, this.Lease.PartitionId, this.Host.EventHubPath, this.Host.ConsumerGroupName);
+            this.cancellationTokenSource = new CancellationTokenSource();
+
+            this.PartitionContext = new PartitionContext(this.Host, this.Lease.PartitionId, this.Host.EventHubPath, this.Host.ConsumerGroupName, this.cancellationTokenSource.Token);
             this.PartitionContext.Lease = this.Lease;
 
             if (this.PumpStatus == PartitionPumpStatus.Opening)
@@ -86,6 +91,8 @@ public async Task CloseAsync(CloseReason reason)
             this.PumpStatus = PartitionPumpStatus.Closing;
             try
             {
+                this.cancellationTokenSource.Cancel();
+
                 await this.OnClosingAsync(reason).ConfigureAwait(false);
 
                 if (this.Processor != null)
@@ -111,8 +118,8 @@ public async Task CloseAsync(CloseReason reason)
 
             if (reason != CloseReason.LeaseLost)
             {
-                // Since this pump is dead, release the lease. 
-                // Ignore LeaseLostException 
+                // Since this pump is dead, release the lease.
+                // Ignore LeaseLostException
                 try
                 {
                     await this.Host.LeaseManager.ReleaseLeaseAsync(this.PartitionContext.Lease).ConfigureAwait(false);
@@ -130,15 +137,7 @@ protected async Task ProcessEventsAsync(IEnumerable<EventData> events)
         {
             if (events == null)
             {
-                if (this.Host.EventProcessorOptions.InvokeProcessorAfterReceiveTimeout)
-                {
-                    // Assumes that .NET Core client will call with empty EventData on receive timeout.
-                    events = Enumerable.Empty<EventData>();
-                }
-                else
-                {
-                    return;
-                }
+                events = Enumerable.Empty<EventData>();
             }
 
             // Synchronize to serialize calls to the processor.

src/Microsoft.Azure.EventHubs/Amqp/AmqpClientConstants.cs

@@ -15,6 +15,7 @@ class AmqpClientConstants
         public static readonly AmqpSymbol EntityTypeName = AmqpConstants.Vendor + ":entity-type";
         public static readonly AmqpSymbol TimeoutName = AmqpConstants.Vendor + ":timeout";
         public static readonly AmqpSymbol EnableReceiverRuntimeMetricName = AmqpConstants.Vendor + ":enable-receiver-runtime-metric";
+        public static readonly AmqpSymbol ReceiverIdentifierName = AmqpConstants.Vendor + ":receiver-name";
 
         // Error codes
         public static readonly AmqpSymbol DeadLetterName = AmqpConstants.Vendor + ":dead-letter";
@@ -42,6 +43,7 @@ class AmqpClientConstants
         public static readonly AmqpSymbol MessageReceiptsFilterName = AmqpConstants.Vendor + ":message-receipts-filter";
         public static readonly AmqpSymbol ClientSideCursorFilterName = AmqpConstants.Vendor + ":client-side-filter";
         public static readonly TimeSpan ClientMinimumTokenRefreshInterval = TimeSpan.FromMinutes(4);
+        public const string FilterSeqNumberName = "amqp.annotation." + ClientConstants.SequenceNumberName;
         public const string FilterOffsetPartName = "amqp.annotation.x-opt-offset";
         public const string FilterOffset = FilterOffsetPartName + " > ";
         public const string FilterInclusiveOffset = FilterOffsetPartName + " >= ";

src/Microsoft.Azure.EventHubs/Amqp/AmqpEventHubClient.cs

@@ -5,7 +5,6 @@ namespace Microsoft.Azure.EventHubs.Amqp
 {
     using System;
     using System.Linq;
-    using System.Net;
     using System.Threading.Tasks;
     using Microsoft.Azure.Amqp.Sasl;
     using Microsoft.Azure.Amqp;
@@ -26,17 +25,33 @@ public AmqpEventHubClient(EventHubsConnectionStringBuilder csb)
 
             if (!string.IsNullOrWhiteSpace(csb.SharedAccessSignature))
             {
-                this.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(csb.SharedAccessSignature);
+                this.InternalTokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(csb.SharedAccessSignature);
             }
             else
             {
-                this.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(csb.SasKeyName, csb.SasKey);
+                this.InternalTokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider(csb.SasKeyName, csb.SasKey);
             }
 
             this.CbsTokenProvider = new TokenProviderAdapter(this);
             this.ConnectionManager = new FaultTolerantAmqpObject<AmqpConnection>(this.CreateConnectionAsync, this.CloseConnection);
         }
 
+        public AmqpEventHubClient(
+            Uri endpointAddress, 
+            string entityPath, 
+            ITokenProvider tokenProvider, 
+            TimeSpan operationTimeout, 
+            TransportType transportType)
+            : base(new EventHubsConnectionStringBuilder(endpointAddress, entityPath, operationTimeout, transportType))
+        {
+            this.ContainerId = Guid.NewGuid().ToString("N");
+            this.AmqpVersion = new Version(1, 0, 0, 0);
+            this.MaxFrameSize = AmqpConstants.DefaultMaxFrameSize;
+            this.InternalTokenProvider = tokenProvider;
+            this.CbsTokenProvider = new TokenProviderAdapter(this);
+            this.ConnectionManager = new FaultTolerantAmqpObject<AmqpConnection>(this.CreateConnectionAsync, this.CloseConnection);
+        }
+
         internal ICbsTokenProvider CbsTokenProvider { get; }
 
         internal FaultTolerantAmqpObject<AmqpConnection> ConnectionManager { get; }
@@ -47,18 +62,18 @@ public AmqpEventHubClient(EventHubsConnectionStringBuilder csb)
 
         uint MaxFrameSize { get; }
 
-        internal TokenProvider TokenProvider { get; }
+        internal ITokenProvider InternalTokenProvider { get; }
 
         internal override EventDataSender OnCreateEventSender(string partitionId)
         {
             return new AmqpEventDataSender(this, partitionId);
         }
 
         protected override PartitionReceiver OnCreateReceiver(
-            string consumerGroupName, string partitionId, string startOffset, bool offsetInclusive, DateTime? startTime, long? epoch, ReceiverOptions receiverOptions)
+            string consumerGroupName, string partitionId, EventPosition eventPosition, long? epoch, ReceiverOptions receiverOptions)
         {
             return new AmqpPartitionReceiver(
-                this, consumerGroupName, partitionId, startOffset, offsetInclusive, startTime, epoch, receiverOptions);
+                this, consumerGroupName, partitionId, eventPosition, epoch, receiverOptions);
         }
 
         protected override Task OnCloseAsync()
@@ -110,7 +125,7 @@ internal static AmqpSettings CreateAmqpSettings(
             string sslHostName = null,
             bool useWebSockets = false,
             bool sslStreamUpgrade = false,
-            NetworkCredential networkCredential = null,
+            System.Net.NetworkCredential networkCredential = null,
             bool forceTokenProvider = true)
         {
             var settings = new AmqpSettings();
@@ -266,11 +281,9 @@ public TokenProviderAdapter(AmqpEventHubClient eventHubClient)
 
             public async Task<CbsToken> GetTokenAsync(Uri namespaceAddress, string appliesTo, string[] requiredClaims)
             {
-                string claim = requiredClaims?.FirstOrDefault();
-                var tokenProvider = this.eventHubClient.TokenProvider;
                 var timeout = this.eventHubClient.ConnectionStringBuilder.OperationTimeout;
-                var token = await tokenProvider.GetTokenAsync(appliesTo, claim, timeout).ConfigureAwait(false);
-                return new CbsToken(token.TokenValue, CbsConstants.ServiceBusSasTokenType, token.ExpiresAtUtc);
+                var token = await this.eventHubClient.InternalTokenProvider.GetTokenAsync(appliesTo, timeout).ConfigureAwait(false);
+                return new CbsToken(token.TokenValue, token.TokenType, token.ExpiresAtUtc);
             }
         }
     }

src/Microsoft.Azure.EventHubs/Amqp/AmqpExceptionHelper.cs

@@ -115,6 +115,10 @@ public static Exception ToMessagingContract(string condition, string message, bo
             {
                 return new ReceiverDisconnectedException(message);
             }
+            else if (string.Equals(condition, AmqpErrorCode.ResourceLimitExceeded.Value))
+            {
+                return new QuotaExceededException(message);
+            }
             else
             {
                 return new EventHubsException(true, message);