a11::nodes::AsyncNode Class Reference

An asynchronous, ordered stream of chunks read from and written to A11. More...

#include <cpp/a11/nodes/async_node.h>

Inheritance diagram for a11::nodes::AsyncNode:

Public Member Functions
	~AsyncNode ()=default
	AsyncNode (const AsyncNode &)=delete
AsyncNode &	operator= (const AsyncNode &)=delete
absl::StatusOr< std::string >	GetId () const
	Return the node's stable identifier.
std::shared_ptr< stores::ChunkStore >	GetChunkStore () const
	Return the underlying chunk store backing this node.
std::shared_ptr< data::SerializationRegistry >	serialization_registry () const
	Return the registry used to (de)serialize typed values.
absl::Status	SetSerializationRegistry (std::shared_ptr< data::SerializationRegistry > registry)
	Replace the serialization registry.
absl::StatusOr< std::shared_ptr< stores::ChunkStoreReader > >	reader ()
	Return the node's chunk store reader (the consuming half).
absl::StatusOr< std::shared_ptr< stores::ChunkStoreWriter > >	writer ()
	Return the node's chunk store writer (the producing half).
stores::ChunkStoreReaderOptions	GetReaderOptions () const
	Return a copy of the reader's current options.
absl::Status	SetReaderOptions (stores::ChunkStoreReaderOptions options)
	Replace the reader options.
absl::Status	ResetReader (std::optional< stores::ChunkStoreReaderOptions > options=std::nullopt)
	Rewind/reconfigure the reader (e.g.
stores::ChunkStoreWriterOptions	GetWriterOptions () const
	Return a copy of the writer's current options.
absl::Status	SetWriterOptions (stores::ChunkStoreWriterOptions options)
	Replace the writer options.
absl::Status	GetReaderStatus () const
	Return the current status of the node's reader.
absl::Status	GetWriterStatus () const
	Return the current status of the node's writer.
std::optional< absl::Status >	GetWriterAbortStatus () const
	Return the status the writer was aborted with.
a11::Future< bool >	IsWritable ()
	Report whether the node can currently accept writes.
a11::Future< std::uint32_t >	PutChunk (data::Chunk chunk, std::optional< std::uint32_t > seq=std::nullopt, bool final=false)
	Enqueue a raw chunk into the stream.
a11::Future< std::uint32_t >	PutFragment (data::NodeFragment fragment)
	Enqueue a fragment carrying its own sequence and final flag.
template<typename T >
a11::Future< std::uint32_t >	Put (const T &value, std::optional< std::uint32_t > seq=std::nullopt, bool final=false, std::string_view mimetype={})
	Serialize and write a typed value to the stream.
a11::Future< std::uint32_t >	PutNullFinal (std::optional< std::uint32_t > seq=std::nullopt)
	Close the stream with an explicit null terminator (no value).
a11::Future< std::optional< data::NodeFragment > >	NextFragment (absl::Duration timeout=absl::InfiniteDuration())
	Read the next raw fragment.
a11::Future< std::optional< data::Chunk > >	NextChunk (absl::Duration timeout=absl::InfiniteDuration())
	Read the next raw chunk.
template<typename T >
a11::Future< std::optional< T > >	NextObject (absl::Duration timeout=absl::InfiniteDuration(), std::vector< std::string > mimetype_patterns={})
	Read and deserialize the next value.
a11::Task	WaitForBufferToDrain ()
	Wait for the write buffer to drain.
a11::Task	DrainAndClose ()
	Flush all buffered chunks and close the writer.
a11::Task	AbortWithStatus (absl::Status status)
	Fail the stream with an error status.
absl::Status	AttachStream (std::shared_ptr< net::WireStream > stream)
	Tee this node's stored chunks onto a wire stream.
absl::Status	DetachStream (const std::shared_ptr< net::WireStream > &stream)
	Stop mirroring chunks over a previously attached wire stream.
void	CancelReader ()
	Cancel the reader, unblocking pending Next* awaits.
void	CancelWriter ()
	Cancel the writer, unblocking pending Put/drain awaits.
void	Cancel ()
	Cancel both reader and writer, tearing down all pending streaming operations on the node at once.

Static Public Member Functions
static absl::StatusOr< std::shared_ptr< AsyncNode > >	Create (std::shared_ptr< stores::ChunkStore > store, std::shared_ptr< data::SerializationRegistry > serialization_registry=nullptr, stores::ChunkStoreReaderOptions reader_options={}, stores::ChunkStoreWriterOptions writer_options={})
	Create a node over an existing chunk store.

Detailed Description

An asynchronous, ordered stream of chunks read from and written to A11.

A node has two halves. The writer end admits values into the backing stores::ChunkStore in sequence; the reader end yields them back, optionally deserializing typed objects on the way out. Every Put* resolves once the backing store accepts the chunk. During the same flush, the writer attempts to enqueue the batch on attached net::WireStreams; this is not a remote-delivery acknowledgement, and a later tee failure cannot revoke the current batch's store confirmation. A null final chunk marks the logical end of the data. The producer closes the writer separately with DrainAndClose().

Instances are always heap-allocated and shared via Create; the class is non-copyable and derives from enable_shared_from_this.

Constructor & Destructor Documentation

~AsyncNode()

a11::nodes::AsyncNode::~AsyncNode ( )

default

AsyncNode()

a11::nodes::AsyncNode::AsyncNode ( const AsyncNode &  )

delete

Member Function Documentation

AbortWithStatus()

a11::Task a11::nodes::AsyncNode::AbortWithStatus

(

absl::Status

status

)

Fail the stream with an error status.

Parameters

status

The error to abort with.

Returns

An awaitable that resolves once the abort has propagated, so consumers observe the error instead of a normal end-of-stream.

AttachStream()

absl::Status a11::nodes::AsyncNode::AttachStream

(

std::shared_ptr< net::WireStream >

stream

)

Tee this node's stored chunks onto a wire stream.

WireStream::Send() confirms local transport admission, not receipt by the remote agent. A send failure stops later writes but cannot revoke the current batch's store confirmations.

Parameters

stream

The transport to attach; kept alive for the node's lifetime.

Returns

OK, or an error status on failure.

Cancel()

void a11::nodes::AsyncNode::Cancel

(

)

Cancel both reader and writer, tearing down all pending streaming operations on the node at once.

CancelReader()

void a11::nodes::AsyncNode::CancelReader

(

)

Cancel the reader, unblocking pending Next* awaits.

CancelWriter()

void a11::nodes::AsyncNode::CancelWriter

(

)

Cancel the writer, unblocking pending Put/drain awaits.

Create()

absl::StatusOr< std::shared_ptr< AsyncNode > > a11::nodes::AsyncNode::Create ( std::shared_ptr< stores::ChunkStore >  store, std::shared_ptr< data::SerializationRegistry >  serialization_registry = nullptr, stores::ChunkStoreReaderOptions  reader_options = {}, stores::ChunkStoreWriterOptions  writer_options = {}  )

static

Create a node over an existing chunk store.

Parameters

store	The backing ordered buffer the reader and writer stream through.
serialization_registry	Registry used to (de)serialize typed values passed to Put/NextObject; defaults to none.
reader_options	Options controlling how the reader buffers and orders chunks.
writer_options	Options controlling how the writer buffers chunks.

Returns

The new node, or an error status on failure.

DetachStream()

absl::Status a11::nodes::AsyncNode::DetachStream

(

const std::shared_ptr< net::WireStream > &

stream

)

Stop mirroring chunks over a previously attached wire stream.

Parameters

stream

The transport to detach.

Returns

OK, or an error status on failure.

DrainAndClose()

a11::Task a11::nodes::AsyncNode::DrainAndClose

(

)

Flush all buffered chunks and close the writer.

This is a storage-lifecycle operation, not an end-of-data marker. It does not append a final fragment or choose a final sequence number. A producer should first call PutChunk(..., final=true) or PutNullFinal() so readers can synchronise on the logical end of the node, then call DrainAndClose() to wait for persistence and release the writer.

Returns

An awaitable that resolves once every produced chunk has been flushed and the backing store is closed to further writes.

GetChunkStore()

std::shared_ptr< stores::ChunkStore > a11::nodes::AsyncNode::GetChunkStore

(

)

const

Return the underlying chunk store backing this node.

Returns

The ordered storage boundary the reader and writer stream through; reach for it when you need lower-level access than the Put/Next API provides.

GetId()

absl::StatusOr< std::string > a11::nodes::AsyncNode::GetId

(

)

const

Return the node's stable identifier.

Returns

The id, or an error status if it cannot be resolved. Use it to correlate the node with the rest of an agent's state or to key it in a NodeMap.

GetReaderOptions()

stores::ChunkStoreReaderOptions a11::nodes::AsyncNode::GetReaderOptions

(

)

const

Return a copy of the reader's current options.

Returns

The reader options in effect.

GetReaderStatus()

absl::Status a11::nodes::AsyncNode::GetReaderStatus

(

)

const

Return the current status of the node's reader.

Returns

Whether the consuming end is healthy, has completed, or has failed while streaming.

GetWriterAbortStatus()

std::optional< absl::Status > a11::nodes::AsyncNode::GetWriterAbortStatus

(

)

const

Return the status the writer was aborted with.

Returns

The abort status, or nullopt if the writer has not been aborted. Use it to surface why a stream was cut short.

GetWriterOptions()

stores::ChunkStoreWriterOptions a11::nodes::AsyncNode::GetWriterOptions

(

)

const

Return a copy of the writer's current options.

Returns

The writer options in effect.

GetWriterStatus()

absl::Status a11::nodes::AsyncNode::GetWriterStatus

(

)

const

Return the current status of the node's writer.

Returns

Whether the producing end is healthy, has completed, or has failed while streaming.

IsWritable()

a11::Future< bool > a11::nodes::AsyncNode::IsWritable

(

)

Report whether the node can currently accept writes.

Returns

An awaitable that resolves once writability is known; await it before producing chunks to respect backpressure.

NextChunk()

a11::Future< std::optional< data::Chunk > > a11::nodes::AsyncNode::NextChunk

(

absl::Duration

timeout = absl::InfiniteDuration()

)

Read the next raw chunk.

Parameters

timeout

How long to wait for a chunk.

Returns

An awaitable that resolves to the next chunk, or nullopt at end of stream.

NextFragment()

a11::Future< std::optional< data::NodeFragment > > a11::nodes::AsyncNode::NextFragment

(

absl::Duration

timeout = absl::InfiniteDuration()

)

Read the next raw fragment.

Parameters

timeout

How long to wait for a fragment.

Returns

An awaitable that resolves to the next fragment, or nullopt at end of stream.

NextObject()

template<typename T >

a11::Future< std::optional< T > > a11::nodes::AsyncNode::NextObject ( absl::Duration  timeout = absl::InfiniteDuration(), std::vector< std::string >  mimetype_patterns = {}  )

inline

Read and deserialize the next value.

Template Parameters

T	The type to deserialize into.

Parameters

timeout	How long to wait for a value.
mimetype_patterns	Optional MIME patterns constraining which encodings are accepted.

Returns

An awaitable that resolves to the next value, or nullopt at end of stream.

operator=()

AsyncNode & a11::nodes::AsyncNode::operator= ( const AsyncNode &  )

delete

Put()

template<typename T >

a11::Future< std::uint32_t > a11::nodes::AsyncNode::Put ( const T &  value, std::optional< std::uint32_t >  seq = std::nullopt, bool  final = false, std::string_view  mimetype = {}  )

inline

Serialize and write a typed value to the stream.

Encodes value via the node's serialization registry and admits the resulting chunk.

Template Parameters

T	The type of the value being written.

Parameters

value	The value to encode and write.
seq	Optional explicit sequence number; assigned in order when omitted.
final	Set true on the last write to establish the logical final sequence. This does not close the writer.
mimetype	Optional MIME type selecting the encoding.

Returns

An awaitable that resolves to the stored sequence number, or a failed future if serialization fails.

PutChunk()

a11::Future< std::uint32_t > a11::nodes::AsyncNode::PutChunk	(	data::Chunk	chunk,
		std::optional< std::uint32_t >	seq = std::nullopt,
		bool	final = false
	)

Enqueue a raw chunk into the stream.

Parameters

chunk	The chunk to admit.
seq	Optional explicit sequence number; assigned in order when omitted.
final	Set true on the last chunk to establish the logical final sequence. This does not close the writer.

Returns

An awaitable that resolves to the sequence number once the backing store accepts the chunk. Attached stream sends are attempted during the flush but do not acknowledge remote delivery.

PutFragment()

a11::Future< std::uint32_t > a11::nodes::AsyncNode::PutFragment

(

data::NodeFragment

fragment

)

Enqueue a fragment carrying its own sequence and final flag.

Parameters

fragment

The fragment to admit.

Returns

An awaitable that resolves to the stored sequence number.

PutNullFinal()

a11::Future< std::uint32_t > a11::nodes::AsyncNode::PutNullFinal

(

std::optional< std::uint32_t >

seq = std::nullopt

)

Close the stream with an explicit null terminator (no value).

Parameters

seq	Optional explicit sequence number for the terminator.

Returns

An awaitable that resolves to the stored sequence number.

reader()

absl::StatusOr< std::shared_ptr< stores::ChunkStoreReader > > a11::nodes::AsyncNode::reader

(

)

Return the node's chunk store reader (the consuming half).

Returns

The reader, or an error status on failure.

ResetReader()

absl::Status a11::nodes::AsyncNode::ResetReader

(

std::optional< stores::ChunkStoreReaderOptions >

options = std::nullopt

)

Rewind/reconfigure the reader (e.g.

to re-read from an offset).

Parameters

options

Optional new reader options; the existing options are kept when omitted.

Returns

OK, or an error status on failure.

serialization_registry()

std::shared_ptr< data::SerializationRegistry > a11::nodes::AsyncNode::serialization_registry

(

)

const

Return the registry used to (de)serialize typed values.

Returns

The current serialization registry (may be null).

SetReaderOptions()

absl::Status a11::nodes::AsyncNode::SetReaderOptions

(

stores::ChunkStoreReaderOptions

options

)

Replace the reader options.

Parameters

options

The new reader options.

Returns

OK, or an error status on failure.

SetSerializationRegistry()

absl::Status a11::nodes::AsyncNode::SetSerializationRegistry

(

std::shared_ptr< data::SerializationRegistry >

registry

)

Replace the serialization registry.

Parameters

registry

The registry to install.

Returns

OK, or an error status on failure.

SetWriterOptions()

absl::Status a11::nodes::AsyncNode::SetWriterOptions

(

stores::ChunkStoreWriterOptions

options

)

Replace the writer options.

Parameters

options

The new writer options.

Returns

OK, or an error status on failure.

WaitForBufferToDrain()

a11::Task a11::nodes::AsyncNode::WaitForBufferToDrain

(

)

Wait for the write buffer to drain.

Returns

An awaitable that resolves once the write buffer has drained; await it to let consumers catch up before pushing more chunks.

writer()

absl::StatusOr< std::shared_ptr< stores::ChunkStoreWriter > > a11::nodes::AsyncNode::writer

(

)

Return the node's chunk store writer (the producing half).

Returns

The writer, or an error status on failure.

---

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

• cpp/a11/nodes/async_node.h
• cpp/a11/nodes/async_node.cc