diff --git a/src/DotPulsar/Abstractions/IConsumerOfT.cs b/src/DotPulsar/Abstractions/IConsumerOfT.cs index 613a15410..593acc3b1 100644 --- a/src/DotPulsar/Abstractions/IConsumerOfT.cs +++ b/src/DotPulsar/Abstractions/IConsumerOfT.cs @@ -17,4 +17,4 @@ namespace DotPulsar.Abstractions; /// /// A generic consumer abstraction. /// -public interface IConsumer : IConsumer, IReceive> { } +public interface IConsumer : IConsumer, IReceive>, IPeek> { } diff --git a/src/DotPulsar/Abstractions/IPeek.cs b/src/DotPulsar/Abstractions/IPeek.cs new file mode 100644 index 000000000..834d9a747 --- /dev/null +++ b/src/DotPulsar/Abstractions/IPeek.cs @@ -0,0 +1,26 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar.Abstractions; + +/// +/// An abstraction for receiving a single message. +/// +public interface IPeek +{ + /// + /// Peek a single message. + /// + ValueTask Peek(CancellationToken cancellationToken = default); +} diff --git a/src/DotPulsar/Extensions/PeekExtensions.cs b/src/DotPulsar/Extensions/PeekExtensions.cs new file mode 100644 index 000000000..ce3481dbf --- /dev/null +++ b/src/DotPulsar/Extensions/PeekExtensions.cs @@ -0,0 +1,54 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar.Extensions; + +using DotPulsar.Abstractions; +using Microsoft.Extensions.ObjectPool; +using System.Runtime.CompilerServices; + +/// +/// Extensions for IReceive. +/// +public static class PeekExtensions +{ + static PeekExtensions() + { + var policy = new DefaultPooledObjectPolicy(); + _ctsPool = new DefaultObjectPool(policy); + } + + private static readonly ObjectPool _ctsPool; + + /// + /// Will return true (and a message) if a message is buffered or false otherwise. + /// + public static bool TryPeek(this IPeek peeker, out TMessage? message) + { + var cts = _ctsPool.Get(); + var messageTask = peeker.Peek(cts.Token); + + if (!messageTask.IsCompleted) + cts.Cancel(); + else + _ctsPool.Return(cts); + + if (messageTask.IsCompletedSuccessfully) + message = messageTask.Result; + else + message = default; + + return messageTask.IsCompletedSuccessfully; + } +} diff --git a/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs b/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs index 139d5ff3e..dc6b8a4a9 100644 --- a/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs +++ b/src/DotPulsar/Internal/Abstractions/IConsumerChannel.cs @@ -25,5 +25,6 @@ public interface IConsumerChannel : IAsyncDisposable Task Send(CommandSeek command, CancellationToken cancellationToken); Task Send(CommandGetLastMessageId command, CancellationToken cancellationToken); ValueTask> Receive(CancellationToken cancellationToken); + ValueTask> Peek(CancellationToken cancellationToken); ValueTask ClosedByClient(CancellationToken cancellationToken); } diff --git a/src/DotPulsar/Internal/Abstractions/IPeek.cs b/src/DotPulsar/Internal/Abstractions/IPeek.cs new file mode 100644 index 000000000..358a5a2ba --- /dev/null +++ b/src/DotPulsar/Internal/Abstractions/IPeek.cs @@ -0,0 +1,20 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace DotPulsar.Internal.Abstractions; + +public interface IPeek +{ + bool TryPeek(out T? result); +} diff --git a/src/DotPulsar/Internal/AsyncQueue.cs b/src/DotPulsar/Internal/AsyncQueue.cs index 067554200..d7625b6fc 100644 --- a/src/DotPulsar/Internal/AsyncQueue.cs +++ b/src/DotPulsar/Internal/AsyncQueue.cs @@ -17,7 +17,7 @@ namespace DotPulsar.Internal; using DotPulsar.Internal.Abstractions; using DotPulsar.Internal.Exceptions; -public sealed class AsyncQueue : IEnqueue, IDequeue, IDisposable +public sealed class AsyncQueue : IEnqueue, IDequeue, IPeek, IDisposable { private readonly object _lock; private readonly Queue _queue; @@ -70,6 +70,32 @@ public ValueTask Dequeue(CancellationToken cancellationToken = default) return new ValueTask(node.Value.Task); } + public bool TryPeek(out T? result) + { + lock (_lock) + { + ThrowIfDisposed(); + + if (_queue.Count > 0) + { + try + { + // .NetStandard 2.0 does not have Queue.TryPeek() + result = _queue.Peek(); + return true; + } + catch + { + result = default; + return false; + } + } + } + + result = default; + return false; + } + public void Dispose() { if (Interlocked.Exchange(ref _isDisposed, 1) != 0) diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs index dc4588316..d7f725388 100644 --- a/src/DotPulsar/Internal/Consumer.cs +++ b/src/DotPulsar/Internal/Consumer.cs @@ -22,7 +22,8 @@ namespace DotPulsar.Internal; public sealed class Consumer : IConsumer { - private readonly TaskCompletionSource> _emptyTaskCompletionSource; + private readonly TaskCompletionSource> _emptyReceiveTaskCompletionSource; + private readonly TaskCompletionSource> _emptyPeekTaskCompletionSource; private readonly IConnectionPool _connectionPool; private readonly ProcessManager _processManager; private readonly StateManager _state; @@ -38,6 +39,7 @@ public sealed class Consumer : IConsumer private bool _isPartitionedTopic; private int _numberOfPartitions; private Task>[] _receiveTasks; + private Task>[] _peekTasks; private int _subConsumerIndex; private Exception? _faultException; @@ -60,6 +62,7 @@ public Consumer( SubscriptionType = consumerOptions.SubscriptionType; Topic = consumerOptions.Topic; _receiveTasks = Array.Empty>>(); + _peekTasks = Array.Empty>>(); _cts = new CancellationTokenSource(); _exceptionHandler = exceptionHandler; _semaphoreSlim = new SemaphoreSlim(1); @@ -73,7 +76,8 @@ public Consumer( _isDisposed = 0; _subConsumers = Array.Empty>(); - _emptyTaskCompletionSource = new TaskCompletionSource>(); + _emptyReceiveTaskCompletionSource = new TaskCompletionSource>(); + _emptyPeekTaskCompletionSource = new TaskCompletionSource>(); _ = Setup(); } @@ -102,6 +106,7 @@ private async Task Monitor() _isPartitionedTopic = _numberOfPartitions != 0; var numberOfSubConsumers = _isPartitionedTopic ? _numberOfPartitions : 1; _receiveTasks = new Task>[numberOfSubConsumers]; + _peekTasks = new Task>[numberOfSubConsumers]; _subConsumers = new SubConsumer[numberOfSubConsumers]; var monitoringTasks = new Task[numberOfSubConsumers]; var states = new ConsumerState[numberOfSubConsumers]; @@ -109,7 +114,8 @@ private async Task Monitor() for (var i = 0; i < numberOfSubConsumers; i++) { - _receiveTasks[i] = _emptyTaskCompletionSource.Task; + _receiveTasks[i] = _emptyReceiveTaskCompletionSource.Task; + _peekTasks[i] = _emptyPeekTaskCompletionSource.Task; var topicName = _isPartitionedTopic ? GetPartitionedTopicName(i) : Topic; _subConsumers[i] = CreateSubConsumer(topicName); monitoringTasks[i] = _subConsumers[i].State.OnStateChangeFrom(ConsumerState.Disconnected, _cts.Token).AsTask(); @@ -189,7 +195,7 @@ public async ValueTask> Receive(CancellationToken cancellatio _subConsumerIndex = 0; var receiveTask = _receiveTasks[_subConsumerIndex]; - if (receiveTask == _emptyTaskCompletionSource.Task) + if (receiveTask == _emptyReceiveTaskCompletionSource.Task) { var receiveTaskValueTask = _subConsumers[_subConsumerIndex].Receive(cancellationToken); if (receiveTaskValueTask.IsCompleted) @@ -200,7 +206,7 @@ public async ValueTask> Receive(CancellationToken cancellatio { if (receiveTask.IsCompleted) { - _receiveTasks[_subConsumerIndex] = _emptyTaskCompletionSource.Task; + _receiveTasks[_subConsumerIndex] = _emptyReceiveTaskCompletionSource.Task; return receiveTask.Result; } } @@ -209,6 +215,45 @@ public async ValueTask> Receive(CancellationToken cancellatio } } + public async ValueTask> Peek(CancellationToken cancellationToken = default) + { + await Guard(cancellationToken).ConfigureAwait(false); + + if (!_isPartitionedTopic) + return await _subConsumers[_subConsumerIndex].Peek(cancellationToken).ConfigureAwait(false); + + var iterations = 0; + using (await _lock.Lock(cancellationToken).ConfigureAwait(false)) + { + while (true) + { + iterations++; + _subConsumerIndex++; + if (_subConsumerIndex == _subConsumers.Length) + _subConsumerIndex = 0; + + var peekTask = _peekTasks[_subConsumerIndex]; + if (peekTask == _emptyPeekTaskCompletionSource.Task) + { + var peekTaskValueTask = _subConsumers[_subConsumerIndex].Peek(cancellationToken); + if (peekTaskValueTask.IsCompleted) + return peekTaskValueTask.Result; + _peekTasks[_subConsumerIndex] = peekTaskValueTask.AsTask(); + } + else + { + if (peekTask.IsCompleted) + { + _peekTasks[_subConsumerIndex] = _emptyPeekTaskCompletionSource.Task; + return peekTask.Result; + } + } + if (iterations == _subConsumers.Length) + await Task.WhenAny(_peekTasks).ConfigureAwait(false); + } + } + } + public async ValueTask Acknowledge(MessageId messageId, CancellationToken cancellationToken) { await Guard(cancellationToken).ConfigureAwait(false); diff --git a/src/DotPulsar/Internal/ConsumerChannel.cs b/src/DotPulsar/Internal/ConsumerChannel.cs index ddc573337..905af66e8 100644 --- a/src/DotPulsar/Internal/ConsumerChannel.cs +++ b/src/DotPulsar/Internal/ConsumerChannel.cs @@ -88,54 +88,71 @@ public async ValueTask> Receive(CancellationToken cancellatio var messagePackage = await _queue.Dequeue(cancellationToken).ConfigureAwait(false); - if (!messagePackage.ValidateMagicNumberAndChecksum()) + try { - await RejectPackage(messagePackage, CommandAck.ValidationErrorType.ChecksumMismatch, cancellationToken).ConfigureAwait(false); - continue; + return CreateFromMessagePackage(messagePackage); } - - var metadataSize = messagePackage.GetMetadataSize(); - var metadata = messagePackage.ExtractMetadata(metadataSize); - var data = messagePackage.ExtractData(metadataSize); - - if (metadata.Compression != CompressionType.None) + catch { - var decompressor = _decompressors[(int) metadata.Compression]; - if (decompressor is null) - throw new CompressionException($"Support for {metadata.Compression} compression was not found"); - - try - { - data = decompressor.Decompress(data, (int) metadata.UncompressedSize); - } - catch - { - await RejectPackage(messagePackage, CommandAck.ValidationErrorType.DecompressionError, cancellationToken).ConfigureAwait(false); - continue; - } + await RejectPackage(messagePackage, CommandAck.ValidationErrorType.DecompressionError, cancellationToken).ConfigureAwait(false); } + } + } + } - var messageId = messagePackage.MessageId; - var redeliveryCount = messagePackage.RedeliveryCount; + public async ValueTask> Peek(CancellationToken cancellationToken = default) + { + if (_sendWhenZero == 0) + await SendFlow(cancellationToken).ConfigureAwait(false); - if (metadata.ShouldSerializeNumMessagesInBatch()) + using (await _lock.Lock(cancellationToken).ConfigureAwait(false)) + { + while (true) + { + if(!_queue.TryPeek(out var messagePackage)) + continue; + + try { - try - { - return _batchHandler.Add(messageId, redeliveryCount, metadata, data); - } - catch - { - await RejectPackage(messagePackage, CommandAck.ValidationErrorType.BatchDeSerializeError, cancellationToken).ConfigureAwait(false); - continue; - } + return CreateFromMessagePackage(messagePackage); + } + catch + { + await RejectPackage(messagePackage, CommandAck.ValidationErrorType.DecompressionError, cancellationToken).ConfigureAwait(false); } - - return _messageFactory.Create(messageId.ToMessageId(_topic), redeliveryCount, data, metadata); } } } + private IMessage CreateFromMessagePackage(MessagePackage messagePackage) + { + if (!messagePackage.ValidateMagicNumberAndChecksum()) + throw new ChecksumException("Checksum validation and magic number validation failed"); + + var metadataSize = messagePackage.GetMetadataSize(); + var metadata = messagePackage.ExtractMetadata(metadataSize); + var data = messagePackage.ExtractData(metadataSize); + + if (metadata.Compression != CompressionType.None) + { + var decompressor = _decompressors[(int) metadata.Compression]; + if (decompressor is null) + throw new CompressionException($"Support for {metadata.Compression} compression was not found"); + + data = decompressor.Decompress(data, (int) metadata.UncompressedSize); + } + + var messageId = messagePackage.MessageId; + var redeliveryCount = messagePackage.RedeliveryCount; + + if (metadata.ShouldSerializeNumMessagesInBatch()) + { + return _batchHandler.Add(messageId, redeliveryCount, metadata, data); + } + + return _messageFactory.Create(messageId.ToMessageId(_topic), redeliveryCount, data, metadata); + } + public async Task Send(CommandAck command, CancellationToken cancellationToken) { var messageId = command.MessageIds[0]; diff --git a/src/DotPulsar/Internal/NotReadyChannel.cs b/src/DotPulsar/Internal/NotReadyChannel.cs index a9e64a375..166e6927e 100644 --- a/src/DotPulsar/Internal/NotReadyChannel.cs +++ b/src/DotPulsar/Internal/NotReadyChannel.cs @@ -25,6 +25,9 @@ public sealed class NotReadyChannel : IConsumerChannel, IPro public ValueTask DisposeAsync() => new(); + public ValueTask?> Peek(CancellationToken cancellationToken) + => throw GetException(); + public ValueTask ClosedByClient(CancellationToken cancellationToken) => new(); diff --git a/src/DotPulsar/Internal/SubConsumer.cs b/src/DotPulsar/Internal/SubConsumer.cs index 6be040d05..19a6bcd03 100644 --- a/src/DotPulsar/Internal/SubConsumer.cs +++ b/src/DotPulsar/Internal/SubConsumer.cs @@ -86,6 +86,9 @@ private async ValueTask DisposeChannel() public async ValueTask> Receive(CancellationToken cancellationToken) => await _executor.Execute(() => InternalReceive(cancellationToken), cancellationToken).ConfigureAwait(false); + public async ValueTask> Peek(CancellationToken cancellationToken = default) + => await _executor.Execute(() => InternalPeek(cancellationToken), cancellationToken).ConfigureAwait(false); + public async ValueTask Acknowledge(MessageId messageId, CancellationToken cancellationToken) => await InternalAcknowledge(messageId, CommandAck.AckType.Individual, cancellationToken).ConfigureAwait(false); @@ -188,6 +191,12 @@ private async ValueTask> InternalReceive(CancellationToken ca return await _channel.Receive(cancellationToken).ConfigureAwait(false); } + private async ValueTask> InternalPeek(CancellationToken cancellationToken) + { + Guard(); + return await _channel.Peek(cancellationToken).ConfigureAwait(false); + } + private async ValueTask InternalUnsubscribe(CommandUnsubscribe command, CancellationToken cancellationToken) { Guard();