analytics_stream_result Class Reference

A streaming result handle for analytics queries. More...

#include <couchbase/analytics_stream_result.hxx>

Classes
struct	end_sentinel
	Sentinel returned by end(). More...
class	iterator
	A single-pass input iterator that synchronously fetches rows one at a time. More...

Public Member Functions
	analytics_stream_result ()=default
	Constructs an empty (no-op) result handle.
	analytics_stream_result (std::shared_ptr< internal_analytics_stream_result > internal)
	Constructs an analytics stream result from an internal result.
void	next (analytics_row_handler &&handler) const
	Fetches the next row asynchronously, invoking the handler when ready.
auto	next () const -> std::future< std::pair< error, std::optional< analytics_row > > >
	Fetches the next row, returning a future.
auto	signature () const -> std::optional< codec::binary >
	Returns the analytics signature captured from the response metadata, if present.
auto	meta_data () const -> std::future< std::pair< error, analytics_meta_data > >
	Returns the analytics metadata.
void	cancel () const
	Cancels the stream and closes the underlying HTTP connection.
auto	begin () const -> iterator
	Returns an iterator to the beginning.
auto	end () const -> end_sentinel
	Returns the end sentinel.

Detailed Description

A streaming result handle for analytics queries.

Rows are fetched one at a time via next(). The stream must be fully drained (or cancel() called) before meta_data() resolves. Only a single next() call may be outstanding at a time.

#include <couchbase/cluster.hxx>
#include <couchbase/codec/tao_json_serializer.hxx>
 
#include <spdlog/fmt/bundled/format.h>
#include <tao/json.hpp>
 
// After the fmt header: these specialize fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/analytics_status.hxx>
#include <couchbase/fmt/error.hxx>
 
#include <cstdint>
#include <string>
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./analytics_stream couchbase://127.0.0.1 Administrator password\n");
    return 1;
  }
 
  const std::string connection_string{ argv[1] };
  const std::string username{ argv[2] };
  const std::string password{ argv[3] };
 
  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;
  }
 
  // Analytics streaming mirrors the query API: the same three-state next(), the same iterator,
  // the same end-of-stream metadata.
  auto [err, result] =
    cluster.analytics_query_stream("SELECT i AS n FROM array_range(0, 2000) AS i ORDER BY i").get();
  if (err) {
    fmt::print("unable to start the streaming analytics query: {}\n", err);
    return 1;
  }
 
  // The row signature comes from the response preamble, so it is available before the stream has
  // been drained (unlike the metadata).
  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() });
  }
 
  std::uint64_t sum{ 0 };
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      fmt::print("streaming analytics query failed mid-stream: {}\n", row_err);
      return 1;
    }
    if (!row) {
      break; // clean end of stream
    }
    const auto value = row->content_as<couchbase::codec::tao_json_serializer, tao::json::value>();
    sum += value.at("n").as<std::uint64_t>();
  }
  fmt::println("sum of the streamed rows: {}", sum);
 
  auto [meta_err, meta] = result.meta_data().get();
  if (meta_err) {
    fmt::print("unable to retrieve the analytics metadata: {}\n", meta_err);
    return 1;
  }
  fmt::println("status={}, rows={}", meta.status(), meta.metrics().result_count());
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./analytics_stream couchbase://127.0.0.1 Administrator password
signature: {"*":"*"}
sum of the streamed rows: 1999000
status=success, rows=2000
 
 */

Note

The future-returning overloads (next(), meta_data()) and the eager iterator block the calling thread until the result is ready. They must not be called from within an SDK completion handler (i.e. the library's I/O thread) — doing so blocks that thread against itself and the stream can never advance. From a completion handler, use the callback next() overload instead.

Copies of a result handle share one underlying stream: they alias the same rows, the same single-outstanding-next() budget, and the same cancel(). Copying does not fork the stream; a next() on any copy consumes the next row for all of them. Use a single handle per stream.

Since

1.4.0

Volatile

Should not be used in production

Constructor & Destructor Documentation

analytics_stream_result() [1/2]

analytics_stream_result ( )

default

Constructs an empty (no-op) result handle.

Since

1.4.0

Internal

Internal interface

analytics_stream_result() [2/2]

analytics_stream_result ( std::shared_ptr< internal_analytics_stream_result > internal )

explicit

Constructs an analytics stream result from an internal result.

Parameters

internal

the internal result handle

Since

1.4.0

Internal

Internal interface

Member Function Documentation

begin()

auto begin ( ) const -> iterator

nodiscard

Returns an iterator to the beginning.

Returns

iterator to the first row

Since

1.4.0

Volatile

Should not be used in production

cancel()

void cancel

(

)

const

Cancels the stream and closes the underlying HTTP connection.

Simply dropping the last handle before the stream is fully drained also tears the stream down, but if a pull is in flight the underlying connection is not released until that pull settles (up to the inter-read idle timeout). Call cancel() explicitly for prompt, deterministic teardown of the connection and its timers.

Since

1.4.0

Volatile

Should not be used in production

end()

auto end ( ) const -> end_sentinel

nodiscard

Returns the end sentinel.

Returns

sentinel that an iterator compares equal to once the stream is drained

Since

1.4.0

Volatile

Should not be used in production

meta_data()

auto meta_data ( ) const -> std::future< std::pair< error, analytics_meta_data > >

nodiscard

Returns the analytics metadata.

The returned future resolves only after the stream has been fully drained (all rows consumed or the stream cancelled). May be called more than once; each call returns its own future and all of them resolve together once the metadata becomes available.

Returns

future carrying (error, analytics_meta_data)

Since

1.4.0

Volatile

Should not be used in production

next() [1/2]

auto next ( ) const -> std::future< std::pair< error, std::optional< analytics_row > > >

nodiscard

Fetches the next row, returning a future.

Only one outstanding call is allowed at a time.

Returns

future object that carries the result of the operation

Since

1.4.0

Volatile

Should not be used in production

next() [2/2]

void next

(

analytics_row_handler &&

handler

)

const

Fetches the next row asynchronously, invoking the handler when ready.

The handler receives ({}, row) for a real row, ({}, {}) at clean end-of-stream, or (error, {}) if the stream ended with an error.

Only one outstanding call is allowed at a time.

Parameters

handler

callable that implements analytics_row_handler

Since

1.4.0

Volatile

Should not be used in production

signature()

auto signature ( ) const -> std::optional< codec::binary >

nodiscard

Returns the analytics signature captured from the response metadata, if present.

The signature is part of the response preamble, so — unlike meta_data() — it is available as soon as the stream has started, without draining it first:

#include <couchbase/cluster.hxx>
#include <couchbase/codec/tao_json_serializer.hxx>
 
#include <spdlog/fmt/bundled/format.h>
#include <tao/json.hpp>
 
// After the fmt header: these specialize fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/analytics_status.hxx>
#include <couchbase/fmt/error.hxx>
 
#include <cstdint>
#include <string>
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./analytics_stream couchbase://127.0.0.1 Administrator password\n");
    return 1;
  }
 
  const std::string connection_string{ argv[1] };
  const std::string username{ argv[2] };
  const std::string password{ argv[3] };
 
  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;
  }
 
  // Analytics streaming mirrors the query API: the same three-state next(), the same iterator,
  // the same end-of-stream metadata.
  auto [err, result] =
    cluster.analytics_query_stream("SELECT i AS n FROM array_range(0, 2000) AS i ORDER BY i").get();
  if (err) {
    fmt::print("unable to start the streaming analytics query: {}\n", err);
    return 1;
  }
 
  // The row signature comes from the response preamble, so it is available before the stream has
  // been drained (unlike the metadata).
  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() });
  }
 
  std::uint64_t sum{ 0 };
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      fmt::print("streaming analytics query failed mid-stream: {}\n", row_err);
      return 1;
    }
    if (!row) {
      break; // clean end of stream
    }
    const auto value = row->content_as<couchbase::codec::tao_json_serializer, tao::json::value>();
    sum += value.at("n").as<std::uint64_t>();
  }
  fmt::println("sum of the streamed rows: {}", sum);
 
  auto [meta_err, meta] = result.meta_data().get();
  if (meta_err) {
    fmt::print("unable to retrieve the analytics metadata: {}\n", meta_err);
    return 1;
  }
  fmt::println("status={}, rows={}", meta.status(), meta.metrics().result_count());
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./analytics_stream couchbase://127.0.0.1 Administrator password
signature: {"*":"*"}
sum of the streamed rows: 1999000
status=success, rows=2000
 
 */

Returns

optional binary JSON signature

Since

1.4.0

Volatile

Should not be used in production

---

The documentation for this class was generated from the following file:

• couchbase/analytics_stream_result.hxx