27

リアクティブ拡張機能を使用して、データを 100 ミリ秒のバッファーに照合しています。

this.subscription = this.dataService
    .Where(x => !string.Equals("FOO", x.Key.Source))
    .Buffer(TimeSpan.FromMilliseconds(100))
    .ObserveOn(this.dispatcherService)
    .Where(x => x.Count != 0)
    .Subscribe(this.OnBufferReceived);

これはうまくいきます。ただし、操作によって提供される動作とは少し異なる動作が必要ですBuffer。基本的に、別のデータ項目を受信した場合にタイマーをリセットしたい。100ミリ秒全体でデータが受信されなかった場合にのみ、それを処理したい. これにより、データをまったく処理しない可能性が開かれるため、最大カウントも指定できるはずです。次のようなものを想像します。

.SlidingBuffer(TimeSpan.FromMilliseconds(100), 10000)

私は周りを見回しましたが、Rxでこのようなものを見つけることができませんでしたか? 誰でもこれを確認/拒否できますか?

4

5 に答える 5

19

Windowこれは、組み込みのとThrottleメソッドを組み合わせることで可能になりますObservable。まず、最大カウント条件を無視する単純な問題を解決しましょう。

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay)
{
    var closes = stream.Throttle(delay);
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

強力なWindow方法で重い物を持ち上げました。これで、最大カウントを追加する方法を簡単に確認できます。

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? max=null)
{
    var closes = stream.Throttle(delay);
    if (max != null)
    {
        var overflows = stream.Where((x,index) => index+1>=max);
        closes = closes.Merge(overflows);
    }
    return stream.Window(() => closes).SelectMany(window => window.ToList());
}

これを説明する記事をブログに書きます。https://gist.github.com/2244036

Window メソッドのドキュメント:

于 2012-03-20T16:56:58.280 に答える
16

あなたが望んでいることのほとんどを行うための拡張機能を書きました - BufferWithInactivity。

ここにあります：

public static IObservable<IEnumerable<T>> BufferWithInactivity<T>(
    this IObservable<T> source,
    TimeSpan inactivity,
    int maximumBufferSize)
{
    return Observable.Create<IEnumerable<T>>(o =>
    {
        var gate = new object();
        var buffer = new List<T>();
        var mutable = new SerialDisposable();
        var subscription = (IDisposable)null;
        var scheduler = Scheduler.ThreadPool;

        Action dump = () =>
        {
            var bts = buffer.ToArray();
            buffer = new List<T>();
            if (o != null)
            {
                o.OnNext(bts);
            }
        };

        Action dispose = () =>
        {
            if (subscription != null)
            {
                subscription.Dispose();
            }
            mutable.Dispose();
        };

        Action<Action<IObserver<IEnumerable<T>>>> onErrorOrCompleted =
            onAction =>
            {
                lock (gate)
                {
                    dispose();
                    dump();
                    if (o != null)
                    {
                        onAction(o);
                    }
                }
            };

        Action<Exception> onError = ex =>
            onErrorOrCompleted(x => x.OnError(ex));

        Action onCompleted = () => onErrorOrCompleted(x => x.OnCompleted());

        Action<T> onNext = t =>
        {
            lock (gate)
            {
                buffer.Add(t);
                if (buffer.Count == maximumBufferSize)
                {
                    dump();
                    mutable.Disposable = Disposable.Empty;
                }
                else
                {
                    mutable.Disposable = scheduler.Schedule(inactivity, () =>
                    {
                        lock (gate)
                        {
                            dump();
                        }
                    });
                }
            }
        };

        subscription =
            source
                .ObserveOn(scheduler)
                .Subscribe(onNext, onError, onCompleted);

        return () =>
        {
            lock (gate)
            {
                o = null;
                dispose();
            }
        };
    });
}
于 2011-09-30T00:32:24.150 に答える
2

Rx Extensions 2.0 では、タイムアウトとサイズを受け入れる新しいバッファー オーバーロードで両方の要件を満たすことができます。

this.subscription = this.dataService
    .Where(x => !string.Equals("FOO", x.Key.Source))
    .Buffer(TimeSpan.FromMilliseconds(100), 1)
    .ObserveOn(this.dispatcherService)
    .Where(x => x.Count != 0)
    .Subscribe(this.OnBufferReceived);

ドキュメントについては、https://msdn.microsoft.com/en-us/library/hh229200 (v=vs.103).aspx を参照してください。

于 2015-03-19T20:06:41.760 に答える
0

以下に示すように、これは Buffer メソッドの上に実装できると思います。

public static IObservable<IList<T>> SlidingBuffer<T>(this IObservable<T> obs, TimeSpan span, int max)
        {
            return Observable.CreateWithDisposable<IList<T>>(cl =>
            {
                var acc = new List<T>();
                return obs.Buffer(span)
                        .Subscribe(next =>
                        {
                            if (next.Count == 0) //no activity in time span
                            {
                                cl.OnNext(acc);
                                acc.Clear();
                            }
                            else
                            {
                                acc.AddRange(next);
                                if (acc.Count >= max) //max items collected
                                {
                                    cl.OnNext(acc);
                                    acc.Clear();
                                }
                            }
                        }, err => cl.OnError(err), () => { cl.OnNext(acc); cl.OnCompleted(); });
            });
        }

注：私はそれをテストしていませんが、アイデアが得られることを願っています.

于 2011-10-01T08:11:53.707 に答える