Represents a single row of a streaming query result. More...

#include <couchbase/query_row.hxx>

Public Member Functions
	query_row ()=default
	query_row (codec::binary content)
	Constructs a row from raw JSON bytes.
template<typename Serializer = codec::tao_json_serializer, typename Document = typename Serializer::document_type, std::enable_if_t< codec::is_serializer_v< Serializer >, bool > = true>
auto	content_as () const -> Document
	Decodes the row content into the requested document type using the given serializer.
auto	content_as_binary () const -> const codec::binary &
	Returns the raw binary content of this row.

Detailed Description

Represents a single row of a streaming query result.

The raw JSON bytes can be decoded via content_as().

Since: 1.4.0

Volatile: Should not be used in production

Constructor & Destructor Documentation

◆ query_row() [1/2]

query_row ( )

default

Since: 1.4.0

Volatile: Should not be used in production

◆ query_row() [2/2]

query_row ( codec::binary content )

inlineexplicit

Constructs a row from raw JSON bytes.

Parameters

content raw JSON bytes for this row

Since: 1.4.0

Volatile: Should not be used in production

Member Function Documentation

◆ content_as()

template<typename Serializer = codec::tao_json_serializer, typename Document = typename Serializer::document_type, std::enable_if_t< codec::is_serializer_v< Serializer >, bool > = true>

auto content_as ( ) const -> Document

inlinenodiscard

Decodes the row content into the requested document type using the given serializer.

Only call this on a row obtained from a successful next() (falsy error, engaged optional). The end-of-stream and error sentinels carry an empty query_row, and decoding empty/malformed content throws the serializer's parse exception — gate on the three-state next() contract (check the error, then the optional) before decoding.

Any type with a serializer-compatible specialization works as the document type:

// A document type the streaming rows are decoded into. Any type with a
// tao::json::traits specialization works with query_row::content_as() /
// analytics_row::content_as().
#include <string>
 
struct product {
  std::string id{};
  std::string name{};
  double price{};
};
 
template<>
struct tao::json::traits<product> {
  template<template<typename...> class Traits>
  static auto as(const tao::json::basic_value<Traits>& v) -> product
  {
    const auto& object = v.get_object();
    return {
      object.at("id").template as<std::string>(),
      object.at("name").template as<std::string>(),
      object.at("price").template as<double>(),
    };
  }
};

Usage:

#include <couchbase/cluster.hxx>
#include <couchbase/codec/tao_json_serializer.hxx>
 
#include <spdlog/fmt/bundled/chrono.h>
#include <spdlog/fmt/bundled/format.h>
#include <tao/json.hpp>
 
// After the fmt headers: these specialize fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
#include <couchbase/fmt/query_status.hxx>
 
#include <chrono>
#include <string>
 
int
main(int argc, const char* argv[])
{
  if (argc != 5) {
    fmt::print("USAGE: ./query_stream couchbase://127.0.0.1 Administrator password default\n");
    return 1;
  }
 
  const std::string connection_string{ argv[1] }; // "couchbase://127.0.0.1"
  const std::string username{ argv[2] };          // "Administrator"
  const std::string password{ argv[3] };          // "password"
  const std::string bucket_name{ argv[4] };       // "default"
 
  auto [connect_err, cluster] =
    couchbase::cluster::connect(connection_string, couchbase::cluster_options(username, password))
      .get();
  if (connect_err) {
    fmt::print("unable to connect to the cluster: {}\n", connect_err);
    return 1;
  }
 
  const auto statement = fmt::format(R"(
      SELECT META(d).id AS id, d.name AS name, d.price AS price
      FROM `{}` AS d
      WHERE d.type = $1
      ORDER BY d.price DESC
  )",
                                     bucket_name);
  const auto options = couchbase::query_options{}
                         .positional_parameters(std::string{ "streaming-example" })
                         .scan_consistency(couchbase::query_scan_consistency::request_plus)
                         .metrics(true); // off by default; the server then omits meta.metrics()
 
  // [1] Start the stream. The future resolves as soon as the response preamble has been parsed --
  //     long before the whole result has been transferred. Rows are then pulled on demand, and the
  //     SDK pauses reading the socket whenever the rows it has already buffered exceed an internal
  //     high-water mark, so the memory it holds does not grow with the size of the result.
  auto [err, result] = cluster.query_stream(statement, options).get();
  if (err) {
    fmt::print("unable to start the streaming query: {}\n", err);
    return 1;
  }
 
  // [2] The row signature is part of that preamble, so unlike the metadata it is available
  //     immediately, without draining the stream first.
  if (const auto signature = result.signature(); signature) {
    // codec::binary is a byte vector; reinterpret to chars to print it as the JSON text it is.
    fmt::println(
      "signature: {}",
      std::string{ reinterpret_cast<const char*>(signature->data()), signature->size() });
  }
 
  // [3] next() is a three-state contract:
  //       (falsy error, engaged row) -- a data row
  //       (falsy error, empty)       -- clean end of stream
  //       (truthy error, empty)      -- the stream terminated with an error
  //     Only decode the row in the first case; the two terminal states carry an empty row.
  //     Only one next() may be outstanding at a time.
  fmt::println("{:<24} {:<12} {:>8}", "ID", "NAME", "PRICE");
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      fmt::print("streaming query failed mid-stream: {}\n", row_err);
      return 1;
    }
    if (!row) {
      break; // clean end of stream
    }
    const auto p = row->content_as<couchbase::codec::tao_json_serializer, product>();
    fmt::println("{:<24} {:<12} {:>8.2f}", p.id, p.name, p.price);
  }
 
  // [4] The metadata resolves only once the stream has been drained (or cancelled), because the
  //     server sends it after the last row. It may be requested more than once.
  auto [meta_err, meta] = result.meta_data().get();
  if (meta_err) {
    fmt::print("unable to retrieve the query metadata: {}\n", meta_err);
    return 1;
  }
  fmt::println("status={}, request_id={}", meta.status(), meta.request_id());
  if (const auto& metrics = meta.metrics(); metrics) {
    fmt::println("rows={}, elapsed={}",
                 metrics->result_count(),
                 std::chrono::duration_cast<std::chrono::milliseconds>(metrics->elapsed_time()));
  }
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream couchbase://127.0.0.1 Administrator password default
signature: {"id":"json","name":"json","price":"json"}
ID                       NAME            PRICE
streaming-example-9      widget-9        19.00
streaming-example-8      widget-8        18.00
streaming-example-7      widget-7        17.00
streaming-example-6      widget-6        16.00
streaming-example-5      widget-5        15.00
streaming-example-4      widget-4        14.00
streaming-example-3      widget-3        13.00
streaming-example-2      widget-2        12.00
streaming-example-1      widget-1        11.00
streaming-example-0      widget-0        10.00
status=success, request_id=2b6619e1-b34e-47d9-b1f7-3fa1d5fbe4b4
rows=10, elapsed=2ms
 
 */

Template Parameters

Serializer	the serializer to use (defaults to tao_json_serializer)
Document	the document type to decode into (defaults to the serializer's document_type)

Returns: the decoded document

Exceptions

the	serializer's deserialization exception if the content is empty or not valid JSON

Since: 1.4.0

Volatile: Should not be used in production

◆ content_as_binary()

auto content_as_binary ( ) const -> const codec::binary &