c# - 使用 RxCpp 构造 Observer/Observable 模式

Question

我正在尝试observer/observable在Rx-cpp. 这是一个非常有趣的教程，Rx.Net介绍了如何做到这一点。

在此C#示例中interfaces，我们必须覆盖特定的内容：

public interface IObserver<in T>
{
    void OnCompleted();
    void OnError(Exception error);
    void OnNext(T value);
}


public interface IObservable<out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}

据我了解，在Rx-cpp没有这样的方便。那么，是否可以为我提供一些类似于上面的标题示例（ myObservable.h/ ），我可以将其用作定义相同通信模式的指导？ myObserver.hinterfaces

任何帮助都非常感谢，谢谢！

编辑 1： 感谢@zentrunix，我正在尝试进行面向课堂的交流。到目前为止，我有以下可观察模式的代码。我想要的是定义一个观察者列表，这些观察者将我附加到可观察对象中，并且当OnNext调用这些观察者时，应该通知这些观察者。但是，有缺失的部分。

当一个函数被调用时，我怎么能subscribe()在那些观察者（）上。Rx::subscribers<int>myObservable::Subscribe()
还有我怎么能unsubscribe()。
最后，o.subscribe(onNext, onEnd);多个onNext观察者的对应关系如何？是否有可能构造一个相应的myObserver类？（再次受到这里的启发）

抱歉问，但这样的组织有意义吗？到目前为止，我一直在使用本教程中提供的架构，这就是我痴迷于这项任务的原因。我发现这是参与RxCpp. 任何意见都受到高度赞赏。^{（再次为我的无知感到抱歉。）}

class myObservable {

private:

std::shared_ptr<std::list<rxcpp::subscriber<int>>> observers;

public:

myObservable() { observers = std::make_shared<std::list<Rx::subscriber<int>>>(); };

Rx::observable<int> Attach(std::shared_ptr<rxcpp::subscriber<int>> out) {

    return Rx::observable<>::create<int>([&, out]() {
        auto it = observers->insert(observers->end(), *out);
        it->add([=]() {
            observers->erase(it);
        });
    });

};

void OnNext(int sendItem) {

    for (Rx::subscriber<int> observer : *observers) {
        (observer).on_next(sendItem);
    }
}

void Disposer(Rx::subscriber<int> out) {

    observers->erase(std::remove(observers->begin(), observers->end(), &out), observers->end());
};
};

score 4 · Accepted Answer

下面是 RxCpp 中的一个非常简单的示例。但是有（至少）一个警告：典型的 RxCpp 代码大量使用 lambda，我非常不喜欢。

我也尝试在互联网上查找文档和教程，但找不到任何内容。我对线程模型的解释特别感兴趣。

如果您愿意浏览代码和 Doxygen 文档，RxCpp GitHub 站点中有很多示例。

#include <iostream>
#include <exception>

#include "rxcpp/rx.hpp"
namespace rx = rxcpp;

static void onNext(int n) { std::cout << "* " << n << "\n"; }
static void onEnd() { std::cout << "* end\n"; }

static void onError(std::exception_ptr ep)
{
  try { std::rethrow_exception(ep); }
  catch (std::exception& e) { std::cout << "* exception " << e.what() << '\n'; }
}

static void observableImpl(rx::subscriber<int> s)
{
  s.on_next(1);
  s.on_next(2);
  s.on_completed();
}

int main()
{
  auto o = rxcpp::observable<>::create<int>(observableImpl);
  std::cout << "*\n";
  o.subscribe(onNext, onEnd);
}

c# - 使用 RxCpp 构造 Observer/Observable 模式

1 回答 1

Related

Reference