system.reactive

Get previous element in IObservable without re-evaluating the sequence

In an IObservable sequence (in Reactive Extensions for .NET), I'd like to get the value of the previous and current elements so that I can compare them. I found an example online similar to below which accomplishes the task: sequence.Zip(sequence.Skip(1), (prev, cur) => new { Previous = prev, Current = cur }) It works fine except tha...

Creating a REST client API using Reactive Extensions (Rx)

I'm trying to get my head around the right use cases for Reactive Extensions (Rx). The examples that keeps coming up are UI events (drag and drop, drawing), and suggestions that Rx is suitable for asynchronous applications/operations such as web service calls. I'm working on an application where I need to write a tiny client API for a R...

IConnectableObservables in Rx

Hi there Can someone explain the differences between an Observable and a ConnectableObservable? The Rx Extensions documentation is very sparse and I don't understand in what cases the ConnectableObservable is useful. This class is used in the Replay/Prune methods. ...

Real world examples of Rx

I've been playing around with the Reactive Extension for a little while now, but mostly limited to handling/composing user driven events within a WPF frontend. It's such a powerful, new way of doing async programming, and I'm curious as to what other people are doing with it, and where do you think it might be able to improve the way we...

How can I get an IObservable<T> in Rx from a "non-standard" event?

Here's what I mean. Suppose I'm working with an API that exposes events, but these events do not follow the standard EventHandler or EventHandler<TEventArgs> signature. One event might look like this, for instance: Public Event Update(ByVal sender As BaseSubscription, ByVal e As BaseEvent) Now, typically, if I want to get an IObservab...

System.Interactive: Difference between Memoize() and MemoizeAll()?

In System.Interactive.dll (v1.0.2521.0) from Reactive Extensions, EnumerableEx has both a Memoize method and a MemoizeAll method. The API documentation is identical for both of them: Creates an enumerable that enumerates the original enumerable only once and caches its results. However, these methods are clearly not identical. If ...

Using Rx to synchronize asynchronous events

I want to put Reactive Extensions for .NET (Rx) to good use and would like to get some input on doing some basic tasks. To illustrate what I'm trying to do I have a contrived example where I have an external component with asyncronous events: class Component { public void BeginStart() { ... } public event EventHandler Started; } ...

How to convert a method that takes an OnError and OnCompleted into an Observable

Hi all, There is probably a really easy answer to this but my brain just isn't working. I have a method I need to call in a framework that is not Observable aware, that has the following pattern. client.GetAsync<TResult>( string resource, Action<Exception> onError, Action<TResult> onCompleted); I need to convert thi...

Rx Reactive extensions: Unit testing with FromAsyncPattern

The Reactive Extensions have a sexy little hook to simplify calling async methods: var func = Observable.FromAsyncPattern<InType, OutType>( myWcfService.BeginDoStuff, myWcfService.EndDoStuff); func(inData).ObserveOnDispatcher().Subscribe(x => Foo(x)); I am using this in an WPF project, and it works great at runtime. Unfortunat...

How to use IObservable/IObserver with ConcurrentQueue or ConcurrentStack

I realized that when I am trying to process items in a concurrent queue using multiple threads while multiple threads can be putting items into it, the ideal solution would be to use the Reactive Extensions with the Concurrent data structures. My original question is at: http://stackoverflow.com/questions/2997797/while-using-concurrent...

Reactive Extensions memory usage

I have the following code in a WPF application using Reactive Extensions for .NET: public MainWindow() { InitializeComponent(); var leftButtonDown = Observable.FromEvent<MouseButtonEventArgs>(this, "MouseLeftButtonDown"); var leftButtonUp = Observable.FromEvent<MouseButtonEventArgs>(this, "MouseLeftButtonUp"); var move...

How come the linq syntax work over the IQueryable interface in Rx Framework

I've started looking over the reactive framework. Very nice stuff. But while looking at code samples it got me confused. The linq syntax works with the IQueryable. I thought that linq only works with IEnumerable. On what does the C# compiler bases it's linq to extension methods conversions? Does it require a set of methods with a specifi...

How to combine intermediate events in a Reactive Framework stream?

I'm still fairly new to Rx and am having a hard time figuring out how to express this (seemingly) simple subscription. I'm looking for something like this: Start: InTransaction.Where(inTransaction => inTransaction) If: ItemChanged or On FlagChanged, let Changed = true End: InTransaction.Where(inTransaction => !inTransaction) All of t...

Help with using the Subject class in Reactive Extensions library

I am trying to use a pair of Subject classes to fire away 2 sets of event sequences. The application is a drawing application, where one subject fires an onNext when the user clicks and the other subject fires an OnNext when the user doubleclicks. I have written GetClick & GetDoubleClick methods that return an observable for the above ca...

Using Reactive Extensions (Rx) for socket programming practical?

What is the most succint way of writing the GetMessages function with Rx: static void Main() { Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); var messages = GetMessages(socket, IPAddress.Loopback, 4000); messages.Subscribe(x => Console.WriteLine(x)); Console.ReadKey(); } s...

Killer benefit of Reactive Extensions?

What is the killer benefit of Reactive Extensions (for either .NET or JavaScript)? Why should a developer learn and use them? ...

How to throttle event stream using RX?

I want to effectively throttle an event stream, so that my delegate is called when the first event is received but then not for 1 second if subsequent events are received. After expiry of that timeout (1 second), if a subsequent event was received I want my delegate to be called. Is there a simple way to do this using Reactive Extension...

Reactive Framework (RX) and dealing with events Asynchronously

So I'm just playing around with RX and learning it. I started playing with Events, and wanted to know how to subscribe to events, and process the results in batches asynchronously. Allow me to explain with code: Simple class that raises events: public class EventRaisingClass { public event EventHandler<SomeEventArgs> EventOccured; ...

Multiple Subscriptions from one Observer in Rx

I am trying to refine the filter in my Observer over time, and I was wondering what they best way of doing this is. e.g. myObservable.Where(x=>x.Color=="red").Subscribe(); and then myObservable.Where(x=>x.Color=="blue").Subscribe(); and then merge the two into one stream so that OnNext() is called on Red OR Blue observables. Maybe it...

Where do I find Observable in VS2010?

I have VS2010 RTM installed and I want to add a reference to the Reactive Framework, I've looked for system.CoreEx and system.Reactive and they are not there. What am I doing wrong? ...