Реактивные расширения RxCpp в C++

Question

У меня есть консольное приложение Win32, и я импортировал ссылки на Rx. Intellisense позволяет мне сделать это....

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;

Затем у меня есть несколько доступных классов, таких как ISubject/Subject и IObserver/Observer. Однако не существует IObservable. Я немного встревожен отсутствием документации для Rx с Cpp. Я пропускаю какие-либо очевидные ресурсы?

Я пробовал группы Channel9, Google, Stackru и Facebook. Это рабочий код C#, который я сделал, я бы хотел, чтобы это работало с C++. Эта функция объединяет все данные из разных источников и выводит их в виде списка.

Таким образом, матрица 1 появляется из источника 1, матрица 2 появляется из источника 2. Они совпадают по id и выдвигаются вместе в виде списка.

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);                
             }
          });
     });
}

Любая помощь здесь будет хорошей. Не могу найти некоторые классы, которые я хочу, и другие аспекты синтаксиса отличаются. Есть ли источники или примеры, которые показывают, как все делается в C++. Это, вероятно, может быть выведено из тех.

Вот оригинальный пост выпуска.

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

Я пытался спросить здесь раньше, но ответа не было. Надеюсь, что это будет немного более плодотворно, теперь у меня есть больше информации о том, чего я пытаюсь достичь.

Спасибо.

Answer 1

Пространство имен верхнего уровня - rxcpp, а не System. Чтобы стать заметным, вам понадобится:

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

Answer 2

Я нашел это, что очень помогает.

Но я не уверен, кто тогда использует это, чтобы создать метод расширения. Это вообще возможно в 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);                
             }
          });
     });
}

Цель состоит в том, чтобы объединить два наблюдаемых источника, которые являются субъектами, в одну наблюдаемую, где они соединяются вместе по их идентификатору и объединяются в метод OnNext.

С Уважением,

Даниил