The Subscribing and Unsubscribing section of the Observable Contract is definitive regarding your question. It states: When an Observable issues an OnError or OnComplete notification to its observers, this ends the subscription. Observers do not need to issue an Unsubscribe notification to end subscriptions that are ended by the Observable in this way. This is … Read more
Here’s a reasonably simple Rx way to do what you want. I’ve created an extension method called Pausable that takes a source observable and a second observable of boolean that pauses or resumes the observable. public static IObservable<T> Pausable<T>( this IObservable<T> source, IObservable<bool> pauser) { return Observable.Create<T>(o => { var paused = new SerialDisposable(); var … Read more
You can use Observable.When like so (I’ve added an additional CheckBox for this example): var checkChanged = Observable.FromEvent<EventArgs>(this.checkBox, “CheckedChanged”); var check1Changed = Observable.FromEvent<EventArgs>(this.checkBox1, “CheckedChanged”); var keyPress = Observable.FromEvent<KeyPressEventArgs>(this.textBox, “KeyPress”); var plan1 = checkChanged .And(check1Changed).And(keyPress) .Then((cc, cc1, kp) => “Done.”); var when = Observable.When(plan1); when.Subscribe((result) => this.resultTextBox.Text = result); Also, if you can join other plans … Read more
Ana Betts’ answer works in most scenarios, but if you want to block the stream while waiting for the async function to finish you need something like this: Observable.Interval(TimeSpan.FromSeconds(1)) .Select(l => Observable.FromAsync(asyncMethod)) .Concat() .Subscribe(); Or: Observable.Interval(TimeSpan.FromSeconds(1)) .Select(_ => Observable.Defer(() => asyncMethod().ToObservable())) .Concat() .Subscribe();
RxJava is often misunderstood when it comes to the asynchronous/multithreaded aspects of it. The coding of multithreaded operations is simple, but understanding the abstraction is another thing. A common question about RxJava is how to achieve parallelization, or emitting multiple items concurrently from an Observable. Of course, this definition breaks the Observable Contract which states … Read more
It helped me to understand this by thinking of SubscribeOn as setting the thread being “passed up” the chain and ObserveOn as setting the thread “passed down” the chain. The code below uses named threads which you can play with. Thread.CurrentThread.Name = “Main”; IScheduler thread1 = new NewThreadScheduler(x => new Thread(x) { Name = “Thread1” … Read more
You are on the right track, you can use Select + Concat to flatten out the observable and limit the number of inflight requests (Note: if your task takes longer than the interval time, then they will start to stack up since they can’t execute fast enough): var source = Observable.Interval(TimeSpan.FromMilliseconds(100)) //I assume you are … Read more
I would do this as follows: Window the data stream using the gate stream as a closing selector We can use DistinctUntilChanged on the gate stream to ensure no repeated values We will also force the gate stream to start closed (false) – it won’t affect the output and allows a neat trick Then use … Read more
Summary The first point to make is that you don’t actually need to use Observable.FromEvent to avoid the string literal reference. This version of FromEventPattern will work: var groupedKeyPresses = Observable.FromEventPattern<KeyPressEventHandler, KeyPressEventArgs>( h => KeyPress += h, h => KeyPress -= h) .Select(k => k.EventArgs.KeyChar) .GroupBy(k => k); If you do want to make FromEvent … Read more
Wow, that was not trivial to find… Deep within the bowels of the System.Reactive.Concurrency namespace, there is an internal static class called SchedulerDefaults, which is declared as: internal static class SchedulerDefaults { internal static IScheduler AsyncConversions { get { return DefaultScheduler.Instance; }} internal static IScheduler ConstantTimeOperations { get { return ImmediateScheduler.Instance; }} internal static IScheduler … Read more