| | | 1 | | // Copyright (c) Microsoft Corporation. All rights reserved. |
| | | 2 | | // Licensed under the MIT License. |
| | | 3 | | |
| | | 4 | | using System; |
| | | 5 | | using Azure.Core; |
| | | 6 | | using Azure.Messaging.EventHubs.Consumer; |
| | | 7 | | using Microsoft.Azure.Amqp; |
| | | 8 | | using Microsoft.Azure.Amqp.Framing; |
| | | 9 | | |
| | | 10 | | namespace Azure.Messaging.EventHubs |
| | | 11 | | { |
| | | 12 | | /// <summary> |
| | | 13 | | /// The set of filters associated with an AMQP messages and |
| | | 14 | | /// entities. |
| | | 15 | | /// </summary> |
| | | 16 | | /// |
| | | 17 | | internal static class AmqpFilter |
| | | 18 | | { |
| | | 19 | | /// <summary>Indicates filtering based on the sequence number of a message.</summary> |
| | | 20 | | public const string SequenceNumberName = "amqp.annotation.x-opt-sequence-number"; |
| | | 21 | | |
| | | 22 | | /// <summary>Indicates filtering based on the offset of a message.</summary> |
| | | 23 | | public const string OffsetName = "amqp.annotation.x-opt-offset"; |
| | | 24 | | |
| | | 25 | | /// <summary>Indicates filtering based on time that a message was enqueued.</summary> |
| | | 26 | | public const string EnqueuedTimeName = "amqp.annotation.x-opt-enqueued-time"; |
| | | 27 | | |
| | | 28 | | /// <summary>Identifies the filter type name.</summary> |
| | | 29 | | public const string ConsumerFilterName = AmqpConstants.Apache + ":selector-filter:string"; |
| | | 30 | | |
| | | 31 | | /// <summary>Identifies the filter type code.</summary> |
| | | 32 | | public const ulong ConsumerFilterCode = 0x00000137000000A; |
| | | 33 | | |
| | | 34 | | /// <summary> |
| | | 35 | | /// Creates an event consumer filter based on the specified expression. |
| | | 36 | | /// </summary> |
| | | 37 | | /// |
| | | 38 | | /// <param name="filterExpression">The SQL-like expression to use for filtering events in the partition.</param> |
| | | 39 | | /// |
| | | 40 | | /// <returns>An <see cref="AmqpDescribed"/> type to use in the filter map for a consumer AMQP link.</returns> |
| | | 41 | | /// |
| | | 42 | | public static AmqpDescribed CreateConsumerFilter(string filterExpression) |
| | | 43 | | { |
| | 22 | 44 | | Argument.AssertNotNullOrEmpty(filterExpression, nameof(filterExpression)); |
| | 18 | 45 | | return new AmqpDescribed(ConsumerFilterName, ConsumerFilterCode) { Value = filterExpression }; |
| | | 46 | | } |
| | | 47 | | |
| | | 48 | | /// <summary> |
| | | 49 | | /// Builds an AMQP filter expression for the specified event position. |
| | | 50 | | /// </summary> |
| | | 51 | | /// |
| | | 52 | | /// <param name="eventPosition">The event position to use as the source for filtering.</param> |
| | | 53 | | /// |
| | | 54 | | /// <returns>The AMQP filter expression that corresponds to the <paramref name="eventPosition"/>.</returns> |
| | | 55 | | /// |
| | | 56 | | public static string BuildFilterExpression(EventPosition eventPosition) |
| | | 57 | | { |
| | | 58 | | // Build the filter expression, in the order of significance. |
| | | 59 | | |
| | 40 | 60 | | if (!string.IsNullOrEmpty(eventPosition.Offset)) |
| | | 61 | | { |
| | 26 | 62 | | return $"{ OffsetName } { (eventPosition.IsInclusive ? ">=" : ">") } { eventPosition.Offset }"; |
| | | 63 | | } |
| | | 64 | | |
| | 14 | 65 | | if (eventPosition.SequenceNumber.HasValue) |
| | | 66 | | { |
| | 6 | 67 | | return $"{ SequenceNumberName } { (eventPosition.IsInclusive ? ">=" : ">") } { eventPosition.SequenceNum |
| | | 68 | | } |
| | | 69 | | |
| | 8 | 70 | | if (eventPosition.EnqueuedTime.HasValue) |
| | | 71 | | { |
| | 6 | 72 | | return $"{ EnqueuedTimeName } > { eventPosition.EnqueuedTime.Value.ToUnixTimeMilliseconds() }"; |
| | | 73 | | } |
| | | 74 | | |
| | | 75 | | // If no filter was built, than the event position is not valid for filtering. |
| | | 76 | | |
| | 2 | 77 | | throw new ArgumentException(Resources.InvalidEventPositionForFilter, nameof(eventPosition)); |
| | | 78 | | } |
| | | 79 | | } |
| | | 80 | | } |