Working on message bus subscriptions options #278
Conversation
| @@ -564,14 +565,16 @@ public abstract class QueueTestBase : TestWithLoggingBase, IDisposable { | |||
| Assert.InRange(sw.Elapsed, TimeSpan.FromMilliseconds(50), TimeSpan.FromMilliseconds(5000)); | |||
|
|
|||
| _ = Task.Run(async () => { | |||
| await SystemClock.SleepAsync(500); | |||
| await Task.Delay(500); | |||
There was a problem hiding this comment.
SystemClock.SleepAsync(500) should be doing this behind the scenes
There was a problem hiding this comment.
Yes I know that, but in this case we never want the sleep to be virtual.
| @@ -553,6 +553,7 @@ public abstract class QueueTestBase : TestWithLoggingBase, IDisposable { | |||
| return; | |||
|
|
|||
| try { | |||
| Log.MinimumLevel = LogLevel.Trace; | |||
There was a problem hiding this comment.
probably should revert this and leave this up to the caller.
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace Foundatio.Messaging { | ||
| public interface IMessageSubscriber { | ||
| Task SubscribeAsync<T>(Func<T, CancellationToken, Task> handler, CancellationToken cancellationToken = default) where T : class; |
There was a problem hiding this comment.
Wondering if we should create extension method overloads for this that just set a token on options. Will users know there is an implicit conversion happening here for cancellation token?
There was a problem hiding this comment.
Maybe, I am worried about us having overload conflicts since there are already a bunch of overloads for this method. I pretty much did this just for compatibility.
src/Foundatio/Messaging/Message.cs
Outdated
| } | ||
|
|
||
| public ValueTask DisposeAsync() { | ||
| throw new NotImplementedException(); |
There was a problem hiding this comment.
Should the default interface implementation throw for dispose? I'd think this would do nothing.
There was a problem hiding this comment.
As you cant see, these haven't been implemented yet. This PR is not done.
| _subscribers.TryRemove(subscriber.Id, out _); | ||
| if (_subscribers.Count == 0) | ||
| RemoveTopicSubscriptionAsync().GetAwaiter().GetResult(); | ||
|
|
||
| return new ValueTask(); |
There was a problem hiding this comment.
Kinda would be nice if this was async.
| string MessageType { get; set; } | ||
| IDictionary<string, string> Properties { get; } |
There was a problem hiding this comment.
Would this work for kafka? I'm dealing with public Func<ConsumeResult<string, byte[]>, string> ResolveMessageType { get; set; } so I can look at specific headers and not have to deserialize every single header to process the message.
There was a problem hiding this comment.
Well it can't be kafka specific. It has access to properties which should include the headers from the kafka message.
concurrent dictionary per topic that has a channel of messages.
publish sends a message to the channel based on the topic for the message
by default the message type full name is the topic (1 topic per message type)
message publish topics can be overridden to publish multiple message types to a single topic
subscription specifies a topic which can have wildcards
in memory subscription will check all of the topics and add all of the matching topics
to the subscription object
for rabbit impl, on subscribe will ensure all exchanges are created for the message types that are being listened for
publish will ensure the topic the message is going to is created
queue groups can be specified on subscriptions and if multiple subscribers use the same queue group, they will round robin the messages
subscriptions have their own channel instances
read from topic channel, write to all matching subscriber channels which in turn call the message handlers with their instances
when writing to a subscriber channel and there is more than one matching subscriber, then all subscribers get a copy of the message
if multiple subscribers have the same queue group name then we round robin them
https://deniskyashif.com/2019/12/08/csharp-channels-part-1/
https://github.com/deniskyashif/trydotnet-channels