OCRトラッカーを独立したループで動かし、保持・安定化を時間ベースにする

WindowTranslator.Abstractions/UserSettings.cs

@@ -108,7 +108,7 @@ public class TargetSettings
     public int OcrRecognitionStability { get; set => field = Math.Clamp(value, 1, 5); } = 3;
 
     /// <summary>
-    /// 表示を保持する連続OCR欠落回数（0～7回）
+    /// 認識できなくなった領域の表示保持の強さ（0: 即時除去、7: 保持重視）
     /// </summary>
     public int OcrMissingFrameRetention { get; set => field = Math.Clamp(value, 0, 7); } = 3;
 

WindowTranslator.Tests/ObservableExtensionsTests.cs

@@ -0,0 +1,241 @@
+using System.Collections.Concurrent;
+using R3;
+using WindowTranslator.Extensions;
+
+namespace WindowTranslator.Tests;
+
+public class ObservableExtensionsTests
+{
+    [Fact]
+    public async Task 処理中に届いた値は最後の値だけを実行する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+        var replaced = new TestValue(2);
+        var last = new TestValue(3);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(replaced);
+        source.OnNext(last);
+        last.Release();
+        first.Release();
+        await last.WaitUntilDisposedAsync();
+
+        Assert.Equal([1, 3], processor.StartedIds.ToArray());
+    }
+
+    [Fact]
+    public async Task 差し替えられた待機値は一度だけ破棄する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+        var replaced = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(replaced);
+        source.OnNext(new TestValue(3));
+        subscription.Dispose();
+
+        Assert.Equal(1, replaced.DisposeCount);
+    }
+
+    [Fact]
+    public async Task 待機値を差し替えても実行中の値は破棄しない()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(new TestValue(2));
+        source.OnNext(new TestValue(3));
+
+        Assert.Equal(0, first.DisposeCount);
+    }
+
+    [Fact]
+    public async Task 購読解除で待機値を一度だけ破棄する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+        var pending = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(pending);
+        subscription.Dispose();
+        subscription.Dispose();
+
+        Assert.Equal(1, pending.DisposeCount);
+    }
+
+    [Fact]
+    public async Task 購読解除で実行中の処理をキャンセルする()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        subscription.Dispose();
+        await first.WaitUntilDisposedAsync();
+
+        Assert.True(first.CancellationToken.IsCancellationRequested);
+    }
+
+    [Fact]
+    public async Task 通常完了では最後の待機値まで処理する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+        var last = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(last);
+        source.OnCompleted();
+        last.Release();
+        first.Release();
+        await last.WaitUntilDisposedAsync();
+
+        Assert.Equal([1, 2], processor.StartedIds.ToArray());
+    }
+
+    [Fact]
+    public async Task 完了時キャンセルを指定すると待機値を破棄する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(
+            processor.ProcessAsync, Discard, configureAwait: false, cancelOnCompleted: true);
+        var first = new TestValue(1);
+        var pending = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(pending);
+        source.OnCompleted();
+
+        Assert.Equal(1, pending.DisposeCount);
+    }
+
+    [Fact]
+    public async Task 失敗による完了では待機値を破棄する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        var errors = new ConcurrentQueue<Exception>();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(
+            processor.ProcessAsync, Discard, configureAwait: false, onErrorResume: errors.Enqueue);
+        var first = new TestValue(1);
+        var pending = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(pending);
+        source.OnCompleted(new InvalidOperationException());
+
+        Assert.Equal(1, pending.DisposeCount);
+    }
+
+    [Fact]
+    public async Task 処理の例外を通知して次の待機値を処理する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        var errors = new ConcurrentQueue<Exception>();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(
+            processor.ProcessAsync, Discard, configureAwait: false, onErrorResume: errors.Enqueue);
+        var first = new TestValue(1);
+        var last = new TestValue(2);
+        var error = new InvalidOperationException();
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(last);
+        last.Release();
+        first.Fail(error);
+        await last.WaitUntilDisposedAsync();
+
+        Assert.Same(error, Assert.Single(errors));
+    }
+
+    [Fact]
+    public async Task 処理がキャンセルされた場合も残った待機値を破棄する()
+    {
+        using var source = new Subject<TestValue>();
+        var processor = new Processor();
+        using var subscription = source.SubscribeAwaitThrottleFirstLast(processor.ProcessAsync, Discard, configureAwait: false);
+        var first = new TestValue(1);
+        var pending = new TestValue(2);
+
+        source.OnNext(first);
+        await first.WaitUntilStartedAsync();
+        source.OnNext(pending);
+        first.Fail(new OperationCanceledException());
+        await pending.WaitUntilDisposedAsync();
+
+        Assert.Equal(1, pending.DisposeCount);
+    }
+
+    private static void Discard(TestValue value) => value.Dispose();
+
+    private sealed class Processor
+    {
+        public ConcurrentQueue<int> StartedIds { get; } = new();
+
+        public async ValueTask ProcessAsync(TestValue value, CancellationToken cancellationToken)
+        {
+            using (value)
+            {
+                this.StartedIds.Enqueue(value.Id);
+                value.Start(cancellationToken);
+                await value.WaitUntilReleasedAsync(cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
+    private sealed class TestValue(int id) : IDisposable
+    {
+        private readonly TaskCompletionSource started = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource released = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource disposed = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int disposeCount;
+
+        public int Id { get; } = id;
+        public int DisposeCount => Volatile.Read(ref this.disposeCount);
+        public CancellationToken CancellationToken { get; private set; }
+
+        public void Start(CancellationToken cancellationToken)
+        {
+            this.CancellationToken = cancellationToken;
+            this.started.TrySetResult();
+        }
+
+        public void Release() => this.released.TrySetResult();
+        public void Fail(Exception error) => this.released.TrySetException(error);
+        public Task WaitUntilStartedAsync() => this.started.Task.WaitAsync(TimeSpan.FromSeconds(10));
+        public Task WaitUntilDisposedAsync() => this.disposed.Task.WaitAsync(TimeSpan.FromSeconds(10));
+        public Task WaitUntilReleasedAsync(CancellationToken cancellationToken) => this.released.Task.WaitAsync(cancellationToken);
+
+        public void Dispose()
+        {
+            Interlocked.Increment(ref this.disposeCount);
+            this.disposed.TrySetResult();
+        }
+    }
+}

WindowTranslator/Extensions/ObservableExtensions.cs

@@ -0,0 +1,144 @@
+using System.Threading.Channels;
+using R3;
+
+namespace WindowTranslator.Extensions;
+
+public static class ObservableExtensions
+{
+    /// <summary>
+    /// 先頭と処理中に届いた最新の値を順次処理し、処理されない待機値を破棄します。
+    /// </summary>
+    /// <remarks>
+    /// 処理を開始した値の後始末は onNextAsync、差し替え・購読終了で不要になった値の後始末は onDiscard が担当します。
+    /// </remarks>
+    public static IDisposable SubscribeAwaitThrottleFirstLast<T>(
+        this Observable<T> source,
+        Func<T, CancellationToken, ValueTask> onNextAsync,
+        Action<T> onDiscard,
+        bool configureAwait = true,
+        bool cancelOnCompleted = false,
+        Action<Exception>? onErrorResume = null)
+    {
+        ArgumentNullException.ThrowIfNull(source);
+        ArgumentNullException.ThrowIfNull(onNextAsync);
+        ArgumentNullException.ThrowIfNull(onDiscard);
+
+        return source.Subscribe(new ThrottleFirstLastObserver<T>(
+            onNextAsync, onDiscard, configureAwait, cancelOnCompleted,
+            onErrorResume ?? ObservableSystem.GetUnhandledExceptionHandler()));
+    }
+
+    private sealed class ThrottleFirstLastObserver<T> : Observer<T>
+    {
+        private readonly Func<T, CancellationToken, ValueTask> onNextAsync;
+        private readonly Action<T> onDiscard;
+        private readonly Action<Exception> onErrorResume;
+        private readonly bool configureAwait;
+        private readonly bool cancelOnCompleted;
+        private readonly Channel<T> channel;
+        private readonly IDisposable worker;
+
+        protected override bool AutoDisposeOnCompleted => false;
+
+        public ThrottleFirstLastObserver(
+            Func<T, CancellationToken, ValueTask> onNextAsync,
+            Action<T> onDiscard,
+            bool configureAwait,
+            bool cancelOnCompleted,
+            Action<Exception> onErrorResume)
+        {
+            this.onNextAsync = onNextAsync;
+            this.onDiscard = onDiscard;
+            this.onErrorResume = onErrorResume;
+            this.configureAwait = configureAwait;
+            this.cancelOnCompleted = cancelOnCompleted;
+            this.channel = Channel.CreateBounded<T>(new BoundedChannelOptions(1)
+            {
+                AllowSynchronousContinuations = true,
+                FullMode = BoundedChannelFullMode.DropOldest,
+            }, onDiscard);
+            this.worker = Observable.FromAsync(RunAsync, configureAwait)
+                .Subscribe(_ => { }, onErrorResume, Complete);
+        }
+
+        protected override void OnNextCore(T value)
+        {
+            if (!this.channel.Writer.TryWrite(value))
+            {
+                this.onDiscard(value);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error) => this.onErrorResume(error);
+
+        protected override void OnCompletedCore(Result result)
+        {
+            this.channel.Writer.TryComplete();
+            if (this.cancelOnCompleted || result.IsFailure)
+            {
+                Complete(result);
+            }
+        }
+
+        protected override void DisposeCore()
+        {
+            this.channel.Writer.TryComplete();
+            this.worker.Dispose();
+            while (this.channel.Reader.TryRead(out var value))
+            {
+                this.onDiscard(value);
+            }
+        }
+
+        private void Complete(Result result)
+        {
+            try
+            {
+                if (result.Exception is { } error)
+                {
+                    this.onErrorResume(error);
+                }
+            }
+            finally
+            {
+                Dispose();
+            }
+        }
+
+        private async ValueTask RunAsync(CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (await this.channel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(this.configureAwait))
+                {
+                    while (this.channel.Reader.TryRead(out var value))
+                    {
+                        if (cancellationToken.IsCancellationRequested)
+                        {
+                            this.onDiscard(value);
+                            return;
+                        }
+
+                        try
+                        {
+                            await this.onNextAsync(value, cancellationToken).ConfigureAwait(this.configureAwait);
+                        }
+                        catch (OperationCanceledException)
+                        {
+                            return;
+                        }
+                        // 購読コールバックの例外は R3 と同様に通知し、次の待機値の処理を続ける。
+                        catch (Exception error) when (error is not OperationCanceledException)
+                        {
+                            this.onErrorResume(error);
+                        }
+                    }
+                }
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                // 購読解除による終了では、待機値の破棄を DisposeCore が担当する。
+            }
+        }
+    }
+}