如何连接返回可观测值的两个函数？

Question

我编写了一个函数，它返回一对可观察到的QDateTime，如下所示：

rxcpp::observable<std::tuple<QDateTime, QDateTime>> experimentOne(const QDateTimeAxis * const axis 
{
   return rxcpp::observable<>::create<std::tuple<QDateTime, QDateTime>>(
     [axis](rxcpp::subscriber<std::tuple<QDateTime, QDateTime>> s) {

       auto rangeCallback = [s](QDateTime minv, QDateTime maxv) {

          if (s.is_subscribed()) {

              // send to the subscriber
              s.on_next(std::make_tuple<QDateTime, QDateTime>(std::move(minv), std::move(maxv)));
          }

       };

       QObject::connect(axis, &QDateTimeAxis::rangeChanged, rangeCallback);
   }); 
}

因此，通过这个函数，我可以订阅QChart轴上日期范围的更改。

我还编写了另一个函数，给定两个日期，返回一个可观察到的值，值来自sqlite，如下所示

rxcpp::observable<std::tuple<double, double>> Database::getValueRange(const std::string& table, unsigned long start, unsigned long end)
{

   return rxcpp::observable<>::create<std::tuple<double, double>>(
      [this, table, start, end](rxcpp::subscriber<std::tuple<double, double>> s) {

    // get the prepared statement for the query 1, i.e. ohlcv values
    // within a date range
    sqlite3_stmt *stmt = this->m_query3_stms[table].get();

    // bind first parameter, the start timestamp
    int rc = sqlite3_bind_int64(stmt, 1, start);
    checkSqliteCode(rc, m_db.get());

    // bind the second parameter, the end timestamp
    rc = sqlite3_bind_int64(stmt, 2, end);
    checkSqliteCode(rc, m_db.get());

    // step through the query results
    while ( sqlite3_step(stmt)==SQLITE_ROW && s.is_subscribed() ) {

        // extract name values from the current result row
        float minv = sqlite3_column_double(stmt, 0);
        float maxv = sqlite3_column_double(stmt, 1);

        // send to the subscriber
        s.on_next(std::make_tuple<double, double>(minv, maxv));
    }

    // reset the statement for reuse
    sqlite3_reset(stmt);

    // send complete to the subscriber
    s.on_completed();

   });
}

如何将第一个函数(两个日期)的值作为RxCpp中惯用形式的第二个函数的输入？在管道的末尾，我可以根据输入日期订阅来自db的值？

Answer 1

为每一对新的日期值创建新值范围的规范方法是使用map，后面跟着一个扁平运算符。

auto valueRanges = experimentOne(/*params*/).
    map(rxcpp::util::apply_to([](QDateTime d1, QDateTime d2){
      return getValueRange(/*params*/).
          map(rxcpp::util::apply_to([=](double db1, double db2){ 
              return std::make_tuple(d1, d2, db1, db2); 
          }));
    })).
    /* switch_on_next() or merge() or concat() */
    /* this will produce std::tuple< QDateTime, QDateTime, double, double>

• switch_on_next将取消以前的值范围并启动新的值范围。
• merge将尽快生成所有的值范围。
• concat将按顺序一次生成一个值范围。

在值范围在不同线程上运行的情况下，必须将线程安全协调传递给merge，以便安全地交织值范围。

若要选择特定范围，请使用filter()。如果您希望能够将范围拆分成不同的表达式，请首先使用publish()共享范围。

auto hotValueRanges = valueRanges.
    publish().ref_count();

auto aDateRange = hotValueRanges.
    filter(rxcpp::util::apply_to([](QDateTime d1, QDateTime d2, double, double){
        return isADesiredDate(d1, d2);
    })).
    subscribe(/*use the range*/);

auto anotherDateRange = hotValueRanges.
    filter(rxcpp::util::apply_to([](QDateTime d1, QDateTime d2, double, double){
        return isAnotherDesiredDate(d1, d2);
    })).
    subscribe(/*use the range*/);