Add functionality to browse queues #412
Conversation
| namespace ActiveMQ.Artemis.Client | ||
| { | ||
| public interface IBrowser : IEnumerator<Message>, IEnumerable<Message>, IAsyncDisposable | ||
| { |
There was a problem hiding this comment.
I'd rather expect this interface to expose a method that returns IAsyncEnumerable:
IAsyncEnumerable<Message> ReceiveAllAsync(CancellationToken cancellationToken);Than implement IEnumerable itself.
As a side note, you cannot implement IEnumerable without blocking.
A naive implementation might look as follows:
public async IAsyncEnumerable<Message> ReceiveAllAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
yield return await _consumer.ReceiveAsync(cancellationToken).ConfigureAwait(false);
// we probably need to ack message, so the browser won't block after we run out of the consumer credit
}
}But I'm not sure that reusing IConsumer to do all the heavy-lifting is the right call. I'd be more inclined to do sth similar to what was done in RequestReplyClient.
| @@ -25,11 +25,12 @@ public interface IConnection : IAsyncDisposable | |||
| /// </summary> | |||
| bool IsOpened { get; } | |||
| Task<ITopologyManager> CreateTopologyManagerAsync(CancellationToken cancellationToken = default); | |||
| Task<IConsumer> CreateConsumerAsync(ConsumerConfiguration configuration, CancellationToken cancellationToken = default); | |||
| Task<IConsumer> CreateConsumerAsync(ConsumerConfiguration configuration, CancellationToken cancellationToken = default, bool isBrowser = false); | |||
There was a problem hiding this comment.
IConsumer != IBrowser. They have different characteristics. I wouldn't pretend they represent the same thing and try to shoehorn them under a single interface umbrella.
| Task<IProducer> CreateProducerAsync(ProducerConfiguration configuration, CancellationToken cancellationToken = default); | ||
| Task<IAnonymousProducer> CreateAnonymousProducerAsync(AnonymousProducerConfiguration configuration, CancellationToken cancellationToken = default); | ||
| Task<IRequestReplyClient> CreateRequestReplyClientAsync(RequestReplyClientConfiguration configuration, CancellationToken cancellationToken = default); | ||
|
|
||
| Task<IBrowser> CreateBrowserAsync(ConsumerConfiguration configuration, CancellationToken cancellationToken = default); |
There was a problem hiding this comment.
We probably need a special BrowserConfiguration type.
A big chunk of properties from ConsumerConfiguration doesn't make sense in a browser scenario:
I think we don't need RoutingType as the queue should already exist, Credit also seems redundant as there is no api to acknowledge message, Shared, Durable.
I'm not sure about NoLocalFilter. It definitely works for message browser scenario, but I'm not sure if it makes sense to expose it as part of the public api.
| try | ||
| { | ||
| _current = null; | ||
| _current = _consumer.ReceiveAsync(token).Result; |
There was a problem hiding this comment.
We don't want blocking on async call.
Relevant for ticket #35