Messaging improvements.

caching/Squidex.Caching.Tests/ReplicatedCacheTests.cs

@@ -16,12 +16,11 @@ namespace Squidex.Caching;
 public class ReplicatedCacheTests
 {
     private readonly IMessageBus pubSub = A.Fake<IMessageBus>();
-    private readonly ReplicatedCacheOptions options = new ReplicatedCacheOptions { Enable = true };
     private readonly ReplicatedCache sut;
 
     public ReplicatedCacheTests()
     {
-        sut = new ReplicatedCache(CreateMemoryCache(), pubSub, Options.Create(options));
+        sut = new ReplicatedCache(CreateMemoryCache(), pubSub);
     }
 
     [Fact]
@@ -51,22 +50,19 @@ await sut.AddAsync(
         AssertCache(sut, "Key2", 1, true);
 
         await sut.RemoveAsync(
-            new[]
-            {
+            [
                 "Key1",
                 "Key2"
-            });
+            ]);
 
         AssertCache(sut, "Key1", null, false);
         AssertCache(sut, "Key2", null, false);
     }
 
     [Fact]
-    public async Task Should_not_serve_from_cache_when_disabled()
+    public async Task Should_not_serve_from_cache_when_expiration_is_not_set()
     {
-        options.Enable = false;
-
-        await sut.AddAsync("Key", 1, TimeSpan.FromMilliseconds(100));
+        await sut.AddAsync("Key", 1, TimeSpan.Zero);
 
         AssertCache(sut, "Key", null, false);
     }
@@ -93,25 +89,12 @@ public async Task Should_not_invalidate_other_instances_when_added()
     [Fact]
     public async Task Should_send_invalidation_message_when_removed()
     {
-        options.Enable = true;
-
         await sut.RemoveAsync("Key");
 
         A.CallTo(() => pubSub.PublishAsync(A<CacheInvalidateMessage>.That.Matches(x => x.Keys.Contains("Key")), null, A<CancellationToken>._))
             .MustHaveHappened();
     }
 
-    [Fact]
-    public async Task Should_not_send_invalidation_message_when_not_enabled()
-    {
-        options.Enable = false;
-
-        await sut.RemoveAsync("Key");
-
-        A.CallTo(() => pubSub.PublishAsync(A<object>._, null, A<CancellationToken>._))
-            .MustNotHaveHappened();
-    }
-
     [Fact]
     public async Task Should_invalidate_keys_when_message_received()
     {

caching/Squidex.Caching/CachingServiceExtensions.cs

@@ -27,16 +27,11 @@ public static IServiceCollection AddAsyncLocalCache(this IServiceCollection serv
         return services;
     }
 
-    public static IServiceCollection AddReplicatedCache(this IServiceCollection services, Action<ReplicatedCacheOptions>? configureOptions = null)
+    public static IServiceCollection AddReplicatedCache(this IServiceCollection services)
     {
-        services.ConfigureOptional(configureOptions!);
-        services.AddSingleton<ReplicatedCache>();
-
-        services.TryAddSingleton<IReplicatedCache>(
-            c => c.GetRequiredService<ReplicatedCache>());
-
-        services.TryAddSingleton<IMessageHandler<CacheInvalidateMessage>>(
-            c => c.GetRequiredService<ReplicatedCache>());
+        services.AddMemoryCache();
+        services.AddSingletonAs<ReplicatedCache>()
+            .As<IReplicatedCache>().As<IMessageHandler<CacheInvalidateMessage>>();
 
         return services;
     }

caching/Squidex.Caching/ReplicatedCache.cs

@@ -6,7 +6,7 @@
 // ==========================================================================
 
 using Microsoft.Extensions.Caching.Memory;
-using Microsoft.Extensions.Options;
+using Squidex.Caching;
 using Squidex.Messaging;
 
 namespace Squidex.Caching;
@@ -15,15 +15,13 @@ public sealed class ReplicatedCache : IReplicatedCache, IMessageHandler<CacheInv
 {
     private readonly IMemoryCache memoryCache;
     private readonly IMessageBus messageBus;
-    private readonly ReplicatedCacheOptions options;
 
     public Guid InstanceId { get; } = Guid.NewGuid();
 
-    public ReplicatedCache(IMemoryCache memoryCache, IMessageBus messageBus, IOptions<ReplicatedCacheOptions> options)
+    public ReplicatedCache(IMemoryCache memoryCache, IMessageBus messageBus)
     {
         this.memoryCache = memoryCache;
         this.messageBus = messageBus;
-        this.options = options.Value;
     }
 
     public Task HandleAsync(CacheInvalidateMessage message,
@@ -46,7 +44,7 @@ public Task HandleAsync(CacheInvalidateMessage message,
     public Task AddAsync(string key, object? value, TimeSpan expiration,
         CancellationToken ct = default)
     {
-        if (!options.Enable)
+        if (expiration <= TimeSpan.Zero)
         {
             return Task.CompletedTask;
         }
@@ -59,7 +57,7 @@ public Task AddAsync(string key, object? value, TimeSpan expiration,
     public Task AddAsync(IEnumerable<KeyValuePair<string, object?>> items, TimeSpan expiration,
         CancellationToken ct = default)
     {
-        if (!options.Enable)
+        if (expiration <= TimeSpan.Zero)
         {
             return Task.CompletedTask;
         }
@@ -87,11 +85,6 @@ public Task RemoveAsync(string key1, string key2,
     public async Task RemoveAsync(string[] keys,
         CancellationToken ct = default)
     {
-        if (!options.Enable)
-        {
-            return;
-        }
-
         foreach (var key in keys)
         {
             if (key != null)
@@ -105,13 +98,6 @@ public async Task RemoveAsync(string[] keys,
 
     public bool TryGetValue(string key, out object? value)
     {
-        if (!options.Enable)
-        {
-            value = null;
-
-            return false;
-        }
-
         return memoryCache.TryGetValue(key, out value);
     }
 

caching/Squidex.Caching/Squidex.Caching.csproj

@@ -17,6 +17,7 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
     <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.0" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />

messaging/Squidex.Messaging.Mongo/MessagingServiceExtensions.cs

@@ -8,6 +8,7 @@
 using Microsoft.Extensions.Configuration;
 using Squidex.Messaging;
 using Squidex.Messaging.Mongo;
+using Squidex.Messaging.Subscriptions;
 
 namespace Microsoft.Extensions.DependencyInjection;
 
@@ -24,13 +25,13 @@ public static IServiceCollection AddMongoTransport(this IServiceCollection servi
         return services;
     }
 
-    public static IServiceCollection AddMongoSubscriptions(this IServiceCollection services, IConfiguration config, Action<MongoSubscriptionStoreOptions>? configure = null,
+    public static IServiceCollection AddMongoMessagingData(this IServiceCollection services, IConfiguration config, Action<MongoMessagingDataOptions>? configure = null,
         string configPath = "messaging:mongoDb:subscriptions")
     {
         services.ConfigureAndValidate(config, configPath, configure);
 
-        services.AddSingletonAs<MongoSubscriptionStore>()
-            .As<IMessagingSubscriptionStore>();
+        services.AddSingletonAs<MongoMessagingDataStore>()
+            .As<IMessagingDataStore>();
 
         return services;
     }

messaging/Squidex.Messaging.Mongo/MongoSubscriptionStoreOptions.cs → messaging/Squidex.Messaging.Mongo/MongoMessagingDataOptions.cs

@@ -9,7 +9,7 @@
 
 namespace Squidex.Messaging.Mongo;
 
-public sealed class MongoSubscriptionStoreOptions : IValidatableOptions
+public sealed class MongoMessagingDataOptions : IValidatableOptions
 {
     public string CollectionName { get; set; } = "Subscriptions";
 

messaging/Squidex.Messaging.Mongo/MongoSubscriptionStore.cs → messaging/Squidex.Messaging.Mongo/MongoMessagingDataStore.cs

@@ -8,10 +8,11 @@
 using Microsoft.Extensions.Options;
 using MongoDB.Driver;
 using Squidex.Hosting;
+using Squidex.Messaging.Subscriptions;
 
 namespace Squidex.Messaging.Mongo;
 
-public sealed class MongoSubscriptionStore : IMessagingSubscriptionStore, IInitializable
+public sealed class MongoMessagingDataStore : IMessagingDataStore, IInitializable
 {
     private readonly IMongoCollection<Entity> collection;
 
@@ -32,7 +33,7 @@ private sealed class Entity
         public DateTime Expiration { get; set; }
     }
 
-    public MongoSubscriptionStore(IMongoDatabase database, IOptions<MongoSubscriptionStoreOptions> options)
+    public MongoMessagingDataStore(IMongoDatabase database, IOptions<MongoMessagingDataOptions> options)
     {
         collection = database.GetCollection<Entity>(options.Value.CollectionName);
     }
@@ -57,10 +58,10 @@ public Task InitializeAsync(
             }, ct);
     }
 
-    public async Task<IReadOnlyList<(string Key, SerializedObject Value, DateTime Expiration)>> GetSubscriptionsAsync(string group,
+    public async Task<IReadOnlyList<Entry>> GetEntriesAsync(string group,
         CancellationToken ct)
     {
-        var result = new List<(string Key, SerializedObject Value, DateTime Expiration)>();
+        var result = new List<Entry>();
 
         var cursor = await collection.Find(x => x.Group == group).ToCursorAsync(ct);
 
@@ -70,14 +71,14 @@ public Task InitializeAsync(
             {
                 var value = new SerializedObject(item.ValueData, item.ValueType, item.ValueFormat);
 
-                result.Add((item.Key, value, item.Expiration));
+                result.Add(new Entry(group, item.Key, value, item.Expiration));
             }
         }
 
         return result;
     }
 
-    public async Task SubscribeManyAsync(SubscribeRequest[] requests,
+    public async Task StoreManyAsync(Entry[] requests,
         CancellationToken ct)
     {
         List<WriteModel<Entity>>? updates = null;
@@ -105,20 +106,14 @@ public async Task SubscribeManyAsync(SubscribeRequest[] requests,
         }
     }
 
-    public Task UnsubscribeAsync(string group, string key,
+    public Task DeleteAsync(string group, string key,
         CancellationToken ct)
     {
         string id = GetId(group, key);
 
         return collection.DeleteOneAsync(x => x.Id == id, ct);
     }
 
-    public Task CleanupAsync(DateTime now,
-        CancellationToken ct)
-    {
-        return Task.CompletedTask;
-    }
-
     private static string GetId(string group, string key)
     {
         return $"{group}/{key}";

messaging/Squidex.Messaging.Mongo/MongoSubscription.cs

@@ -19,20 +19,24 @@ internal sealed class MongoSubscription : IAsyncDisposable, IMessageAck
     private readonly string? queueFilter;
     private readonly IMongoCollection<MongoMessage> collection;
     private readonly MongoTransportOptions options;
-    private readonly IClock clock;
-    private readonly ILogger log;
+    private readonly TimeProvider timeProvider;
     private readonly SimpleTimer timer;
+    private readonly ILogger log;
 
-    public MongoSubscription(MessageTransportCallback callback, IMongoCollection<MongoMessage> collection,
+    public MongoSubscription(
+        MessageTransportCallback callback,
+        IMongoCollection<MongoMessage> collection,
         string? collectionName,
         string? queueFilter,
-        MongoTransportOptions options, IClock clock, ILogger log)
+        MongoTransportOptions options,
+        TimeProvider timeProvider,
+        ILogger log)
     {
         this.queueFilter = queueFilter;
         this.collectionName = collectionName;
         this.collection = collection;
         this.options = options;
-        this.clock = clock;
+        this.timeProvider = timeProvider;
         this.log = log;
 
         timer = new SimpleTimer(async ct =>
@@ -60,7 +64,7 @@ private Task<bool> PollMessageAsync(MessageTransportCallback callback,
     private async Task<bool> PollNormalAsync(MessageTransportCallback callback,
         CancellationToken ct)
     {
-        var now = clock.UtcNow;
+        var now = timeProvider.GetUtcNow().UtcDateTime;
 
         // We can fetch an document in one go with this operation.
         var mongoMessage =
@@ -81,7 +85,7 @@ await collection.FindOneAndUpdateAsync(CreateFilter(now),
     private async Task<bool> PollPrefetchAsync(MessageTransportCallback callback,
         CancellationToken ct)
     {
-        var now = clock.UtcNow;
+        var now = timeProvider.GetUtcNow().UtcDateTime;
 
         // There is no way to limit the updates, therefore we have to query candidates first.
         var candidates =