我有一个 Observable 序列，它以快速爆发的方式产生事件(即:五个事件一个接一个地发生，然后是长时间的延迟，然后是另一个事件的快速爆发，等等)。我想通过在事件之间插入一个短暂的延迟来消除这些爆发。以下图为例:
Raw: --oooo--------------ooooo-----oo----------------ooo| Buffered: --o--o--o--o--------o--o--o--o--o--o--o---------o--o--o|
My current approach is to generate a metronome-like timer via Observable.Interval() that signals when it's ok to pull another event from the raw stream. The problem is that I can't figure out how to then combine that timer with my raw unbuffered observable sequence.
IObservable.Zip() is close to doing what I want, but it only works so long as the raw stream is producing events faster than the timer. As soon as there is a significant lull in the raw stream, the timer builds up a series of unwanted events that then immediately pair up with the next burst of events from the raw stream.
Ideally, I want an IObservable extension method with the following function signature that produces the bevaior I've outlined above. Now, come to my rescue StackOverflow :)
public static IObservable<T> Buffered(this IObservable<T> src, TimeSpan minDelay)
附言。我是 Rx 的新手，如果这是一个非常简单的问题，我深表歉意...
这是我最初的天真和简单的解决方案，但有很多问题:
public static IObservable<T> Buffered<T>(this IObservable<T> source, TimeSpan minDelay)
{
Queue<T> q = new Queue<T>();
source.Subscribe(x => q.Enqueue(x));
return Observable.Interval(minDelay).Where(_ => q.Count > 0).Select(_ => q.Dequeue());
}
第一个明显的问题是，原始源的内部订阅返回的 IDisposable 丢失了，因此无法终止订阅。在此方法返回的 IDisposable 上调用 Dispose 会终止计时器，但不会终止现在不必要地填充队列而没有人从队列中提取事件的底层原始事件馈送。
第二个问题是无法将异常或流结束通知从原始事件流传播到缓冲流 - 在订阅原始源时它们会被忽略。
最后但并非最不重要的一点是，现在我得到了定期唤醒的代码，无论实际上是否有任何工作要做，我宁愿在这个美妙的新 react 世界中避免这种情况。
为了解决在我最初的简单方法中遇到的问题，我写了一个远更复杂的函数，它的行为很像IObservable.Delay()(我使用了.NET Reflector阅读该代码并将其用作我的功能的基础)。不幸的是，许多样板逻辑(例如 AnonymousObservable)在 system.reactive 代码之外无法公开访问，因此我不得不复制并粘贴大量 代码。此解决方案似乎可行，但考虑到它的复杂性，我不太相信它没有错误。
我简直不敢相信没有一种方法可以使用标准 Reactive 扩展的某种组合来完成此操作。我讨厌自己在无谓地重新发明轮子，而我尝试构建的模式似乎是一个相当标准的模式。
最佳答案
这实际上是 A way to push buffered events in even intervals 的副本，但我将在此处包含一个摘要(原文看起来很困惑，因为它考虑了一些备选方案)。
public static IObservable<T> Buffered<T>(this IObservable<T> source, TimeSpan minDelay)
{
return source.Drain(x =>
Observable.Empty<int>()
.Delay(minDelay)
.StartWith(x)
);
}
我对 Drain 的实现与 SelectMany 类似，只是它会先等待先前的输出完成(您可以将其视为 ConactMany，而 SelectMany 更像是 MergeMany)。内置的 Drain 不会以这种方式工作，因此您需要包含以下实现:
public static class ObservableDrainExtensions
{
public static IObservable<TOut> Drain<TSource, TOut>(
this IObservable<TSource> source,
Func<TSource, IObservable<TOut>> selector)
{
return Observable.Defer(() =>
{
BehaviorSubject<Unit> queue = new BehaviorSubject<Unit>(new Unit());
return source
.Zip(queue, (v, q) => v)
.SelectMany(v => selector(v)
.Do(_ => { }, () => queue.OnNext(new Unit()))
);
});
}
}
关于c# - Rx IObservable 缓冲以平滑突发事件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/4505529/
如何在ruby中调用C#dll？ 最佳答案 我能想到几种可能性:为您的DLL编写(或找人编写)一个COM包装器，如果它还没有，则使用Ruby的WIN32OLE库来调用它；看看RubyCLR,其中一位作者是JohnLam，他继续在Microsoft从事IronRuby方面的工作。(估计不会再维护了，可能不支持.Net2.0以上的版本)；正如其他地方已经提到的，看看使用IronRuby，如果这是您的技术选择。有一个主题是here.请注意，最后一篇文章实际上来自JohnLam(看起来像是2009年3月)，他似乎很自在地断言RubyCL
我正在尝试在Ruby中复制Convert.ToBase64String()行为。这是我的C#代码:varsha1=newSHA1CryptoServiceProvider();varpasswordBytes=Encoding.UTF8.GetBytes("password");varpasswordHash=sha1.ComputeHash(passwordBytes);returnConvert.ToBase64String(passwordHash);//returns"W6ph5Mm5Pz8GgiULbPgzG37mj9g="当我在Ruby中尝试同样的事情时，我得到了相同sha
是否有简单的方法来更改默认ISO格式(yyyy-mm-dd)的ActiveAdmin日期过滤器显示格式？ 最佳答案 您可以像这样为日期选择器提供额外的选项，而不是覆盖js:=f.input:my_date,as::datepicker,datepicker_options:{dateFormat:"mm/dd/yy"} 关于ruby-on-rails-事件管理员日期过滤器日期格式自定义，我们在StackOverflow上找到一个类似的问题： https://s
C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.
我正在尝试将以下SQL查询转换为ActiveRecord，它正在融化我的大脑。deletefromtablewhereid有什么想法吗？我想做的是限制表中的行数。所以，我想删除少于最近10个条目的所有内容。编辑:通过结合以下几个答案找到了解决方案。Temperature.where('id这给我留下了最新的10个条目。 最佳答案 从您的SQL来看，您似乎想要从表中删除前10条记录。我相信到目前为止的大多数答案都会如此。这里有两个额外的选择:基于MurifoX的版本:Table.where(:id=>Table.order(:id).
这是我在ActiveAdmin中的自定义页面ActiveAdmin.register_page"Settings"doaction_itemdolink_to('Importprojects','settings/importprojects')endcontentdopara"Text"endcontrollerdodefimportprojectssystem"rakedataspider:import_projects_ninja"para"OK"endendend我想做的是，当我单击“导入项目”按钮时，我想在Controller中执行rake任务。但是我无法访问该方法。可能是什
例如，假设我有一个名为Products的模型，并且在ProductsController中，我有以下代码用于product_listView以显示已排序的产品。@products=Product.order(params[:order_by])让我们想象一下，在product_listView中，用户可以使用下拉菜单按价格、评级、重量等进行排序。数据库中的产品不会经常更改。我很难理解的是，每次用户选择新的order_by过滤器时，rails是否必须查询，或者rails是否能够以某种方式缓存事件记录以在服务器端重新排序？有没有一种方法可以编写它，以便在用户排序时rails不会重新查询结果
我有一个将某些事件写入队列的Rails3应用。现在我想在服务器上创建一个服务，每x秒轮询一次队列，并按计划执行其他任务。除了创建ruby脚本并通过cron作业运行它之外，还有其他稳定的替代方案吗？ 最佳答案 尽管启动基于Rails的持久任务是一种选择，但您可能希望查看更有序的系统，例如delayed_job或Starling管理您的工作量。我建议不要在cron中运行某些东西，因为启动整个Rails堆栈的开销可能很大。每隔几秒运行一次它是不切实际的，因为Rails上的启动时间通常为5-15秒，具体取决于您的硬件。不过，每天这样做几
我有一个帖子属于城市的关系，城市又属于一个州，例如:classPost现在我想找到所有帖子及其所属的城市和州。我编写了以下查询来获取带有城市的帖子，但不知道如何在同一查找器中获取带有城市的相应州:@post=Post.find:all,:include=>[:city]感谢任何帮助。谢谢。 最佳答案 Post.all(:include=>{:city=>:state}) 关于ruby-on-rails-使用Rails事件记录获取二级模型，我们在StackOverflow上找到一个类似的问
我觉得我错过了什么。我正在编写一个rubygem，它允许与事件记录进行交互，作为其主要功能的附加功能。在为其编写测试用例时，我需要能够指定虚拟事件记录模型来测试此功能。如果我可以获得一个事件记录模型的实例，它不需要与数据库的任何连接，可以有关系，所有这些东西，但不需要我在数据库中设置表，那就太棒了。我对测试还很陌生，在Rails测试之外我也很陌生，但似乎我应该能够相当轻松地完成类似的事情，但我什么也没找到。谁能告诉我我错过了什么？我看过工厂、制造商、固定装置，所有这些似乎都想达到目标。人们如何在您只需要AR对象进行测试的地方测试gem？ 最佳答案