我对响应式编程的概念相当陌生。我正在使用盆景，它通过 c# 公开一些但不是全部的 .Net rx 命令。
我正在尝试获得像这样的大理石图的行为：
input1: ---1--------2--------3--------4--------5--------6--------7
input2: -------abc----------------------------------def-----------
result: ------------a--------b--------c--------c---------d-------e
基本上，输入 2 生成应存储在队列中的事件波。 输入 1 充当从此队列发出单个项目的触发器。
当队列为空时，应发出队列的最后一项。 我尝试了zip和combineLate的各种组合，但我无法获得所需的行为。
我还尝试了基于这篇文章的WithLatestFrom实现，但回想起来，我意识到这也不会产生所需的行为。
public IObservable<Tuple<TSource, TOther>> Process<TSource, TOther>(
IObservable<TSource> source,
IObservable<TOther> other)
{
// return source1.WithLatestFrom(source2, (xs, ys) => Tuple.Create(xs, ys));
return source.Publish(os => other.Select(a => os.Select(b => Tuple.Create(b, a))).Switch());
}
是否有任何运算符或运算符组合会产生此行为？一旦我了解了要使用哪些运算符，我就可以对 Bonsai 进行实现。
更新1： 2018/05/18
根据 Sentinel 的帖子，我在 Bonsai 命名空间中编写了一个新的类DiscriminatedUnion。不过，我没有设法指定适当的类型。编译器声明"无法推断Merge的类型参数"(.Merge(input1.Select...)。 在哪里添加正确的类型规范？
using System.Reactive.Linq;
using System.ComponentModel;
using System.Collections.Immutable;
namespace Bonsai.Reactive
{
[Combinator]
// [XmlType(Namespace = Constants.XmlNamespace)]
[Description("Implementation of Discriminated Union")]
public class DiscriminatedUnion
{
public IObservable<int?> Process<TInput1, TInput2>(
IObservable<TInput1> input1,
IObservable<TInput2> input2)
{
var merged =
input2.Select(s2 => Tuple.Create(2, (TInput2)s2))
.Merge(input1.Select(s1 => Tuple.Create(1, (TInput1)s1)))
.Scan(Tuple.Create((int?)null, new Queue<int>(), 0), (state, val) =>
{
int? next = state.Item1;
if (val.Item1 == 1)
{
if (state.Item2.Count > 0)
{
next = state.Item2.Dequeue();
}
}
else
{
state.Item2.Enqueue(val.Item2);
}
return Tuple.Create(next, state.Item2, val.Item1);
})
.Where(x => (x.Item1 != null && x.Item3 == 1))
.Select(x => x.Item1);
return merged;
}
}
}
下面是使用 NuGet 包Microsoft.Reactive.Testing的问题(或弹珠图)的可测试表示形式：
var scheduler = new TestScheduler();
var input1 = scheduler.CreateColdObservable<int>(
ReactiveTest.OnNext(1000.Ms(), 1),
ReactiveTest.OnNext(2000.Ms(), 2),
ReactiveTest.OnNext(3000.Ms(), 3),
ReactiveTest.OnNext(4000.Ms(), 4),
ReactiveTest.OnNext(5000.Ms(), 5),
ReactiveTest.OnNext(6000.Ms(), 6),
ReactiveTest.OnNext(7000.Ms(), 7)
);
var input2 = scheduler.CreateColdObservable<string>(
ReactiveTest.OnNext(1400.Ms(), "a"),
ReactiveTest.OnNext(1500.Ms(), "b"),
ReactiveTest.OnNext(1600.Ms(), "c"),
ReactiveTest.OnNext(5500.Ms(), "d"),
ReactiveTest.OnNext(5600.Ms(), "e"),
ReactiveTest.OnNext(5700.Ms(), "f")
);
它使用此扩展方法：
public static class TickExtensions
{
public static long Ms(this int ms)
{
return TimeSpan.FromMilliseconds(ms).Ticks;
}
}
该问题基本上是一个状态机问题，涉及两个不同类型的可观察量。解决此问题的最佳方法是使用可区分联合类型，该类型在 C# 中不存在，因此我们将创建一个。@Sentinel的回答是用元组做到了这一点，这也可以工作：
public class DUnion<T1, T2>
{
public DUnion(T1 t1)
{
Type1Item = t1;
Type2Item = default(T2);
IsType1 = true;
}
public DUnion(T2 t2)
{
Type2Item = t2;
Type1Item = default(T1);
IsType1 = false;
}
public bool IsType1 { get; }
public bool IsType2 => !IsType1;
public T1 Type1Item { get; }
public T2 Type2Item { get; }
}
然后，我们可以将两个不同类型的流Select并将它们Merge到一个可区分的联合流中，在那里我们可以用Scan管理状态。您的状态逻辑有点棘手，但可行：
- 如果号码到达并且队列中没有项目，则不执行任何操作
- 如果数字到达并且队列中有项目，请发出队列中的第一项。
- 如果有多个项目，请从队列中删除最近的发射。
- 如果队列只有一个项目，请不要将其删除，并进入"假空"状态。
- 如果字符串到达，请将其粘贴到队列中。 如果队列为"假空">
- ，则弹出最后一项并退出"假空"状态。
下面是生成的可观察对象(使用 NuGet 包System.Collections.Immutable)：
var result = input1.Select(i => new DUnion<int, string>(i))
.Merge(input2.Select(s => new DUnion<int, string>(s)))
.Scan((queue: ImmutableQueue<string>.Empty, item: (string)null, isFakeEmptyState: false, emit: false), (state, dItem) => dItem.IsType1
? state.queue.IsEmpty
? (state.queue, null, false, false) //Is integer, but empty queue, so don't emit item
: state.queue.Dequeue().IsEmpty //Is integer, at least one item: dequeue unless only one item, then emit either way
? (state.queue, state.queue.Peek(), true, true)
: (state.queue.Dequeue(), state.queue.Peek(), false, true)
: state.isFakeEmptyState //Is new string, just add to queue, don't emit
? (state.queue.Dequeue().Enqueue(dItem.Type2Item), null, false, false)
: (state.queue.Enqueue(dItem.Type2Item), (string)null, false, false)
)
.Where(t => t.emit)
.Select(t => t.item);
然后可以按如下方式进行测试：
var observer = scheduler.CreateObserver<string>();
result.Subscribe(observer);
scheduler.Start();
observer.Messages.Dump(); //Linqpad. Can replace with Console.Writeline loop.
更新：我想了一下，我认为围绕歧视联盟功能抛出一些运算符是有意义的。这样，您就不必显式处理类型：
public static class DUnionExtensions
{
public class DUnion<T1, T2>
{
public DUnion(T1 t1)
{
Type1Item = t1;
Type2Item = default(T2);
IsType1 = true;
}
public DUnion(T2 t2)
{
Type2Item = t2;
Type1Item = default(T1);
IsType1 = false;
}
public bool IsType1 { get; }
public bool IsType2 => !IsType1;
public T1 Type1Item { get; }
public T2 Type2Item { get; }
}
public static IObservable<DUnion<T1, T2>> Union<T1, T2>(this IObservable<T1> a, IObservable<T2> b)
{
return a.Select(x => new DUnion<T1, T2>(x))
.Merge(b.Select(x => new DUnion<T1, T2>(x)));
}
public static IObservable<TState> ScanUnion<T1, T2, TState>(this IObservable<DUnion<T1, T2>> source,
TState initialState,
Func<TState, T1, TState> type1Handler,
Func<TState, T2, TState> type2Handler)
{
return source.Scan(initialState, (state, u) => u.IsType1
? type1Handler(state, u.Type1Item)
: type2Handler(state, u.Type2Item)
);
}
}
使用这些扩展方法，解决方案将更改为此，我认为读起来更好：
var result = input1
.Union(input2)
.ScanUnion((queue: ImmutableQueue<string>.Empty, item: (string)null, isFakeEmptyState: false, emit: false),
(state, _) => state.queue.IsEmpty
? (state.queue, null, false, false) //empty queue, so don't emit item
: state.queue.Dequeue().IsEmpty //At least one item: dequeue unless only one item, then emit either way
? (state.queue, state.queue.Peek(), true, true) //maintain last item, enter Fake-EmptyState
: (state.queue.Dequeue(), state.queue.Peek(), false, true),
(state, s) => state.isFakeEmptyState
? (state.queue.Dequeue().Enqueue(s), null, false, false)
: (state.queue.Enqueue(s), (string)null, false, false)
)
.Where(t => t.emit)
.Select(t => t.item);
如果您在使用命名元组语法时遇到问题，则可以使用旧元组：
var result = input1
.Union(input2)
.ScanUnion(Tuple.Create(ImmutableQueue<string>.Empty, (string)null, false, false),
(state, _) => state.Item1.IsEmpty
? Tuple.Create(state.Item1, (string)null, false, false) //empty queue, so don't emit item
: state.Item1.Dequeue().IsEmpty //At least one item: dequeue unless only one item, then emit either way
? Tuple.Create(state.Item1, state.Item1.Peek(), true, true) //maintain last item, enter Fake-EmptyState
: Tuple.Create(state.Item1.Dequeue(), state.Item1.Peek(), false, true),
(state, s) => state.Item3
? Tuple.Create(state.Item1.Dequeue().Enqueue(s), (string)null, false, false)
: Tuple.Create(state.Item1.Enqueue(s), (string)null, false, false)
)
.Where(t => t.Item4)
.Select(t => t.Item2);
这能解决问题吗？可能有一种更好的方法来执行此缓冲区，因此可能值得重新访问。
Random r = new Random();
var source1 = Observable.Interval(TimeSpan.FromSeconds(1)).Publish().RefCount();
var source2 = Observable.Interval(TimeSpan.FromMilliseconds(7000)).Select(x => Enumerable.Range(1, 3).Select(y => r.Next(200)).ToObservable()).SelectMany(x => x).Publish().RefCount();
source1.Subscribe(x => Console.WriteLine("Source1 " + x));
source2.Subscribe(x => Console.WriteLine("Source2 " + x));
var merged =
source2.Select(s2 => Tuple.Create(2, s2))
.Merge(source1.Select(s1 => Tuple.Create(1, (int)s1)))
.Scan(Tuple.Create((int?)null, new Queue<int>(),0), (state, val) =>
{
int? next = state.Item1;
if (val.Item1 == 1)
{
if (state.Item2.Count > 0)
{
next = state.Item2.Dequeue();
}
}
else
{
state.Item2.Enqueue(val.Item2);
}
return Tuple.Create(next, state.Item2,val.Item1);
})
.Where(x=>(x.Item1!=null && x.Item3==1))
.Select(x => x.Item1);
merged.Subscribe(x => Console.WriteLine("Merged "+x));
更新 固定 OP 代码：
public class DiscriminatedUnion
{
public static IObservable<TInput2> Process<TInput1, TInput2>(
IObservable<TInput1> input1,
IObservable<TInput2> input2)
{
var merged =
input2.Select(s2 => Tuple.Create(2, (object)s2))
.Merge(input1.Select(s1 => Tuple.Create(1, (object)s1)))
.Scan(Tuple.Create(default(TInput2), new Queue<TInput2>(), 0), (state, val) =>
{
TInput2 next = state.Item1;
if (val.Item1 == 1)
{
if (state.Item2.Count > 0)
{
next = state.Item2.Dequeue();
}
}
else
{
state.Item2.Enqueue((TInput2)val.Item2);
}
return Tuple.Create(next, state.Item2, val.Item1);
})
.Where(x => (!x.Item1.Equals(default(TInput2)) && x.Item3 == 1))
.Select(x => x.Item1);
return merged;
}
}
我喜欢这些Rx谜题。不敢相信一个人这样做会得到报酬。所以我想出了一个稍微不同的方法。我认为这里的竞争条件存在一些弱点，但我很好奇你的想法以及如何消除这些弱点。
基本思想是将队列视为递归缓冲区，直到超过source1，其中缓冲区被重播到没有第一个元素的队列中。
更新
基于shlomo的观察，需要publish().refcount()，我更新了代码并将解决方案转换为扩展"ReregulationatedQueue"。请参阅下面的代码。Input2 是通过队列进行调节的源，Input1 是调节信号。
public static class RxHelpers
{
public static IObservable<TInput2> RegulatedQueue<TInput1, TInput2>(this IObservable<TInput2> input2,
IObservable<TInput1> input1
)
{
return Observable.Using(() => new Subject<TInput2>(),
queue =>
{
input2.Subscribe(queue);
return queue
.Buffer(() => input1)
.Do(l => { foreach (var n in l.Skip(l.Count > 1 ? 1 : 0)) queue.OnNext(n); })
.Where(l => l.Count > 0)
.Select(l => l.First()).
Publish().
RefCount();
});
}
}
class Program
{
static void Main(string[] args)
{
Random r = new Random();
var source1 = Observable.Interval(TimeSpan.FromSeconds(1)).Publish().RefCount();
var source2 = Observable.Interval(TimeSpan.FromMilliseconds(2000)).Select(x => Enumerable.Range(1, 3).Select(y => r.Next(200)).ToObservable()).SelectMany(x => x).Publish().RefCount();
source1.Subscribe(x => Console.WriteLine("Source1 " + x));
source2.Subscribe(x => Console.WriteLine("Source2 " + x));
var merged = source2.RegulatedQueue(source1);
merged.Subscribe(x => Console.WriteLine("Merged1 " + x));
merged.Subscribe(x => Console.WriteLine("Merged2 " + x));
Console.ReadKey();
}
}
过时
static void Main(string[] args)
{
Random r = new Random();
var source1 = Observable.Interval(TimeSpan.FromSeconds(1)).Publish().RefCount();
var source2 = Observable.Interval(TimeSpan.FromMilliseconds(7000)).Select(x => Enumerable.Range(1, 3).Select(y => r.Next(200)).ToObservable()).SelectMany(x => x).Publish().RefCount();
source1.Subscribe(x => Console.WriteLine("Source1 " + x));
source2.Subscribe(x => Console.WriteLine("Source2 " + x));
//THIS BIT
Subject<int> queue = new Subject<int>();
source2.Subscribe(queue);
var merged=queue
.Buffer(() => source1)
.Do(l => { foreach (var n in l.Skip(l.Count > 1 ? 1 : 0)) queue.OnNext(n); })
.Where(l=>l.Count > 0)
.Select(l => l.First());
merged.Subscribe(x => Console.WriteLine("Merged "+x));
Console.ReadKey();
}
测试代码：
var scheduler = new TestScheduler();
var input1 = scheduler.CreateColdObservable<int>(
ReactiveTest.OnNext(1000.Ms(), 1),
ReactiveTest.OnNext(2000.Ms(), 2),
ReactiveTest.OnNext(3000.Ms(), 3),
ReactiveTest.OnNext(4000.Ms(), 4),
ReactiveTest.OnNext(5000.Ms(), 5),
ReactiveTest.OnNext(6000.Ms(), 6),
ReactiveTest.OnNext(7000.Ms(), 7)
);
var input2 = scheduler.CreateColdObservable<string>(
ReactiveTest.OnNext(1400.Ms(), "a"),
ReactiveTest.OnNext(1500.Ms(), "b"),
ReactiveTest.OnNext(1600.Ms(), "c"),
ReactiveTest.OnNext(5500.Ms(), "d"),
ReactiveTest.OnNext(5600.Ms(), "e"),
ReactiveTest.OnNext(5700.Ms(), "f")
);
Subject<string> queue = new Subject<string>();
input2.Subscribe(queue);
var result = queue
.Buffer(() => input1)
.Do(l => { foreach (var n in l.Skip(l.Count > 1 ? 1 : 0)) queue.OnNext(n); })
.Where(l => l.Count > 0)
.Select(l => l[0]);
result.Timestamp(scheduler)
.Select(t => $"{t.Timestamp.Ticks} ticks: {t.Value}")
.Dump(); //Linqpad
预期输出：
//14000000 enqueue a
//15000000 enqueue b
//16000000 enqueue c
20000000 ticks: a
30000000 ticks: b
40000000 ticks: c
50000000 ticks: c
//55000000 enqueue d
//56000000 enqueue e
//57000000 enqueue f
60000000 ticks: c //should really be d, but there's no handling for fake-empty ejection
70000000 ticks: d
80000000 ticks: e
90000000 ticks: f
100000000 ticks: f
110000000 ticks: f
120000000 ticks: f
130000000 ticks: f
140000000 ticks: f
...
实际输出：
20000000 ticks: a
30000000 ticks: b
40000000 ticks: c
50000000 ticks: b
60000000 ticks: c
70000000 ticks: b
80000000 ticks: c
90000000 ticks: c
100000000 ticks: b
110000000 ticks: c
120000000 ticks: c
130000000 ticks: b
140000000 ticks: c
150000000 ticks: b
160000000 ticks: c
170000000 ticks: b
180000000 ticks: c
190000000 ticks: c