Subscription performance improvements

src/DataCore.Adapter/PatternMatchingExtensions.cs

@@ -9,6 +9,11 @@ namespace DataCore.Adapter {
     /// </summary>
     public static class PatternMatchingExtensions {
 
+        /// <summary>
+        /// The wildcard characters that can be used in a pattern.
+        /// </summary>
+        private static readonly char[] s_wildcardChars = new char[] { '*', '?' };
+
         /// <summary>
         /// Determines if the string matches the specified regular expression.
         /// </summary>
@@ -35,7 +40,7 @@ public static bool Like(this string? s, Regex expression) {
             }
 
             if (s == null) {
-                return expression.Match(string.Empty).Success;
+                return false;
             }
 
             return expression.Match(s).Success;
@@ -89,6 +94,14 @@ public static bool Like(this string? s, string pattern) {
                 throw new ArgumentNullException(nameof(pattern));
             }
 
+            if (s == null) {
+                return false;
+            }
+
+            if (pattern.IndexOfAny(s_wildcardChars) < 0) {
+                return pattern.Equals(s, StringComparison.OrdinalIgnoreCase);
+            }
+
             // Construct a regex that can be used to search the string using the specified pattern 
             // as its base.
             // 

src/DataCore.Adapter/SubscriptionChannel.cs

@@ -141,17 +141,20 @@ public SubscriptionChannel(
             // to _outChannel by a dedicated background task.
             _inChannel = channelCapacity <= 0
                 ? Channel.CreateUnbounded<(TValue, bool)>(new UnboundedChannelOptions() {
+                    AllowSynchronousContinuations = false,
                     SingleReader = true,
                     SingleWriter = false
                 })
                 : Channel.CreateBounded<(TValue, bool)>(new BoundedChannelOptions(channelCapacity) {
+                    AllowSynchronousContinuations = false,
                     SingleReader = true,
                     SingleWriter = false,
                     FullMode = BoundedChannelFullMode.DropWrite
                 });
 
             // This is the actual channel exposed via the Reader property.
             _outChannel = Channel.CreateUnbounded<TValue>(new UnboundedChannelOptions() {
+                AllowSynchronousContinuations = false,
                 SingleReader = true,
                 SingleWriter = false,
             });
@@ -247,7 +250,7 @@ public bool Publish(TValue value, bool immediate = false) {
         private async Task RunIngressLoop(CancellationToken cancellationToken) {
             try {
                 while (await _inChannel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) {
-                    while (_inChannel.Reader.TryRead(out var val)) {
+                    while (!cancellationToken.IsCancellationRequested && _inChannel.Reader.TryRead(out var val)) {
                         if (val.Immediate || PublishInterval <= TimeSpan.Zero) {
                             if (PublishInterval > TimeSpan.Zero) {
                                 // Cancel next publish if one is already pending.

src/DataCore.Adapter/SubscriptionManager.cs

@@ -404,8 +404,10 @@ public virtual async ValueTask<bool> ValueReceived(TValue message, CancellationT
 
             try {
                 using (var ctSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedTokenSource.Token)) {
-                    await _masterChannel.Writer.WaitToWriteAsync(ctSource.Token).ConfigureAwait(false);
-                    return _masterChannel.Writer.TryWrite((message, subscribers));
+                    if (await _masterChannel.Writer.WaitToWriteAsync(ctSource.Token).ConfigureAwait(false)) {
+                        return _masterChannel.Writer.TryWrite((message, subscribers));
+                    }
+                    return false;
                 }
             }
             catch (OperationCanceledException) {
@@ -585,7 +587,7 @@ public string? AdapterId {
         /// capacity, attempts to write additional values into the channel will fail. A value 
         /// less than one indicates no limit.
         /// </summary>
-        public int ChannelCapacity { get; set; } = 100;
+        public int ChannelCapacity { get; set; } = 10000;
 
     }