What's the IO you're reading? A socket?
(Which I do anticipate since my consumer is sometimes a single-threaded event loop)
but only if that scheduler is applied after the producer
Not sure what you mean by that, in fact
What I meant here was that even when I specifically ask the ThreadPoolScheduler to schedule events, OnNext() continues to block. Eg this blocks -
return Observable.Create<int>(o =>
{
return ThreadPoolScheduler.Instance.Schedule(1, (ii, recurse) =>
{
Thread.Sleep(TimeSpan.FromSeconds(1));
o.OnNext(ii);
recurse(ii + 1);
});
});
Here's a full example of
OnNext() blocking causing the loss of clock ticks, which I'd like to avoid. Uncommenting the ObserveOn() call is the only way I've found to solve it, but I'd love to know if there is a more nuanced approach I am missing:using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading;
namespace rx
{
internal static class Program
{
private static void Main()
{
using var subscription = TickProducer()
//.ObserveOn(ThreadPoolScheduler.Instance)
.Select(Transform)
.Subscribe(Consume);
Thread.Sleep(10000);
}
private static IObservable<int> TickProducer()
{
return Observable.Create<int>(o =>
{
var stopping = false;
var thread = new Thread(() =>
{
var started = DateTime.UtcNow;
var lastTick = 0;
while (!stopping)
{
var seconds = (int) DateTime.UtcNow.Subtract(started).TotalSeconds;
if (seconds != lastTick)
{
o.OnNext(lastTick = seconds);
}
}
});
thread.Start();
return () =>
{
stopping = true;
thread.Join();
};
});
}
private static int Transform(int counter)
{
Thread.Sleep(TimeSpan.FromSeconds(0.5));
return -counter;
}
private static void Consume(int counter)
{
Console.WriteLine(counter);
if (new Random().Next(3) == 0)
Thread.Sleep(TimeSpan.FromSeconds(2)); // Occasional delay
}
}
}
Expected output is -1, -2, -3, ... with every counting number, no skipped numbers.
Otherwise if you dont complete, i do think your subscriptions will eventually be GC'd, but there might be some delay in running this cleanup.
Maybe Window(TimeSpan) operator can do this for you… http://introtorx.com/Content/v1.0.10621.0/17_SequencesOfCoincidence.html#Window @mgnslndh
I want to do some calculation when one of many observable changes. This can be done by project (Select) each observable to a dummy value and merging it all together and finally calling select-operator with desired calculation.
Question: Is there an easy way to do this instead of projecting all observable to dummy value and merging it together?
Question: Is there an easy way to do this instead of projecting all observable to dummy value and merging it together?
To combine observables into an array of observables you can use https://rxjs.dev/api/index/function/combineLatest. It requires all observablea to send at least 1 value before it continues though
Since this is the rx.net chat, better to link to the general documentation: http://reactivex.io/documentation/operators/combinelatest.html
On System.Reactive v.5.0.0 I was experiencing duplicate timestamped (!) values after DistinctUntilChanged, so I put some side effect code with a semicolon in the screenshot above. The breakpoint at the semicolon has been hit. How can that be?
The UtcTicks values are indeed the same in the "marble" and in the memorized last value.
It seems like
It seems like
DistinctUntilChanged() is not working as expected, I am obviously missing something here
What's worse is that even replacing
DistinctUntilChanged() with Distinct() does not help, which surprises me even more
var message = new MessageRecivedArgs() { Data = buffer };
// fire message event args
MessageRecivedSubject.OnNext(message);MessageRecivedArgs is disposible object that consumes large memory
how i can dispose it after all OnNext() done? even async listiners
I've since figured out how to get the desired behavior, but it feels very odd to me that
Timeout documentation states timeout will issue an error notification if a particular period of time elapses without any emitted items. This doesn't seem to be the case though, it seems like no matter what the observable must be completed in order for this to be registered. I'm just wondering why that bit isn't documented 🤔
