cluster Class Reference

The cluster is the main entry point when connecting to a Couchbase cluster. More...

#include <couchbase/cluster.hxx>

Public Member Functions
	cluster ()=default
	cluster (const cluster &other)=default
	cluster (cluster &&other)=default
	~cluster ()=default
auto	operator= (const cluster &other) -> cluster &=default
auto	operator= (cluster &&other) -> cluster &=default
void	notify_fork (fork_event event)
	Notify the cluster of a fork-related event, so it can keep its I/O usable across fork().
void	close (std::function< void()> &&handler)
auto	close () -> std::future< void >
auto	set_authenticator (const password_authenticator &authenticator) -> error
	Replaces the current authenticator used by this cluster.
auto	set_authenticator (const certificate_authenticator &authenticator) -> error
	Replaces the current authenticator used by this cluster.
auto	set_authenticator (const jwt_authenticator &authenticator) -> error
	Replaces the current authenticator used by this cluster.
auto	bucket (std::string_view bucket_name) const -> bucket
	Opens a bucket with the given name.
void	query (std::string statement, const query_options &options, query_handler &&handler) const
	Performs a query against the query (N1QL) services.
auto	query (std::string statement, const query_options &options) const -> std::future< std::pair< error, query_result > >
	Performs a query against the query (N1QL) services.
void	query_stream (std::string statement, const query_options &options, query_stream_handler &&handler) const
	Performs a streaming query against the query (N1QL) services.
auto	query_stream (std::string statement, const query_options &options={}) const -> std::future< std::pair< error, query_stream_result > >
	Performs a streaming query against the query (N1QL) services.
void	search (std::string index_name, search_request request, const search_options &options, search_handler &&handler) const
	Performs a request against the full text search services.
auto	search (std::string index_name, search_request request, const search_options &options={}) const -> std::future< std::pair< error, search_result > >
	Performs a request against the full text search services.
void	analytics_query (std::string statement, const analytics_options &options, analytics_handler &&handler) const
	Performs a query against the analytics services.
auto	analytics_query (std::string statement, const analytics_options &options={}) const -> std::future< std::pair< error, analytics_result > >
	Performs a query against the analytics services.
void	analytics_query_stream (std::string statement, const analytics_options &options, analytics_stream_handler &&handler) const
	Performs a streaming query against the analytics services.
auto	analytics_query_stream (std::string statement, const analytics_options &options={}) const -> std::future< std::pair< error, analytics_stream_result > >
	Performs a streaming query against the analytics services.
void	ping (const ping_options &options, ping_handler &&handler) const
	Performs application-level ping requests against services in the Couchbase cluster.
auto	ping (const ping_options &options={}) const -> std::future< std::pair< error, ping_result > >
	Performs application-level ping requests against services in the Couchbase cluster.
void	diagnostics (const diagnostics_options &options, diagnostics_handler &&handler) const
	Assembles a diagnostics report on the current state of the cluster from the SDK's point of view.
auto	diagnostics (const diagnostics_options &options={}) const -> std::future< std::pair< error, diagnostics_result > >
	Assembles a diagnostics report on the current state of the cluster from the SDK's point of view.
void	wait_until_ready (std::chrono::milliseconds timeout, const wait_until_ready_options &options, wait_until_ready_handler &&handler) const
	Waits until the desired state of the cluster is reached, actively pinging its services until they respond or the timeout elapses.
auto	wait_until_ready (std::chrono::milliseconds timeout, const wait_until_ready_options &options={}) const -> std::future< error >
	Waits until the desired state of the cluster is reached, actively pinging its services until they respond or the timeout elapses.
auto	query_indexes () const -> query_index_manager
	Provides access to the N1QL index management services.
auto	analytics_indexes () const -> analytics_index_manager
	Provides access ot the Analytics index management services.
auto	buckets () const -> bucket_manager
	Provides access to the bucket management services.
auto	search_indexes () const -> search_index_manager
	Provides access to search index management services.
auto	transactions () const -> std::shared_ptr< couchbase::transactions::transactions >
	Provides access to transaction services.

Static Public Member Functions
static void	connect (const std::string &connection_string, const cluster_options &options, cluster_connect_handler &&handler)
	Connect to a Couchbase cluster.
static auto	connect (const std::string &connection_string, const cluster_options &options) -> std::future< std::pair< error, cluster > >
	Connect to a Couchbase cluster.

Detailed Description

The cluster is the main entry point when connecting to a Couchbase cluster.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Examples

async_game_server.cxx, and game_server.cxx.

Constructor & Destructor Documentation

cluster() [1/3]

cluster ( )

default

cluster() [2/3]

cluster ( const cluster & other )

default

cluster() [3/3]

cluster ( cluster && other )

default

~cluster()

~cluster ( )

default

Member Function Documentation

analytics_indexes()

auto analytics_indexes ( ) const -> analytics_index_manager

nodiscard

Provides access ot the Analytics index management services.

Returns

a manager instance

Since

1.0.0

Committed

Generally available API and should be preferred in production

analytics_query() [1/2]

void analytics_query	(	std::string	statement,
		const analytics_options &	options,
		analytics_handler &&	handler ) const

Performs a query against the analytics services.

Parameters

statement	the query statement.
options	options to customize the query request.
handler	the handler that implements query_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/analytics/introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

analytics_query() [2/2]

auto analytics_query ( std::string statement, const analytics_options & options = {} ) const -> std::future< std::pair< error, analytics_result > >

nodiscard

Performs a query against the analytics services.

Parameters

statement	the query statement.
options	options to customize the query request.

Returns

future object that carries result of the operation

See also

https://docs.couchbase.com/server/current/analytics/introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

analytics_query_stream() [1/2]

void analytics_query_stream	(	std::string	statement,
		const analytics_options &	options,
		analytics_stream_handler &&	handler ) const

Performs a streaming query against the analytics services.

The handler resolves as soon as the response preamble has been parsed; rows are then pulled lazily from the analytics_stream_result, so the full result is never buffered in memory.

The handler runs on the library's I/O thread, so the stream must be drained with the callback next() overload rather than the blocking ones. The query equivalent shows the pattern:

#include <couchbase/cluster.hxx>
 
#include <spdlog/fmt/bundled/format.h>
 
// After the fmt header: this specializes fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
 
#include <cstdint>
#include <future>
#include <memory>
#include <optional>
#include <string>
#include <utility>
 
// A drain that never blocks a thread: each next() completion issues the following pull from
// inside the completion handler, so the whole result is consumed on the library's I/O threads.
//
// The blocking overloads -- next().get(), meta_data().get(), and the eager iterator -- must NOT
// be used from a completion handler: they park the very thread that has to deliver the row, and
// the stream can never advance. From a handler, always use the callback next() overload.
//
// Chaining the next pull from inside the handler does not grow the stack: a row is delivered from
// the I/O event loop rather than synchronously from within next(), so the pulls do not nest.
class row_counter : public std::enable_shared_from_this<row_counter>
{
public:
  using result_type = std::pair<couchbase::error, std::uint64_t>;
 
  row_counter(couchbase::query_stream_result result,
              std::shared_ptr<std::promise<result_type>> done)
    : result_{ std::move(result) }
    , done_{ std::move(done) }
  {
  }
 
  void pull()
  {
    // shared_from_this() keeps the counter (and the stream handle it owns) alive for exactly as
    // long as pulls are outstanding.
    result_.next(
      [self = shared_from_this()](couchbase::error err, std::optional<couchbase::query_row> row) {
        if (err) {
          self->done_->set_value({ std::move(err), self->rows_ });
          return;
        }
        if (!row) {
          self->done_->set_value({ {}, self->rows_ }); // clean end of stream
          return;
        }
        ++self->rows_;
        self->pull();
      });
  }
 
private:
  couchbase::query_stream_result result_;
  std::shared_ptr<std::promise<result_type>> done_;
  std::uint64_t rows_{ 0 };
};
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./query_stream_async 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;
  }
 
  // However many rows the statement returns, the SDK holds only a bounded window of them: it stops
  // reading the socket once the rows it has buffered exceed a high-water mark and resumes once the
  // consumer drains back below the low-water mark. The consumer sets the pace.
  auto done = std::make_shared<std::promise<row_counter::result_type>>();
  auto finished = done->get_future();
 
  cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 5000) AS n",
                       couchbase::query_options{},
                       [done](couchbase::error err, couchbase::query_stream_result result) {
                         if (err) {
                           done->set_value({ std::move(err), 0 });
                           return;
                         }
                         std::make_shared<row_counter>(std::move(result), done)->pull();
                       });
 
  // The pump reports the row count alongside the error, because a stream can fail *after* rows
  // have already been delivered: those rows were valid and any work done on them stands. See the
  // error-handling example for the full set of failure channels.
  auto [stream_err, rows] = finished.get();
  if (stream_err) {
    fmt::print("streaming query failed after {} row(s): {}\n", rows, stream_err);
    return 1;
  }
  fmt::println("counted {} rows without blocking a thread", rows);
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_async couchbase://127.0.0.1 Administrator password
counted 5000 rows without blocking a thread
 
 */

Parameters

statement	the analytics query statement.
options	options to customize the query request.
handler	the handler that implements analytics_stream_handler

Since

1.4.0

Volatile

Should not be used in production

analytics_query_stream() [2/2]

auto analytics_query_stream ( std::string statement, const analytics_options & options = {} ) const -> std::future< std::pair< error, analytics_stream_result > >

nodiscard

Performs a streaming query against the analytics services.

#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
 
 */

Parameters

statement	the analytics query statement.
options	options to customize the query request.

Returns

future object that carries the streaming result handle

Since

1.4.0

Volatile

Should not be used in production

bucket()

auto bucket ( std::string_view bucket_name ) const -> bucket

nodiscard

Opens a bucket with the given name.

Parameters

bucket_name

the name of the bucket to open.

Returns

a bucket once opened.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Examples

async_game_server.cxx, and game_server.cxx.

buckets()

auto buckets ( ) const -> bucket_manager

nodiscard

Provides access to the bucket management services.

Returns

a manager instance

Since

1.0.0

Committed

Generally available API and should be preferred in production

close() [1/2]

auto close ( ) -> std::future< void >

nodiscard

close() [2/2]

void close

(

std::function< void()> &&

handler

)

Examples

async_game_server.cxx, and game_server.cxx.

connect() [1/2]

auto connect ( const std::string & connection_string, const cluster_options & options ) -> std::future< std::pair< error, cluster > >

staticnodiscard

Connect to a Couchbase cluster.

Parameters

connection_string	connection string used to locate the Couchbase cluster object.
options	options to customize connection (note, that connection_string takes precedence over this options).

Returns

future object that carries cluster object and operation status

Note

A connection string with the couchbase2:// scheme selects the couchbase2 transport, which routes every operation through a Cloud Native Gateway (CNG). It requires a library built with that support and restricts how the connection string may be written; see The couchbase2:// transport.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

connect() [2/2]

void connect ( const std::string & connection_string, const cluster_options & options, cluster_connect_handler && handler )

static

Connect to a Couchbase cluster.

Parameters

connection_string	connection string used to locate the Couchbase cluster object.
options	options to customize connection (note, that connection_string takes precedence over this options).
handler	the handler

Note

A connection string with the couchbase2:// scheme selects the couchbase2 transport, which routes every operation through a Cloud Native Gateway (CNG). It requires a library built with that support and restricts how the connection string may be written; see The couchbase2:// transport.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

Examples

async_game_server.cxx, distributed_mutex.cxx, game_server.cxx, and minimal.cxx.

diagnostics() [1/2]

void diagnostics	(	const diagnostics_options &	options,
		diagnostics_handler &&	handler ) const

Assembles a diagnostics report on the current state of the cluster from the SDK's point of view.

Note

This operation does not perform any I/O to produce the report. It will only use the current known state of the cluster to assemble the report So, if for example, no SQL++ queries have been run, the Query service's socket pool might be empty and as a result not show up in the report. If you wish to actively assess the health of the cluster by performing I/O, consider using ping() instead.

Parameters

options	custom options to change the default behavior.
handler	the handler that implements diagnostics_handler.

Note

Not served over the couchbase2 transport, which keeps no per-node connection state to report on and returns feature_not_available rather than an empty report.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

diagnostics() [2/2]

auto diagnostics ( const diagnostics_options & options = {} ) const -> std::future< std::pair< error, diagnostics_result > >

nodiscard

Assembles a diagnostics report on the current state of the cluster from the SDK's point of view.

Note

This operation does not perform any I/O to produce the report. It will only use the current known state of the cluster to assemble the report So, if for example, no SQL++ queries have been run, the Query service's socket pool might be empty and as a result not show up in the report. If you wish to actively assess the health of the cluster by performing I/O, consider using ping() instead.

Parameters

options

custom options to change the default behavior.

Returns

future object that carries result of the operation.

Note

Not served over the couchbase2 transport, which keeps no per-node connection state to report on and returns feature_not_available rather than an empty report.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

notify_fork()

void notify_fork

(

fork_event

event

)

Notify the cluster of a fork-related event, so it can keep its I/O usable across fork().

Call it three times, in this order: prepare in the parent before forking, then child in the child and parent in the parent.

NOTE: child reconnects this cluster on a fresh set of sockets, because the inherited ones belong to the parent. Any bucket, scope or collection handle obtained before the fork still refers to the pre-fork connections, so in the child those handles must be re-acquired from the cluster – using a stale one fails with cluster_closed. Handles held in the parent stay valid.

This function must not be called while operations are in flight on this cluster from another thread.

On platforms without fork(), Windows in particular, this function does nothing. It remains callable everywhere so that portable code does not have to guard the calls, but nothing about the cluster changes.

Parameters

event

the fork-related event that is about to happen, or just happened.

Exceptions

std::system_error

if the I/O backend cannot be carried across the fork. There is no recovery for this: the cluster's io_context is unusable afterwards and the only thing left to do with the cluster is destroy it. It is reported rather than swallowed because a caller that went on using such a cluster would be operating on a reactor that no longer tracks its sockets.

Since

1.0.0

Uncommitted

Might be changed in the future, and eventually promoted to committed

operator=() [1/2]

auto operator= ( cluster && other ) -> cluster &=default

default

operator=() [2/2]

auto operator= ( const cluster & other ) -> cluster &=default

default

ping() [1/2]

void ping	(	const ping_options &	options,
		ping_handler &&	handler ) const

Performs application-level ping requests against services in the Couchbase cluster.

Note

This operation performs active I/O against services and endpoints to assess their health. If you do not wish to performs I/O, consider using diagnostics() instead.

Parameters

options	custom options to change the default behavior.
handler	the handler that implements ping_handler.

Note

Not served over the couchbase2 transport, which has no per-node service endpoints to probe and reports feature_not_available instead.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

ping() [2/2]

auto ping ( const ping_options & options = {} ) const -> std::future< std::pair< error, ping_result > >

nodiscard

Performs application-level ping requests against services in the Couchbase cluster.

Note

This operation performs active I/O against services and endpoints to assess their health. If you do not wish to performs I/O, consider using diagnostics() instead.

Parameters

options

custom options to change the default behavior.

Returns

future object that carries result of the operation.

Note

Not served over the couchbase2 transport, which has no per-node service endpoints to probe and reports feature_not_available instead.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Since

1.4.0 – Cloud Native Gateway (CNG, couchbase2://) support

Uncommitted

Cloud Native Gateway (CNG, couchbase2://) support is uncommitted and may change in a future release, independently of the stability of the API it applies to

query() [1/2]

auto query ( std::string statement, const query_options & options ) const -> std::future< std::pair< error, query_result > >

nodiscard

Performs a query against the query (N1QL) services.

Parameters

statement	the N1QL query statement.
options	options to customize the query request.

Returns

future object that carries result of the operation

Since

1.0.0

Committed

Generally available API and should be preferred in production

query() [2/2]

void query	(	std::string	statement,
		const query_options &	options,
		query_handler &&	handler ) const

Performs a query against the query (N1QL) services.

Parameters

statement	the N1QL query statement.
options	options to customize the query request.
handler	the handler that implements query_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

Since

1.0.0

Committed

Generally available API and should be preferred in production

query_indexes()

auto query_indexes ( ) const -> query_index_manager

nodiscard

Provides access to the N1QL index management services.

Returns

a manager instance

Since

1.0.0

Committed

Generally available API and should be preferred in production

query_stream() [1/2]

void query_stream	(	std::string	statement,
		const query_options &	options,
		query_stream_handler &&	handler ) const

Performs a streaming query against the query (N1QL) services.

The handler resolves as soon as the response preamble has been parsed; rows are then pulled lazily from the query_stream_result, so the full result is never buffered in memory.

Note

Prepared statements (adhoc set to false) are not streamed: such a request falls back to the buffered query() path and its rows are replayed through the returned handle.

The handler runs on the library's I/O thread, so the stream must be drained with the callback next() overload rather than the blocking ones:

#include <couchbase/cluster.hxx>
 
#include <spdlog/fmt/bundled/format.h>
 
// After the fmt header: this specializes fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
 
#include <cstdint>
#include <future>
#include <memory>
#include <optional>
#include <string>
#include <utility>
 
// A drain that never blocks a thread: each next() completion issues the following pull from
// inside the completion handler, so the whole result is consumed on the library's I/O threads.
//
// The blocking overloads -- next().get(), meta_data().get(), and the eager iterator -- must NOT
// be used from a completion handler: they park the very thread that has to deliver the row, and
// the stream can never advance. From a handler, always use the callback next() overload.
//
// Chaining the next pull from inside the handler does not grow the stack: a row is delivered from
// the I/O event loop rather than synchronously from within next(), so the pulls do not nest.
class row_counter : public std::enable_shared_from_this<row_counter>
{
public:
  using result_type = std::pair<couchbase::error, std::uint64_t>;
 
  row_counter(couchbase::query_stream_result result,
              std::shared_ptr<std::promise<result_type>> done)
    : result_{ std::move(result) }
    , done_{ std::move(done) }
  {
  }
 
  void pull()
  {
    // shared_from_this() keeps the counter (and the stream handle it owns) alive for exactly as
    // long as pulls are outstanding.
    result_.next(
      [self = shared_from_this()](couchbase::error err, std::optional<couchbase::query_row> row) {
        if (err) {
          self->done_->set_value({ std::move(err), self->rows_ });
          return;
        }
        if (!row) {
          self->done_->set_value({ {}, self->rows_ }); // clean end of stream
          return;
        }
        ++self->rows_;
        self->pull();
      });
  }
 
private:
  couchbase::query_stream_result result_;
  std::shared_ptr<std::promise<result_type>> done_;
  std::uint64_t rows_{ 0 };
};
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./query_stream_async 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;
  }
 
  // However many rows the statement returns, the SDK holds only a bounded window of them: it stops
  // reading the socket once the rows it has buffered exceed a high-water mark and resumes once the
  // consumer drains back below the low-water mark. The consumer sets the pace.
  auto done = std::make_shared<std::promise<row_counter::result_type>>();
  auto finished = done->get_future();
 
  cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 5000) AS n",
                       couchbase::query_options{},
                       [done](couchbase::error err, couchbase::query_stream_result result) {
                         if (err) {
                           done->set_value({ std::move(err), 0 });
                           return;
                         }
                         std::make_shared<row_counter>(std::move(result), done)->pull();
                       });
 
  // The pump reports the row count alongside the error, because a stream can fail *after* rows
  // have already been delivered: those rows were valid and any work done on them stands. See the
  // error-handling example for the full set of failure channels.
  auto [stream_err, rows] = finished.get();
  if (stream_err) {
    fmt::print("streaming query failed after {} row(s): {}\n", rows, stream_err);
    return 1;
  }
  fmt::println("counted {} rows without blocking a thread", rows);
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_async couchbase://127.0.0.1 Administrator password
counted 5000 rows without blocking a thread
 
 */

Parameters

statement	the N1QL query statement.
options	options to customize the query request.
handler	the handler that implements query_stream_handler

Since

1.4.0

Volatile

Should not be used in production

query_stream() [2/2]

auto query_stream ( std::string statement, const query_options & options = {} ) const -> std::future< std::pair< error, query_stream_result > >

nodiscard

Performs a streaming query against the query (N1QL) services.

Rows are pulled from the returned handle one at a time, so a result of any size costs the same bounded amount of memory:

#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
 
 */

The document type the rows are decoded into:

// 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>(),
    };
  }
};

A failure surfaces either here, before the stream starts, or as the terminal of next() — possibly after rows have already been consumed, which a buffered query() cannot produce. Both channels have to be handled:

#include <couchbase/cluster.hxx>
#include <couchbase/error_codes.hxx>
 
#include <spdlog/fmt/bundled/format.h>
 
// After the fmt header: this specializes fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
 
#include <chrono>
#include <cstdint>
#include <string>
 
namespace
{
// How far a drain got. A streaming query is not all-or-nothing the way a buffered query() is: it
// can hand over rows and only then fail, and those rows were valid. An application that mutates
// state per row has to decide whether that partial work is acceptable, so the row count travels
// with the error rather than being discarded.
struct drain_report {
  couchbase::error error{};
  std::uint64_t rows{ 0 };
};
 
auto
drain(const couchbase::query_stream_result& result) -> drain_report
{
  drain_report report{};
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      report.error = std::move(row_err); // terminal error: no further row will arrive
      return report;
    }
    if (!row) {
      return report; // clean end of stream
    }
    ++report.rows;
  }
}
} // namespace
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./query_stream_errors 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;
  }
 
  // A syntactically valid statement over a keyspace that does not exist. Where the failure
  // surfaces is not fixed: the query service may reject the request while the SDK is still
  // reading the response preamble, or accept it and then terminate the stream. Both channels
  // have to be handled -- a caller that only checks one of them will miss the failure.
  const auto invalid_statement = R"(SELECT * FROM `nonexistent_keyspace_xyz` LIMIT 1)";
  bool saw_error{ false };
 
  { // [1] Both channels for one failing statement. The error is either carried by the
    //     query_stream() future (the request never started, and the returned handle is not a
    //     usable stream), or delivered as the stream's terminal by next(). The row count that
    //     travels with the terminal is the part of the result that was consumed successfully.
    // Formatting an error renders its context too: the code and message the service itself
    // reported, the statement and encoded parameters, and the node that answered. That is what to
    // log. couchbase::error::ctx().to_json() returns the same context on its own, for a structured
    // log line.
    auto [err, result] = cluster.query_stream(invalid_statement).get();
    if (err) {
      saw_error = true;
      fmt::println("[1] rejected before streaming began: {}", err);
    } else {
      auto report = drain(result);
      if (report.error) {
        saw_error = true;
        fmt::println("[1] stream failed after {} row(s): {}", report.rows, report.error);
      } else {
        fmt::println("[1] stream ended cleanly with {} row(s)", report.rows);
      }
    }
  }
 
  { // [2] The iterator reports a terminal error as one final element (truthy error, empty row)
    //     before it compares equal to end(), so a range-based for loop surfaces the failure
    //     instead of quietly stopping short of the result. Ignoring the error half of the pair is
    //     how a truncated result gets mistaken for a complete one.
    auto [err, result] = cluster.query_stream(invalid_statement).get();
    if (err) {
      saw_error = true;
      fmt::println("[2] rejected before streaming began: {}", err);
    } else {
      std::uint64_t rows{ 0 };
      for (auto [row_err, row] : result) {
        if (row_err) {
          saw_error = true;
          fmt::println("[2] iterator saw the terminal error after {} row(s): {}", rows, row_err);
          break;
        }
        ++rows;
      }
    }
  }
 
  { // [3] Deliberate teardown is a terminal too, and has to be told apart from a real failure:
    //     after cancel(), next() reports errc::common::request_canceled. The rows consumed before
    //     the cancel are unaffected -- this is the partial-success case that a buffered query()
    //     cannot produce.
    auto [err, result] = cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 15000) AS n").get();
    if (err) {
      fmt::print("unable to start the streaming query: {}\n", err);
      return 1;
    }
 
    std::uint64_t rows{ 0 };
    couchbase::error before_cancel{};
    while (rows < 3) {
      auto [row_err, row] = result.next().get();
      if (row_err) {
        before_cancel = std::move(row_err);
        break;
      }
      if (!row) {
        break; // the result was shorter than expected
      }
      ++rows;
    }
    result.cancel();
 
    const auto after_cancel = drain(result);
    if (before_cancel) {
      fmt::println("[3] the stream failed before it could be cancelled: {}", before_cancel);
    } else if (after_cancel.error.ec() == couchbase::errc::common::request_canceled) {
      fmt::println("[3] stream cancelled after {} row(s); those rows are still valid", rows);
    } else {
      fmt::println("[3] unexpected terminal after cancel(): \"{}\"",
                   after_cancel.error.ec().message());
    }
 
    // [4] The terminal is sticky and idempotent: further next() calls re-deliver it rather than
    //     blocking on a drained stream, so a drain loop can never hang on a dead stream.
    auto [again_err, again_row] = result.next().get();
    fmt::println("[4] re-reading past the terminal: error=\"{}\", row_present={}",
                 again_err.ec().message(),
                 again_row.has_value());
 
    // [5] meta_data() resolves with the failure instead of parking forever waiting for a trailer
    //     that will never arrive.
    auto [meta_err, meta] = result.meta_data().get();
    fmt::println("[5] meta_data() after a torn-down stream: \"{}\"", meta_err.ec().message());
  }
 
  { // [6] Timeouts are classified by whether the request could have applied a mutation:
    //     errc::common::unambiguous_timeout when the request is read-only (it definitely did not
    //     apply, so retrying is safe) and errc::common::ambiguous_timeout otherwise (it may
    //     already have been applied -- do not blindly retry). The classification follows the
    //     request's read-only flag, which defaults to false: the SDK does not infer read-only-ness
    //     from the statement text, so mark read-only queries explicitly to get the retryable
    //     classification.
    //
    //     The deadline applies as a whole-request timeout until the response headers arrive, and
    //     from then on as an *inter-read idle* timeout that is armed only while a socket read is in
    //     flight. So a slow consumer, which generates no socket traffic, is never timed out; a
    //     mid-stream fire means the server stalled mid-body. Either way the classification is the
    //     same, and the code below handles the deadline firing on either side of the preamble.
    const auto options = couchbase::query_options{}.readonly(true).timeout(
      std::chrono::milliseconds{ 1 }); // far too short, on purpose
    auto [err, result] =
      cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 15000) AS n", options).get();
    auto ec = err.ec();
    std::uint64_t rows{ 0 };
    if (!ec) {
      const auto report = drain(result);
      ec = report.error.ec();
      rows = report.rows;
    }
    if (ec == couchbase::errc::common::unambiguous_timeout) {
      fmt::println("[6] read-only statement timed out after {} row(s); retrying it is safe", rows);
    } else if (ec == couchbase::errc::common::ambiguous_timeout) {
      fmt::println("[6] statement timed out ambiguously after {} row(s); do not blindly retry",
                   rows);
    } else {
      fmt::println("[6] statement did not time out ({} row(s), \"{}\")", rows, ec.message());
    }
  }
 
  if (!saw_error) {
    fmt::print("expected the invalid statement to fail, but it did not\n");
    return 1;
  }
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_errors couchbase://127.0.0.1 Administrator password
[1] rejected before streaming began: index_failure (202) | {"client_context_id":"1f2170-81d8-224f-
    45e4-6df76260782b22","first_error_code":12003,"first_error_message":"Keyspace not found in CB
    datastore: default:nonexistent_keyspace_xyz (near line 1, column 15) - cause: No bucket named
    nonexistent_keyspace_xyz","hostname":"172.18.0.5", ... ,"statement":"SELECT * FROM
    `nonexistent_keyspace_xyz` LIMIT 1"}
    (one line in reality; wrapped here, and the remaining context fields elided, for readability)
[2] rejected before streaming began: index_failure (202) | { ... same context ... }
[3] stream cancelled after 3 row(s); those rows are still valid
[4] re-reading past the terminal: error="request_canceled (2)", row_present=false
[5] meta_data() after a torn-down stream: "request_canceled (2)"
[6] read-only statement timed out after 0 row(s); retrying it is safe
 
 */

Parameters

statement	the N1QL query statement.
options	options to customize the query request.

Returns

future object that carries the streaming result handle

Since

1.4.0

Volatile

Should not be used in production

search() [1/2]

void search	(	std::string	index_name,
		search_request	request,
		const search_options &	options,
		search_handler &&	handler ) const

Performs a request against the full text search services.

This can be used to perform a traditional FTS query, and/or a vector search.

Parameters

index_name	name of the search index
request	request object, see search_request for more details.
options	options to customize the query request.
handler	the handler that implements search_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/fts/fts-introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

search() [2/2]

auto search ( std::string index_name, search_request request, const search_options & options = {} ) const -> std::future< std::pair< error, search_result > >

nodiscard

Performs a request against the full text search services.

This can be used to perform a traditional FTS query, and/or a vector search.

Parameters

index_name	name of the search index
request	request object, see search_request for more details.
options	options to customize the query request.

Returns

future object that carries result of the operation

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/fts/fts-introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

search_indexes()

auto search_indexes ( ) const -> search_index_manager

nodiscard

Provides access to search index management services.

Returns

a manager instance

Since

1.0.0

Committed

Generally available API and should be preferred in production

set_authenticator() [1/3]

auto set_authenticator

(

const certificate_authenticator &

authenticator

)

-> error

Replaces the current authenticator used by this cluster.

NOTE: Setting a new authenticator does not change the authentication status of existing connections.

Parameters

authenticator

the authenticator to replace

Exceptions

errc::common::invalid_argument

if TLS is not enabled.

Returns

error

Since

1.3.0

Committed

Generally available API and should be preferred in production

set_authenticator() [2/3]

auto set_authenticator

(

const jwt_authenticator &

authenticator

)

-> error

Replaces the current authenticator used by this cluster.

NOTE: Setting a new authenticator does not change the authentication status of existing connections.

Parameters

authenticator

the authenticator to replace

Returns

error

Since

1.3.0

Uncommitted

Might be changed in the future, and eventually promoted to committed

set_authenticator() [3/3]

auto set_authenticator

(

const password_authenticator &

authenticator

)

-> error

Replaces the current authenticator used by this cluster.

NOTE: Setting a new authenticator does not change the authentication status of existing connections.

Parameters

authenticator

the authenticator to replace

Returns

error

Since

1.3.0

Committed

Generally available API and should be preferred in production

transactions()

auto transactions ( ) const -> std::shared_ptr< couchbase::transactions::transactions >

nodiscard

Provides access to transaction services.

See transactions for details on using the transactions object.

Returns

an transactions object

Since

1.0.0

Committed

Generally available API and should be preferred in production

wait_until_ready() [1/2]

void wait_until_ready	(	std::chrono::milliseconds	timeout,
		const wait_until_ready_options &	options,
		wait_until_ready_handler &&	handler ) const

Waits until the desired state of the cluster is reached, actively pinging its services until they respond or the timeout elapses.

Parameters

timeout	the maximum time to wait for readiness.
options	custom options to change the default behavior.
handler	the handler that implements wait_until_ready_handler.

Since

1.4.0

Committed

Generally available API and should be preferred in production

wait_until_ready() [2/2]

auto wait_until_ready ( std::chrono::milliseconds timeout, const wait_until_ready_options & options = {} ) const -> std::future< error >

nodiscard

Waits until the desired state of the cluster is reached, actively pinging its services until they respond or the timeout elapses.

Parameters

timeout	the maximum time to wait for readiness.
options	custom options to change the default behavior.

Returns

future object that carries result of the operation.

Since

1.4.0

Committed

Generally available API and should be preferred in production

---

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

• couchbase/cluster.hxx