-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Async event handlers Write events to channel and publish from a background task. This allows handlers to be async and also prevents event listeners from blocking the NatsConnection code. * Added WaitToReadAsync return value check * Added NatsEventArgs and renamed MessageDroppedError to NatsMessageDroppedEventArgs. Changed events to use EventArgs and adjusted internal subscribers. Renamed OnError to MessageDropped, as that better reflects what error is signaled and also is naming-wise more consistent with the other events. * Use inner while to read all queued messages before waiting again * Changed AsyncEventHandler to return ValueTask. Removed license blurp from AsyncEventHandler as it now different from the original delegate. Adjusted code to match new signature. Note that "default" for ValueTask is a completed task. * Fixed build * dotnet format --------- Co-authored-by: Ziya Suzen <[email protected]>
- Loading branch information
Showing
16 changed files
with
270 additions
and
69 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| namespace NATS.Client.Core; | ||
|
|
||
| /// <summary> | ||
| /// An asynchronous event handler. | ||
| /// </summary> | ||
| /// <param name="sender">The sender of the event.</param> | ||
| /// <param name="args">Event arguments.</param> | ||
| /// <returns>A value task whose completion signals handling is finished.</returns> | ||
| public delegate ValueTask AsyncEventHandler(object? sender, EventArgs args); | ||
|
|
||
| /// <summary> | ||
| /// An asynchronous event handler. | ||
| /// </summary> | ||
| /// <typeparam name="TEventArgs">The type of event arguments.</typeparam> | ||
| /// <param name="sender">The sender of the event.</param> | ||
| /// <param name="args">Event arguments.</param> | ||
| /// <returns>A value task whose completion signals handling is finished.</returns> | ||
| public delegate ValueTask AsyncEventHandler<in TEventArgs>(object? sender, TEventArgs args); |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| // origin: https://github.com/microsoft/vs-threading/blob/9065e6e4b5593e6ed6e3ff0a9159d4e2765430d6/src/Microsoft.VisualStudio.Threading/TplExtensions.cs | ||
| // license: MIT | ||
| // | ||
| // Microsoft.VisualStudio.Threading | ||
| // Copyright (c) Microsoft Corporation | ||
| // All rights reserved. | ||
| // | ||
| // Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| // of this software and associated documentation files (the "Software"), to deal | ||
| // in the Software without restriction, including without limitation the rights | ||
| // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| // copies of the Software, and to permit persons to whom the Software is | ||
| // furnished to do so, subject to the following conditions: | ||
| // | ||
| // The above copyright notice and this permission notice shall be included in all | ||
| // copies or substantial portions of the Software. | ||
|
|
||
| namespace NATS.Client.Core.Internal; | ||
|
|
||
| internal static class AsyncEventExtensions | ||
| { | ||
| /// <summary> | ||
| /// Invokes asynchronous event handlers, returning a task that completes when all event handlers have been invoked. | ||
| /// Each handler is fully executed (including continuations) before the next handler in the list is invoked. | ||
| /// </summary> | ||
| /// <param name="handlers">The event handlers. May be <see langword="null" />.</param> | ||
| /// <param name="sender">The event source.</param> | ||
| /// <param name="args">The event argument.</param> | ||
| /// <returns>The task that completes when all handlers have completed.</returns> | ||
| /// <exception cref="AggregateException">Thrown if any handlers fail. It contains a collection of all failures.</exception> | ||
| public static async Task InvokeAsync(this AsyncEventHandler? handlers, object? sender, EventArgs args) | ||
| { | ||
| if (handlers != null) | ||
| { | ||
| var individualHandlers = handlers.GetInvocationList(); | ||
| List<Exception>? exceptions = null; | ||
| foreach (var asyncHandler in individualHandlers) | ||
| { | ||
| var handler = (AsyncEventHandler)asyncHandler; | ||
| try | ||
| { | ||
| await handler(sender, args).ConfigureAwait(true); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| exceptions ??= new List<Exception>(2); | ||
| exceptions.Add(ex); | ||
| } | ||
| } | ||
|
|
||
| if (exceptions != null) | ||
| { | ||
| throw new AggregateException(exceptions); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Invokes asynchronous event handlers, returning a task that completes when all event handlers have been invoked. | ||
| /// Each handler is fully executed (including continuations) before the next handler in the list is invoked. | ||
| /// </summary> | ||
| /// <typeparam name="TEventArgs">The type of argument passed to each handler.</typeparam> | ||
| /// <param name="handlers">The event handlers. May be <see langword="null" />.</param> | ||
| /// <param name="sender">The event source.</param> | ||
| /// <param name="args">The event argument.</param> | ||
| /// <returns>The task that completes when all handlers have completed. The task is faulted if any handlers throw an exception.</returns> | ||
| /// <exception cref="AggregateException">Thrown if any handlers fail. It contains a collection of all failures.</exception> | ||
| public static async Task InvokeAsync<TEventArgs>(this AsyncEventHandler<TEventArgs>? handlers, object? sender, TEventArgs args) | ||
| { | ||
| if (handlers != null) | ||
| { | ||
| var individualHandlers = handlers.GetInvocationList(); | ||
| List<Exception>? exceptions = null; | ||
| foreach (var asyncHandler in individualHandlers) | ||
| { | ||
| var handler = (AsyncEventHandler<TEventArgs>)asyncHandler; | ||
| try | ||
| { | ||
| await handler(sender, args).ConfigureAwait(true); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| exceptions ??= new List<Exception>(2); | ||
| exceptions.Add(ex); | ||
| } | ||
| } | ||
|
|
||
| if (exceptions != null) | ||
| { | ||
| throw new AggregateException(exceptions); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| // ReSharper disable UnusedAutoPropertyAccessor.Global - properties are used by consumers outside of this library | ||
| namespace NATS.Client.Core; | ||
|
|
||
| public class NatsEventArgs : EventArgs | ||
| { | ||
| public NatsEventArgs(string message) => Message = message; | ||
|
|
||
| public string Message { get; } | ||
| } | ||
|
|
||
| public class NatsMessageDroppedEventArgs : NatsEventArgs | ||
| { | ||
| public NatsMessageDroppedEventArgs(NatsSubBase subscription, int pending, string subject, string? replyTo, NatsHeaders? headers, object? data) | ||
| : base($"Dropped message from {subject} with {pending} pending messages") | ||
| { | ||
| Subscription = subscription; | ||
| Pending = pending; | ||
| Subject = subject; | ||
| ReplyTo = replyTo; | ||
| Headers = headers; | ||
| Data = data; | ||
| } | ||
|
|
||
| public NatsSubBase Subscription { get; } | ||
|
|
||
| public int Pending { get; } | ||
|
|
||
| public string Subject { get; } | ||
|
|
||
| public string? ReplyTo { get; } | ||
|
|
||
| public NatsHeaders? Headers { get; } | ||
|
|
||
| public object? Data { get; } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.