Stream request middleware is ready for testing

Sample/Contracts/MyMessageContracts.cs

@@ -1,6 +1,11 @@
+using Shiny.Mediator.Middleware;
+
 namespace Sample.Contracts;
 
-public record MyMessageRequest(string Arg, bool FireAndForgetEvents) : IRequest<MyMessageResponse>;
+public record MyMessageRequest(string Arg, bool FireAndForgetEvents) : IRequest<MyMessageResponse>, ICacheItem
+{
+    public string CacheKey { get; }
+};
 
 public record MyMessageResponse(string Response);
 

Sample/Handlers/MyStreamRequestMiddleware.cs

@@ -0,0 +1,14 @@
+namespace Sample.Handlers;
+
+public class MyStreamRequestMiddleware<TRequest, TResult> : IStreamRequestMiddleware<TRequest, TResult> where TRequest : IStreamRequest<TResult>
+{
+    public IAsyncEnumerator<TResult> Process(
+        TRequest request, 
+        StreamRequestDelegate<TResult> next, 
+        IStreamRequestHandler<TRequest, TResult> requestHandler,
+        CancellationToken cancellationToken
+    )
+    {
+        return next();
+    }
+}

Sample/Handlers/SingletonRequestHandler.cs

@@ -4,8 +4,10 @@ namespace Sample.Handlers;
 
 
 [RegisterHandler]
+[TimedLogging(3000)]
 public class SingletonRequestHandler(IMediator mediator, AppSqliteConnection data) : IRequestHandler<MyMessageRequest, MyMessageResponse>
 {
+    [Cache(Storage = StoreType.File, MaxAgeSeconds = 30, OnlyForOffline = true)]
     public async Task<MyMessageResponse> Handle(MyMessageRequest request, CancellationToken cancellationToken)
     {
         var e = new MyMessageEvent(

Sample/TriggerViewModel.cs

@@ -25,6 +25,8 @@ AppSqliteConnection data
                     this.FireAndForgetEvents
                 );
                 var result = await mediator.Request(request, this.cancelSource.Token);
+
+
                 await data.Log(
                     "TriggerViewModel-Response",
                     new MyMessageEvent(

src/Shiny.Mediator.Maui/Attributes.cs

@@ -35,9 +35,9 @@ public class MainThreadAttribute : Attribute {}
 
 
 [AttributeUsage(AttributeTargets.Class, Inherited = false)]
-public class TimedLoggingAttribute : Attribute
+public class TimedLoggingAttribute(double errorThresholdMillis) : Attribute
 {
-    public double ErrorThresholdMillis { get; set; } = 0;
+    public double ErrorThresholdMillis => errorThresholdMillis;
 }
 
 

src/Shiny.Mediator.Maui/Middleware/CacheRequestMiddleware.cs

@@ -30,9 +30,9 @@ public virtual Task<TResult> Process(
         TRequest request, 
         RequestHandlerDelegate<TResult> next
     )
-    => cacheManager.CacheOrGet<TResult>(
+    => cacheManager.CacheOrGet(
         cfg, 
-        request, 
+        request!, 
         async () => await next().ConfigureAwait(false)
     );
 }

src/Shiny.Mediator/IRequestHandler.cs

@@ -3,12 +3,18 @@ namespace Shiny.Mediator;
 
 public interface IRequestHandler { }
 
-public interface IRequestHandler<in TRequest> : IRequestHandler where TRequest : IRequest
+public interface IRequestHandler<TRequest> : IRequestHandler where TRequest : IRequest
 {
     Task Handle(TRequest request, CancellationToken cancellationToken);
 }
 
-public interface IRequestHandler<in TRequest, TResult> : IRequestHandler where TRequest : IRequest<TResult>
+public interface IRequestHandler<TRequest, TResult> : IRequestHandler where TRequest : IRequest<TResult>
 {
     Task<TResult> Handle(TRequest request, CancellationToken cancellationToken);
+}
+
+
+public interface IStreamRequestHandler<TRequest, TResult> : IRequestHandler where TRequest : IStreamRequest<TResult>
+{
+    IAsyncEnumerable<TResult> Handle(TRequest request, CancellationToken cancellationToken);
 }

src/Shiny.Mediator/IStreamRequestMiddleware.cs

@@ -1,11 +1,14 @@
 namespace Shiny.Mediator;
 
 
+public delegate IAsyncEnumerator<TResult> StreamRequestDelegate<TResult>(); 
 public interface IStreamRequestMiddleware<TRequest, TResult> where TRequest : IStreamRequest<TResult>
 {
     // TODO: I want to be able to pump the async enumerable from the middleware as well
     IAsyncEnumerator<TResult> Process(
         TRequest request, 
+        StreamRequestDelegate<TResult> next,
+        IStreamRequestHandler<TRequest, TResult> requestHandler,
         CancellationToken cancellationToken
     );
 }

src/Shiny.Mediator/MediatorExtensions.cs

@@ -3,6 +3,7 @@
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Shiny.Mediator.Impl;
 using Shiny.Mediator.Infrastructure;
+using Shiny.Mediator.Middleware;
 
 namespace Shiny.Mediator;
 
@@ -83,4 +84,11 @@ public static IServiceCollection AddScopedAsImplementedInterfaces<TImplementatio
 
         return services;
     }
+
+
+    public static ShinyConfigurator AddReplayStreamMiddleware(this ShinyConfigurator configurator)
+    {
+        configurator.AddOpenStreamMiddleware(typeof(ReplayStreamMiddleware<,>));
+        return configurator;
+    }
 }

src/Shiny.Mediator/Middleware/ReplayStreamMiddleware.cs

@@ -0,0 +1,55 @@
+using System.Text.Json;
+
+namespace Shiny.Mediator.Middleware;
+
+/// <summary>
+/// Replays the last result before requesting a new one
+/// </summary>
+/// <typeparam name="TRequest"></typeparam>
+/// <typeparam name="TResult"></typeparam>
+public class ReplayStreamMiddleware<TRequest, TResult> : IStreamRequestMiddleware<TRequest, TResult> 
+    where TRequest : IStreamRequest<TResult>
+{
+    public IAsyncEnumerator<TResult> Process(
+        TRequest request, 
+        StreamRequestDelegate<TResult> next, 
+        IStreamRequestHandler<TRequest, TResult> requestHandler,
+        CancellationToken cancellationToken
+    )
+    {
+        var attribute = requestHandler.GetHandlerHandleMethodAttribute<TRequest, ReplayAttribute>();
+        if (attribute == null)
+            return next();
+
+        return this.Iterate(request, next, cancellationToken);
+    }
+
+
+    protected virtual async IAsyncEnumerator<TResult> Iterate(TRequest request, StreamRequestDelegate<TResult> next, CancellationToken ct)
+    {
+        var path = this.GetCacheFilePath(request);
+        if (File.Exists(path))
+        {
+            var json = File.ReadAllText(path);
+            var obj = JsonSerializer.Deserialize<TResult>(json);
+            yield return obj;
+        }
+
+        var nxt = next();
+        while (await nxt.MoveNextAsync() && !ct.IsCancellationRequested)
+        {
+            var json = JsonSerializer.Serialize(nxt.Current);
+            File.WriteAllText(path, json);
+            yield return nxt.Current;
+        }
+    }
+
+    protected virtual string GetCacheFilePath(TRequest request)
+    {
+        if (request is IReplayKey<TResult> key)
+            return key.Key;
+
+        var t = request.GetType();
+        return $"{t.Namespace}_{t.Name}.replay";
+    }
+}

src/Shiny.Mediator/Middleware/TimerRefreshStreamRequestMiddleware.cs

@@ -0,0 +1,32 @@
+namespace Shiny.Mediator.Middleware;
+
+public class TimerRefreshStreamRequestMiddleware<TRequest, TResult> : IStreamRequestMiddleware<TRequest, TResult> 
+    where TRequest : IStreamRequest<TResult>
+{
+    public IAsyncEnumerator<TResult> Process(
+        TRequest request, 
+        StreamRequestDelegate<TResult> next, 
+        IStreamRequestHandler<TRequest, TResult> requestHandler,
+        CancellationToken cancellationToken
+    )
+    {
+        var attribute = requestHandler.GetHandlerHandleMethodAttribute<TRequest, TimerRefreshAttribute>();
+        if (attribute == null)
+            return next();
+
+        return Iterate(attribute, next, cancellationToken);
+    }
+
+
+    async IAsyncEnumerator<TResult> Iterate(TimerRefreshAttribute attribute, StreamRequestDelegate<TResult> next, CancellationToken ct)
+    {
+        while (!ct.IsCancellationRequested)
+        {
+            await Task.Delay(attribute.RefreshSeconds, ct);
+
+            var nxt = next();
+            while (await nxt.MoveNextAsync() && !ct.IsCancellationRequested)
+                yield return nxt.Current;
+        }
+    }
+}

src/Shiny.Mediator/MiddlewareModels.cs

@@ -0,0 +1,17 @@
+namespace Shiny.Mediator;
+
+[AttributeUsage(AttributeTargets.Method, Inherited = false, AllowMultiple = false)]
+public class TimerRefreshAttribute(int refreshSeconds, bool ignoreErrors = true) : Attribute
+{
+    public int RefreshSeconds => refreshSeconds;
+    public bool IgnoreErrors => ignoreErrors;
+}
+
+
+[AttributeUsage(AttributeTargets.Method, Inherited = false, AllowMultiple = false)]
+public class ReplayAttribute : Attribute {}
+
+public interface IReplayKey<TResult> : IStreamRequest<TResult>
+{
+    string Key { get; }
+}

src/Shiny.Mediator/ShinyConfigurator.cs

@@ -52,7 +52,14 @@ public ShinyConfigurator AddOpenRequestMiddleware(Type implementationType, Servi
         services.Add(new ServiceDescriptor(typeof(IRequestMiddleware<,>), null, implementationType, lifetime));
         return this;
     }
-
+
+
+    public ShinyConfigurator AddOpenStreamMiddleware(Type implementationType, ServiceLifetime lifetime = ServiceLifetime.Scoped) 
+    {
+        // TODO: validate open generic
+        services.Add(new ServiceDescriptor(typeof(IStreamRequestMiddleware<,>), null, implementationType, lifetime));
+        return this;
+    }
 
     public ShinyConfigurator AddOpenEventMiddleware(Type implementationType, ServiceLifetime lifetime = ServiceLifetime.Scoped)
     {