具有重复的示例运算符

Question

我想要一个与示例运算符等价的、不隐式过滤其结果流的方法：

var ints = new Subject<int>();
var pulse = new Subject<Unit>();
ints.Sample(pulse)
    .Subscribe(Console.WriteLine);
ints.OnNext(1);
pulse.OnNext(Unit.Default);
ints.OnNext(2);
pulse.OnNext(Unit.Default);
pulse.OnNext(Unit.Default);
ints.OnNext(3);
// Prints 1, 2 instead of 1, 2, 2

我的连接尝试在上面的例子中起到了作用，但在其他情况下（在热可观测性上）失败了：

public static IObservable<TIn> SampleWithRepetition<TIn, TPulse>(this IObservable<TIn> source, IObservable<TPulse> pulse)
{
    return from x in source
           join _ in pulse on source equals Observable.Empty<Unit>()
           select x;
}

是否可以在不使用基于状态的Observable.Create方法的情况下定义这样的运算符？

Answer 1

以下运算符是基于状态的；可能有一种更实用的方法

public static IObservable<TIn> SampleWithRepetition<TIn, TPulse>(this IObservable<TIn> source, IObservable<TPulse> pulse)
{
    return Observable.Create<TIn>(obs =>
        {
            var gate = new object();
            var x = new {Value = default(TIn), HasValue = false};
            return new CompositeDisposable()
                {
                    source.Subscribe(
                        y =>
                            {
                                lock (gate)
                                    x = new {Value = y, HasValue = true};
                            },
                        obs.OnError,
                        obs.OnCompleted),
                    pulse.Subscribe(
                        _ =>
                            {
                                bool hasValue = false;
                                TIn value = default(TIn);
                                lock (gate)
                                {
                                    hasValue = x.HasValue;
                                    value = x.Value;
                                }
                                if (hasValue)
                                    obs.OnNext(value);
                            },
                        obs.OnError,
                        obs.OnCompleted)
                };
        });
}

}