是否有 Observable.Interval 的“热"等价物
如果我执行以下操作: var obs =可观察的.Interval(TimeSpan.FromSeconds(1)).Select(x => "A" + x.ToString());观测.Subscribe(x => Console.WriteLine("从第一个:" + x.ToString()));可观察的.Timer(TimeSpan.FromSeconds(3)).SelectMany(
..
system.reactive相关内容
IObservable 按下的键
所以我可以试验 Reactive Extensions，我想创建一个由用户按下的键的 IObservable.我该怎么做? 这是一个 C# 控制台应用程序 解决方案 试试这个以获得可观察的读取键序列: IObservable键 =可观察的.Defer(() =>可观察的.Start(() =>Console.ReadKey())).重复(); 我对此进行了测试，效果很好.
..
Rx:用于从 Observable 流中获取第一个和最新值的运算符
对于基于 Rx 的更改跟踪解决方案，我需要一个操作员，它可以为我提供可观察序列中的第一个和最新的项目. 我将如何编写生成以下弹珠图的 Rx 运算符(注意:括号仅用于排列项目...我不确定如何最好地在文本中表示这一点): xs:---[a ]---[b ]-----[c ]-----[d ]---------|期望:---[a,a]---[a,b]-----[a,c]-----[a,d]-
..
.PublishLast() 的用例(以前是 Prune)
在我看来，我对 RX 函数有很好的“感觉"——我使用了其中的许多函数，或者可以想象其他函数是如何有用的——但我找不到 .Prune 函数的位置.我知道这是到 AsyncSubject 的多播，但这在实际场景中如何有用? 编辑:Richard 说 WebRequest 是 Prune() 的一个很好的候选者.我仍然不明白如何.让我们举个例子 - 我想将传入的 uri 转换为图像: publ
..
学习 Rx:如何在 .Window 的输出上使用 .Scan 以获得可观察的布尔值序列
我有一个像这样的真假值序列 varalternatingTrueFalse = Observable.Generate(真的，_ =>真的，x =>！X，x =>X，_ =>TimeSpan.FromMilliseconds(new Random().Next(2000))).Take(20).Publish();交替TrueFalse.Connect();var 缓冲 = 交替真假.Buff
..
一次强制执行一个异步可观察对象
我正在尝试使用 Observable.FromAsync 将一些 TPL 异步集成到更大的 Rx 链中，就像这个小例子一样: 使用系统；使用 System.Reactive.Linq;使用 System.Threading.Tasks;命名空间 rxtest{课程计划{static void Main(string[] args){MainAsync().Wait();}静态异步任务 MainA
..
如何获得 IObservable 以在订阅时推送最新值
通常，当您订阅某个值的更改时，您也有兴趣了解初始值.我希望我的 IObservable 缓存最新(或初始)值并将该值推送到订阅中. 当使用普通事件时，我经常会得到看起来像这样的代码 x.SomeEvent += SomeEventHandler;SomeEventHandler(x, EventArgs.Empty); 使用 IObservable 我希望用推动初始值的东西包装事件.如果
..
我可以通过匹配键将两个序列配对吗?
假设序列 1 访问网络以检索站点 1、2、3、4、5 的内容(但将以不可预测的顺序返回). 序列 2 将转到数据库以检索有关这些相同记录 1、2、3、4、5 的上下文(但出于本示例的目的，将以不可预测的顺序返回). 是否有一种 Rx 扩展方法可以将这些组合成一个序列当两个序列中的每个匹配对都准备就绪时?即，如果第一个序列以 4,2,3,5,1 的顺序返回，而第二个序列以 1,4,3,2
..
Rx 主题中观察者的数量
使用 Rx，获取 Subject 中当前观察者数量的最佳方法是什么? 我有一个场景，我想发布一条消息，但前提是有观察者.如果没有观察者，我需要做点别的. 为了解决这个问题，我所做的是创建了自己的 ISubject 实现并公开了内部 IObserver 集合的计数.我确信一定有一种开箱即用的方式来做到这一点，我只是不完全熟悉 Rx 所提供的东西. 谢谢！ 解决方案 您应该尽
..
不停止序列的反应性扩展超时?
我正在尝试制作一个 IObservable 返回 true 如果在过去 5 秒内收到了 UDP 消息并且发生超时，则返回 true返回假. 到目前为止，我有这个: public IObservable获取UDP消息(IPEndPoint localEP){var udp = BaseComms.UDPBaseStringListener(localEP).Where(msg => msg.
..
Rx - 取消订阅事件
我有一个 INotifyPropertyChanged 对象，Foo.我使用 Rx 的 FromEvent 方法将 Foo 变成可观察的事件流: var myFoo = new Foo();var eventStream = Observable.FromEvent(myFoo, "PropertyChanged"); 现在我想监听某个特定属性的变化，如果 .Progress == 100，取
..
Rx 框架:在不中断原始可观察序列的情况下执行超时操作
给定一个可观察的源，通过轮询低级设备的(a)状态生成... //可观察的源元代码:IObservable源 = Observable.Interval(TimeSpan.FromSeconds(0.5)).Select(tick => new DeviceState(_device.ReadValue())).DistinctUntilChanged(); ... 以及更新 UI 的消费者..
..
IObservable.ToTask方法返回等待激活的任务
为什么 task 永远等待?: var task = Observable.FromEventPattern(通信器，“PushMessageRecieved").Where(i => i.EventArgs.GetRequestFromReceivedMessage().Name == requestName).Select(i => i.EventArgs).RunAsync(System.
..
为什么这个 Observable.Generate 重载会导致内存泄漏?[使用时间跨度 <15ms]
以下 Rx.NET 代码将在我的机器上大约 10 秒后使用大约 500 MB 的内存. var 流 =Observable.Range(0, 10000).SelectMany(i => Observable.Generate(0,j＝＞真的，j＝＞j + 1,j＝＞新 { N = j },j＝＞TimeSpan.FromMilliseconds(1)));流订阅()； 如果我在没有 Func
..
如何在 Windows Phone 7 中取消基于 Rx 的 Web 请求
在我的应用程序中，我使用基于 Rx 的 Web 请求从服务器获取数据.假设如果我想取消请求，我需要做什么? 解决方案 在 Subscribe 之前尝试使用扩展方法 TakeUntil(someEvent)，例如，其中 SomeEvent 可以是从按钮单击事件触发的 Observable.FromEventPattern. 告诉我们， 问候，
..
使用 Reactive Extensions 测试 Observables 时如何为 Task.Run 使用虚拟时间
我想测试以下功能 ///项目到达时通过 fn 异步处理.然而///如果一个项目在最后一个异步操作之前到达///完成然后传递给 fn 的取消令牌将是///触发以尽力取消任务///道路.公共静态 IObservableSelectWithCancellation( 这个 IObservable 这个，Funcfn){返回这个.Select
..
WPF 绑定:限制绑定列表的更新
我正在将 xamdatagrid 绑定到一个列表.但是由于列表 GUI 上的更新太多，因此卡住了.如何停止这些更新并在间隔(例如 500 毫秒)后刷新网格.将响应式扩展的节流方法有用吗? 解决方案 您可以使用 DeferRefresh 推迟绑定更新，直到您完成对集合的修改: using (collection.DeferRefresh()){//对集合进行更改...} 如果集合是实时更
..
如何制作 IObservable从控制台输入
我尝试编写控制台可观察对象，如下例所示，但它不起作用.订阅存在一些问题.如何解决这些问题? 静态类程序{静态异步任务 Main(string[] args){//var observable = Observable.Interval(TimeSpan.FromMilliseconds(1000)).Publish().RefCount();//有效//var observable = From
..
如何合并嵌套的 observable IObservable>并发性和缓冲容量有限?
我注意到 Rx Merge 运算符接受一个可选的 maxConcurrent 参数.这可用于通过同时订阅有限数量的子序列来限制最大并发.当新子序列的推送速度低于订阅子序列的完成速度时，它可以完美地工作，但是当新子序列的推送速度比这个速度快时，它就会出现问题.发生的情况是子序列被缓冲在一个内部缓冲区中，其大小永远增加，而且当前订阅的子序列变得越来越老.下面是这个问题的演示: await Obser
..
将轮询 Web 服务转换为 RX
给定: 公共部分类天气{私有只读 DispatcherTimer _timer = new DispatcherTimer();私有只读 IWeatherDataProvider _weatherDataProvider;公共天气(IWeatherDataProvider weatherDataProvider){初始化组件();_weatherDataProvider = 天气数据提供者；已加
..