重新整理 Workers

Shared/General/Progress.cs

@@ -121,26 +121,17 @@ public class ProgressProvider {
 /// <summary>
 /// 监听进度变化，并将观测到的进度节流后传给更新方法。
 /// </summary>
-public class ProgressObserver : IDisposable {
+public class ProgressObserver : RateLimitedWorker {
     private readonly ProgressProvider provider;
-    private readonly RateLimitedAction updateAction;
 
-    public ProgressObserver(ProgressProvider provider, Action<double> updateAction, double minimalIntervalMs = 70) {
+    public ProgressObserver(ProgressProvider provider, Action<double> updateAction, double minimalIntervalMs = 70)
+        : base(() => updateAction(provider.Observe()), minimalIntervalMs, RateLimitMode.ImmediateThenMerge) {
         this.provider = provider;
-        this.updateAction = Throttler.Throttle(() => updateAction(provider.Observe()), 
-            TimeSpan.FromMilliseconds(minimalIntervalMs), leading: true, trailing: true);
-        provider.ProgressChanged += OnProgressChanged;
+        provider.ProgressChanged += Invoke;
     }
 
-    private void OnProgressChanged() {
-        if (!disposed) updateAction.Invoke();
+    public override void Dispose() {
+        provider.ProgressChanged -= Invoke;
+        base.Dispose();
     }
-
-    private bool disposed = false;
-    public void Dispose() {
-        if (disposed) return;
-        disposed = true;
-        provider.ProgressChanged -= OnProgressChanged;
-        updateAction.Dispose();
-    }
-}
+}

Shared/General/Workers/RateLimitedWorker.cs

@@ -0,0 +1,35 @@
+namespace MeloongCore;
+
+public enum RateLimitMode {
+    /// <summary>Throttle：立即执行一次，然后一段时间内合并为一次尾随执行。</summary>
+    ImmediateThenMerge,
+    /// <summary>Debounce (leading)：立即执行一次，然后忽略一段时间内的后续调用。</summary>
+    ImmediateThenIgnore,
+    /// <summary>Debounce (trailing)：不立即执行，直到一段时间内不再有任何新调用后再执行一次。</summary>
+    WaitUnillNoNewInvoke,
+}
+
+/// <summary>
+/// 对高频调用进行限制。
+/// 详见：<see href="https://aldaviva.com/portfolio.html#ratelimiting" />
+/// </summary>
+public class RateLimitedWorker(Action workload, double minimalIntervalMs, RateLimitMode mode) : IDisposable {
+
+    private readonly RateLimitedAction action = mode switch {
+        RateLimitMode.ImmediateThenMerge => Throttler.Throttle(workload, TimeSpan.FromMilliseconds(minimalIntervalMs), leading: true, trailing: true),
+        RateLimitMode.WaitUnillNoNewInvoke => Debouncer.Debounce(workload, TimeSpan.FromMilliseconds(minimalIntervalMs), leading: false, trailing: true),
+        RateLimitMode.ImmediateThenIgnore => Debouncer.Debounce(workload, TimeSpan.FromMilliseconds(minimalIntervalMs), leading: true, trailing: false)
+    };
+
+    public void Invoke() {
+        if (!disposed) action.Invoke();
+    }
+
+    private bool disposed = false;
+    public virtual void Dispose() {
+        if (disposed) return;
+        disposed = true;
+        action.Dispose();
+    }
+
+}

Shared/General/Worker.cs → Shared/General/Workers/RedoableWorker.cs

@@ -2,7 +2,7 @@ namespace MeloongCore;
 
 /// <summary>
 /// 可合并和重启的工作器。
-/// <para/> 在工作负载运行期间多次调用 <see cref="Run()"/> 会取消当前负载并使用最新令牌重新执行；多次调用也只重新执行一次。
+/// <para/> 在工作负载运行期间多次调用 <see cref="Invoke(CancellationToken)"/> 会取消当前负载并使用最新令牌重新执行；多次调用也只重新执行一次。
 /// </summary>
 public interface IRedoableWorker {
     /// <summary>当前是否正在运行。</summary>
@@ -12,7 +12,7 @@ public interface IRedoableWorker {
     /// <summary>标识上次进入空闲时，并非因为取消或失败，而是正常运行到结束。</summary>
     bool LastSucceeded { get; }
     /// <summary>在工作线程运行工作负载。若当前已在运行，则取消当前负载并使用最新令牌重启。</summary>
-    void Run(CancellationToken cancellationToken = default);
+    void Invoke(CancellationToken cancellationToken = default);
     /// <summary>取消当前运行。</summary>
     void Cancel();
     /// <summary>仅在当前处于运行状态时，等待其完成。</summary>
@@ -57,7 +57,7 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken, TOut> wor
     private readonly ManualResetEventSlim idleEvent = new(initialState: true);
 
     /// <inheritdoc/>
-    public void Run(CancellationToken cancellationToken = default) {
+    public void Invoke(CancellationToken cancellationToken = default) {
         // 接取运行状态
         lock (this) {
             lastToken = cancellationToken;
@@ -69,10 +69,10 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken, TOut> wor
             running = true; idleEvent.Reset();
             realCts = CancellationTokenSource.CreateLinkedTokenSource(lastToken);
         }
-        var th = new Thread(_Run) { IsBackground = true };
+        var th = new Thread(_Invoke) { IsBackground = true };
         th.Start();
     }
-    private void _Run() {
+    private void _Invoke() {
         while (true) {
             try {
                 realCts!.Token.ThrowIfCancellationRequested();
@@ -125,13 +125,13 @@ public abstract class RedoableWorkerBase<TOut>(Func<CancellationToken, TOut> wor
 }
 
 /// <inheritdoc />
-public sealed class RedoableWorker<TOut> : RedoableWorkerBase<TOut> {
+public class RedoableWorker<TOut> : RedoableWorkerBase<TOut> {
     public RedoableWorker(Func<CancellationToken, TOut> workload) : base(workload) { }
     public RedoableWorker(Func<TOut> workload) : base(_ => workload()) { }
 }
 
 /// <inheritdoc />
-public sealed class RedoableWorker : RedoableWorkerBase<object?> {
+public class RedoableWorker : RedoableWorkerBase<object?> {
     public RedoableWorker(Action<CancellationToken> workload) : base(ct => { workload(ct); return null; }) { }
     public RedoableWorker(Action workload) : base(_ => { workload(); return null; }) { }
 }

Tests/General/WorkerTest.cs

@@ -12,7 +12,7 @@ public class WorkerTest : TestBase {
             started.Set();
         });
 
-        worker.Run();
+        worker.Invoke();
 
         await Assert.That(started.Wait(3000)).IsTrue();
         await Assert.That(worker.WaitIfRunning(3000)).IsTrue();
@@ -38,10 +38,10 @@ public class WorkerTest : TestBase {
             }
         });
 
-        worker.Run();
+        worker.Invoke();
         await Assert.That(firstStarted.Wait(3000)).IsTrue();
-        worker.Run();
-        worker.Run();
+        worker.Invoke();
+        worker.Invoke();
 
         await Assert.That(worker.WaitIfRunning(3000)).IsTrue();
         await Assert.That(runCount).IsEqualTo(2);