The new IObservable/IObserver frameworks in the System.Reactive library coming in .NET 4.0 are very exciting (see this and this link). It may be too early to speculate, but will there also be a (for lack of a better term) IQueryable-like framework built for these new interfaces as well? One particular use case would be to assist in pre...
IObservable<T> and IObserver<T> interfaces are placed directly in the System namespace in .NET Framework 4.0 Beta2. Why not in System.Collections.Generic, like IEnumerable<T>? p.s. Reactive Framework preview from Silverlight Toolkit contains IObserver<T> in the System.Collections.Generic namespace. ...
The Reactive Extensions come with a lot of helper methods for turning existing events and asynchronous operations into observables but how would you implement an IObservable<T> from scratch? IEnumerable has the lovely yield keyword to make it very simple to implement. What is the proper way of implementing IObservable<T>? Do I need to...
After experimenting with IObservables, I've decided to test them for the processing of incoming messages over a message bus. Essentially I get an IObservable<Request> and the Request contains the necessary functions to reply to the response. At a point during processing I have to deserialize the data and convert it from a Request to a C...
I understand the IObservable & IObserver are implementations of the observer pattern and can be used in similiar circumstances to .Net events. I was wondering if there is any relationship to INotifyPropertyChanged? I currently use INotifyPropertyChanged for data binding in winforms & WPF applications and was wondering if I'll be able t...
I'm struggling with my first simple "hello world" RX application. I'm using VS2010 RC, plus the latest RX download. The following is the simple console app; class Program { static void Main(string[] args) { var channel = new MessageChannel() .Where(m => m.process) .Su...
I wondered if it's possible to use a queue (specifically as ConcurrentQueue) as the source of an IObservable? Something like; Queue = new ConcurrentQueue<IMessage>(); var xs = Queue.AsEnumerable().ToObservable(); xs.Subscribe((IMessage msg) => { Console.WriteLine("Msg :" + msg.subject); }); I guess it doesn't ...
In a system I'm currently working on, I have many components which are defined as interfaces and base classes. Each part of the system has some specific points where they interact with other parts of the system. For example, the data readying component readies some data which eventually needs to go to the data processing portion, the...
It's possible to implement INotifyCollectionChanged or other interface like IObservable to enable to bind filtered data from xml file on this file changed ? I see examples with properties or collection, but what with files changes ? I have that code to filter and bind xml data to list box: XmlDocument channelsDoc = new XmlDocument(); c...
I have a IObservable [named rows in the sample below] from Reactive extensions framework and I want to add index numbers to each object it observes. I've tried to implement this using Zip function: rows.Zip(Enumerable.Range(1, int.MaxValue), (row, index) => new { Row = row, Index = index }) .Subscribe(a => ProcessRow(a.Row, a....
Hello, I'm creating multiple asynchronous web requests using IObservables and reactive extensions. So this creates observable for "GET" web request: var tweetObservalue = from request in WebRequestExtensions.CreateWebRequest(outUrl + querystring, method) from response in request.GetResponseAsync()...
Within Async handler I'm creating an IObservable from webrequest which returns a redirect string. I'm subscribing to that observable and calling AsyncResult.CompleteCall() but I'm forced to use Thread.Sleep(100) in order to get it executed. And it doesn't work every time. I'm pretty sure this is not correct. Could you please shine some ...
Update Looks like Jon Skeet was right (big surprise!) and the issue was with my assumption about the Average extension providing a continuous average (it doesn't). For the behavior I'm after, I wrote a simple ContinuousAverage extension method, the implementation of which I am including here for the benefit of others who may want somet...
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...
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...
When implementing IObserver yourself, you know how well you would cope with a situation where OnNext is invoked from different threads, concurrently or sequentially, but what are the expectations of the built in Reactive Extension primitives when it comes to this? Will BufferWithTime, for example, cope with OnNext being invoked from mult...
hi all, Well, MS introduced the IObservable interface as part of the .net FW 4 and i thought "great finally, i must use it !". So i dug deep and read posts and documentation and even implemented the pattern. After doing so i've realized that the basic implementation actually sends all the Observable events to all of its subscribers wi...
I am trying to use Reactive Extensions to write code to handle an asynchronous call where both the initiating method and the completed event are static. I can't use var languageSetsLoaded = Observable .FromEvent<LoadLanguageSetsCompletedEventArgs>( LanguageManager, "LanguageSetsLoaded") as LanguageManager is a static class rathe...
I want to use the .NET 4.0 interface IObserver for a library that needs to support previous versions of the framework as well. I already have conditional compilation which lets me build for each of the framework versions. I don't want to use the Rx Extensions' version of IObserver<T> as that would add an unnecessary dependency to an oth...
I want to use the iobservable pattern to expose a stream of events. The problem is that I'm using unity to create both the observer and the event generator. I would rather not have to new up both of these at application start just so I can start listening for events. Does any one have any suggestions about this? ...