RxCpp Reactive Extensions in C++

Question

I have a Win32 console app and i have imported the references to Rx. Intellisense allows me to do this....

using namespace System::Reactive;
using namespace System::Reactive::Concurrency;
using namespace System::Reactive::Disposables;
using namespace System::Reactive::Joins;
using namespace System::Reactive::Linq;
using namespace System::Reactive::PlatformServices;
using namespace System::Reactive::Subjects;
using namespace System::Reactive::Threading;
using namespace System::Reactive::Threading;
using namespace System::Data::Linq;
using namespace System::Xml::Linq;

I then have a number of classes available, such as ISubject/Subject and IObserver/Observer. However there is no IObservable. I am a little alarmed at the lack of documentation for Rx with Cpp. Am i missing any obvious resources?

I've tried Channel9, Google, Stackoverflow and Facebook groups. This is the working C# code that i have made, i'd like to get this working with C++. This function merges all of the data from different observane sources and puts it out as a list.

So matrix one appears from source one, matrix two appears from source two. They are matched by id and are pushed forward together as a list.

public static IObservable<IList<TSource>> MergeById<TSource>(this IObservable<TSource> source, Func<IList<TSource>, TSource> mergeFunc, Func<TSource, int> keySelector, int bufferCount)
{
   return Observable.Create<IList<TSource>>(o =>  
   {
     var buffer = new Dictionary<int, IList<TSource>>();
     return source.Subscribe<TSource>(i =>
          { 
             var index = keySelector(i);
             if (buffer.ContainsKey(index))
             {                  
               buffer[index].Add(i);
             }
             else
             {                                     
               buffer.Add(index, new List<TSource>(){i});                
             }
             if (buffer.Count==bufferCount)
             { 
               o.OnNext(mergeFunc(buffer[index]));    
               buffer.Remove(index);                
             }
          });
     });
}

Any help here would be good. Can't find some of the classes i want and other aspects of the syntax are different. Are there any sources or examples that show how things are done in C++. It could probably be inferred from those.

Here is the original post of the issue.

http://social.msdn.microsoft.com/Forums/en-US/58a25f70-a7b8-498b-ad7a-b57f3e1152da/rxcpp?forum=rx

I have attempted to ask here before, but there was no response. Hoping this will be a bit more fruitful, now i have more information on what i am trying to achieve.

Thank you.

Answer 1

The top level namespace is rxcpp, not System. To get observable, you'll need:

#include "cpprx/rx.hpp"
using namespace rxcpp;

Answer 2

I found this which helps a great deal.

But i am unsure who to then using this, to create an extension method. Is that even possible in C++?

http://social.msdn.microsoft.com/Forums/en-US/a500d5de-8c22-4222-829f-09ccd4c7920e/using-rx-from-ccx-metro-app?forum=rx&prof=required

public static IObservable<IList<TSource>> MergeById<TSource>(this IObservable<TSource> source, Func<IList<TSource>, TSource> mergeFunc, Func<TSource, int> keySelector, int bufferCount)
{
   return Observable.Create<IList<TSource>>(o =>  
   {
     var buffer = new Dictionary<int, IList<TSource>>();
     return source.Subscribe<TSource>(i =>
          { 
             var index = keySelector(i);
             if (buffer.ContainsKey(index))
             {                  
               buffer[index].Add(i);
             }
             else
             {                                     
               buffer.Add(index, new List<TSource>(){i});                
             }
             if (buffer.Count==bufferCount)
             { 
               o.OnNext(mergeFunc(buffer[index]));    
               buffer.Remove(index);                
             }
          });
     });
}

The aim is to get two observable sources, which are subjects, into one observable where they are paired together by their id and push together to the OnNext method.

Regards,

Daniel