Base class for all data sinks in NES. More...

#include <SinkMedium.hpp>

Collaboration diagram for NES::SinkMedium:

Public Member Functions
	SinkMedium (SinkFormatPtr sinkFormat, Runtime::NodeEnginePtr nodeEngine, uint32_t numOfProducers, SharedQueryId sharedQueryId, DecomposedQueryId decomposedQueryId, DecomposedQueryPlanVersion decomposedQueryVersion)
	public constructor for data sink More...

	SinkMedium (SinkFormatPtr sinkFormat, Runtime::NodeEnginePtr nodeEngine, uint32_t numOfProducers, SharedQueryId sharedQueryId, DecomposedQueryId decomposedQueryId, DecomposedQueryPlanVersion decomposedQueryVersion, uint64_t numberOfOrigins)
	public constructor for data sink More...

virtual void	setup ()=0
	virtual method to setup sink @Note this method will be overwritten by derived classes More...

virtual void	shutdown ()=0
	virtual method to shutdown sink @Note this method will be overwritten by derived classes More...

virtual bool	writeData (Runtime::TupleBuffer &inputBuffer, Runtime::WorkerContext &workerContext)=0
	method to write a TupleBuffer More...

SharedQueryId	getSharedQueryId () const
	get the id of the owning plan More...

DecomposedQueryId	getParentPlanId () const
	get the subplan id of the owning plan More...

DecomposedQueryPlanVersion	getParentPlanVersion () const
	get the subplan version of the owning plan More...

uint64_t	getNumberOfWrittenOutBuffers ()
	debug function for testing to get number of written buffers More...

uint64_t	getNumberOfWrittenOutTuples ()
	debug function for testing to get number of written tuples More...

virtual std::string	toString () const =0
	virtual function to get a string describing the particular sink @Note this function is overwritten by the particular data sink More...

SchemaPtr	getSchemaPtr () const
	method to return the current schema of the sink More...

std::string	getSinkFormat ()
	method to get the format as string More...

virtual SinkMediumTypes	getSinkMediumType ()=0
	method to return the type of medium More...

void	reconfigure (Runtime::ReconfigurationMessage &message, Runtime::WorkerContext &context) override

void	postReconfigurationCallback (Runtime::ReconfigurationMessage &message) override

OperatorId	getOperatorId () const
	Get operator id. More...

void	setMigrationFlag ()
	Sets that sink is used for state migration. More...

bool	isForMigration () const
	Check whether this sink is used for state migration. More...

Public Member Functions inherited from NES::Runtime::Reconfigurable
	~Reconfigurable () NES_NOEXCEPT(false) override=default

Public Member Functions inherited from NES::detail::virtual_enable_shared_from_this< Reconfigurable, false >
	~virtual_enable_shared_from_this () NES_NOEXCEPT(isNoexceptDestructible) override=default

std::shared_ptr< T1 >	shared_from_this ()

std::weak_ptr< T1 >	weak_from_this ()

Public Member Functions inherited from NES::detail::virtual_enable_shared_from_this_base< isNoexceptDestructible >
virtual	~virtual_enable_shared_from_this_base () NES_NOEXCEPT(isNoexceptDestructible)=default

Protected Attributes
SinkFormatPtr	sinkFormat

bool	schemaWritten

Runtime::NodeEnginePtr	nodeEngine

std::atomic< uint32_t >	activeProducers
	termination machinery More...

SharedQueryId	sharedQueryId

DecomposedQueryId	decomposedQueryId

DecomposedQueryPlanVersion	decomposedQueryVersion

uint64_t	numberOfOrigins

uint64_t	sentBuffer {0}

uint64_t	sentTuples {0}

std::recursive_mutex	writeMutex

bool	migration {false}

Detailed Description

Base class for all data sinks in NES.

Note: this code is not thread safe

Constructor & Destructor Documentation

◆ SinkMedium() [1/2]

NES::SinkMedium::SinkMedium	(	SinkFormatPtr	sinkFormat,
		Runtime::NodeEnginePtr	nodeEngine,
		uint32_t	numOfProducers,
		SharedQueryId	sharedQueryId,
		DecomposedQueryId	decomposedQueryId,
		DecomposedQueryPlanVersion	decomposedQueryVersion
	)

explicit

public constructor for data sink

◆ SinkMedium() [2/2]

NES::SinkMedium::SinkMedium	(	SinkFormatPtr	sinkFormat,
		Runtime::NodeEnginePtr	nodeEngine,
		uint32_t	numOfProducers,
		SharedQueryId	sharedQueryId,
		DecomposedQueryId	decomposedQueryId,
		DecomposedQueryPlanVersion	decomposedQueryVersion,
		uint64_t	numberOfOrigins
	)

explicit

public constructor for data sink

References NES_ASSERT2_FMT, and schemaWritten.

Member Function Documentation

◆ getNumberOfWrittenOutBuffers()

uint64_t NES::SinkMedium::getNumberOfWrittenOutBuffers ( )

debug function for testing to get number of written buffers

Returns: number of sent buffer

References sentBuffer, and writeMutex.

◆ getNumberOfWrittenOutTuples()

uint64_t NES::SinkMedium::getNumberOfWrittenOutTuples ( )

debug function for testing to get number of written tuples

Returns: number of sent buffer

References sentTuples, and writeMutex.

◆ getOperatorId()

OperatorId NES::SinkMedium::getOperatorId ( ) const

Get operator id.

Returns: operator id

◆ getParentPlanId()

DecomposedQueryId NES::SinkMedium::getParentPlanId ( ) const

get the subplan id of the owning plan

Returns: QuerySubPlanId

References decomposedQueryId.

◆ getParentPlanVersion()

DecomposedQueryPlanVersion NES::SinkMedium::getParentPlanVersion ( ) const

get the subplan version of the owning plan

Returns: QuerySubPlanVersion

References decomposedQueryVersion.

◆ getSchemaPtr()

SchemaPtr NES::SinkMedium::getSchemaPtr ( ) const

method to return the current schema of the sink

Returns: schema description of the sink

References sinkFormat.

Referenced by NES::MonitoringSink::writeData().

Here is the caller graph for this function:

◆ getSharedQueryId()

SharedQueryId NES::SinkMedium::getSharedQueryId ( ) const

get the id of the owning plan

Returns: sharedQueryId

References sharedQueryId.

◆ getSinkFormat()

std::string NES::SinkMedium::getSinkFormat ( )

method to get the format as string

Returns: format as string

References sinkFormat.

◆ getSinkMediumType()

virtual SinkMediumTypes NES::SinkMedium::getSinkMediumType ( )

pure virtual

method to return the type of medium

Returns: type of medium

Implemented in NES::ZmqSink, NES::Statistic::StatisticSink, NES::RawBufferSink, NES::PrintSink, NES::NullOutputSink, NES::MonitoringSink, NES::FileSink, NES::Network::NetworkSink, NES::Network::TestSink, NES::Network::TestSink, NES::CollectTestSink< Type >, and NES::TestSink.

◆ isForMigration()

bool NES::SinkMedium::isForMigration ( ) const

Check whether this sink is used for state migration.

Returns: migration flag

References migration.

◆ postReconfigurationCallback()

void NES::SinkMedium::postReconfigurationCallback ( Runtime::ReconfigurationMessage & message )

overridevirtual

Parameters

message

Reimplemented from NES::Runtime::Reconfigurable.

References activeProducers, decomposedQueryId, decomposedQueryVersion, NES::Runtime::FailEndOfStream, NES::Runtime::Failure, NES::Runtime::ReconfigurationMessage::getType(), NES::Runtime::ReconfigurationMessage::getUserData(), NES::Runtime::Graceful, NES::Runtime::HardEndOfStream, NES::Runtime::HardStop, NES::ReconfigurationMetadata::instanceOf(), NES::Runtime::Invalid, NES_ASSERT2_FMT, NES_DEBUG, NES_NOT_IMPLEMENTED, NES_WARNING, nodeEngine, NES::Runtime::Reconfigurable::postReconfigurationCallback(), NES::Runtime::ReconfigurationMarker, NES::Runtime::reconfigurationTypeToTerminationType(), NES::detail::virtual_enable_shared_from_this< Reconfigurable, false >::shared_from_this(), shutdown(), NES::Runtime::SoftEndOfStream, and toString().

Referenced by NES::Network::NetworkSink::postReconfigurationCallback().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ reconfigure()

void NES::SinkMedium::reconfigure	(	Runtime::ReconfigurationMessage &	message,
		Runtime::WorkerContext &	context
	)

overridevirtual

Parameters

message
context

Reimplemented from NES::Runtime::Reconfigurable.

References NES::Runtime::Reconfigurable::reconfigure().

Referenced by NES::Network::NetworkSink::reconfigure().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ setMigrationFlag()

void NES::SinkMedium::setMigrationFlag ( )

Sets that sink is used for state migration.

References migration.

◆ setup()

virtual void NES::SinkMedium::setup ( )

pure virtual

virtual method to setup sink @Note this method will be overwritten by derived classes

Implemented in NES::ZmqSink, NES::Statistic::StatisticSink, NES::RawBufferSink, NES::PrintSink, NES::NullOutputSink, NES::MonitoringSink, NES::FileSink, NES::Network::NetworkSink, NES::Network::TestSink, NES::Network::TestSink, NES::CollectTestSink< Type >, and NES::TestSink.

◆ shutdown()

virtual void NES::SinkMedium::shutdown ( )

pure virtual

virtual method to shutdown sink @Note this method will be overwritten by derived classes

Implemented in NES::ZmqSink, NES::Statistic::StatisticSink, NES::RawBufferSink, NES::PrintSink, NES::NullOutputSink, NES::MonitoringSink, NES::FileSink, NES::Network::NetworkSink, NES::Network::TestSink, NES::Network::TestSink, NES::CollectTestSink< Type >, and NES::TestSink.

Referenced by postReconfigurationCallback().

Here is the caller graph for this function:

◆ toString()

virtual std::string NES::SinkMedium::toString ( ) const

pure virtual

virtual function to get a string describing the particular sink @Note this function is overwritten by the particular data sink

Returns: string with name and additional information about the sink

Implemented in NES::ZmqSink, NES::Statistic::StatisticSink, NES::RawBufferSink, NES::PrintSink, NES::NullOutputSink, NES::MonitoringSink, NES::FileSink, NES::Network::NetworkSink, NES::Network::TestSink, NES::Network::TestSink, NES::CollectTestSink< Type >, and NES::TestSink.

Referenced by postReconfigurationCallback().

Here is the caller graph for this function:

◆ writeData()

virtual bool NES::SinkMedium::writeData	(	Runtime::TupleBuffer &	inputBuffer,
		Runtime::WorkerContext &	workerContext
	)

pure virtual

method to write a TupleBuffer

Parameters

a	tuple buffers pointer

Returns: bool indicating if the write was complete

Implemented in NES::Statistic::StatisticSink, NES::Network::NetworkSink, NES::CollectTestSink< Type >, and NES::TestSink.

Member Data Documentation

◆ activeProducers

std::atomic<uint32_t> NES::SinkMedium::activeProducers

protected

termination machinery

Referenced by postReconfigurationCallback().

◆ decomposedQueryId

DecomposedQueryId NES::SinkMedium::decomposedQueryId

protected

Referenced by NES::Network::NetworkSink::configureNewSinkDescriptor(), getParentPlanId(), NES::Network::NetworkSink::onEvent(), NES::Network::NetworkSink::postReconfigurationCallback(), postReconfigurationCallback(), NES::Network::NetworkSink::preSetup(), NES::Network::NetworkSink::reconfigure(), NES::Network::NetworkSink::setup(), NES::Network::NetworkSink::shutdown(), and NES::Network::NetworkSink::writeData().

◆ decomposedQueryVersion

DecomposedQueryPlanVersion NES::SinkMedium::decomposedQueryVersion

protected

Referenced by getParentPlanVersion(), NES::Network::NetworkSink::postReconfigurationCallback(), postReconfigurationCallback(), NES::Network::NetworkSink::reconfigure(), and NES::Network::NetworkSink::writeData().

◆ migration

bool NES::SinkMedium::migration {false}

protected

Referenced by isForMigration(), and setMigrationFlag().

◆ nodeEngine

Runtime::NodeEnginePtr NES::SinkMedium::nodeEngine

protected

Referenced by NES::CollectTestSink< Type >::CollectTestSink(), postReconfigurationCallback(), and NES::TestSink::TestSink().

◆ numberOfOrigins

uint64_t NES::SinkMedium::numberOfOrigins

protected

◆ schemaWritten

bool NES::SinkMedium::schemaWritten

protected

Referenced by NES::FileSink::setup(), SinkMedium(), NES::FileSink::writeData(), and NES::ZmqSink::writeData().

◆ sentBuffer

uint64_t NES::SinkMedium::sentBuffer {0}

protected

Referenced by getNumberOfWrittenOutBuffers(), and NES::ZmqSink::writeData().

◆ sentTuples

uint64_t NES::SinkMedium::sentTuples {0}

protected

Referenced by getNumberOfWrittenOutTuples().

◆ sharedQueryId

SharedQueryId NES::SinkMedium::sharedQueryId

protected

Referenced by getSharedQueryId(), NES::Network::NetworkSink::reconfigure(), NES::Network::NetworkSink::setup(), and NES::Network::NetworkSink::shutdown().

◆ sinkFormat

SinkFormatPtr NES::SinkMedium::sinkFormat

protected

Referenced by getSchemaPtr(), getSinkFormat(), NES::FileSink::setup(), NES::FileSink::toString(), NES::MonitoringSink::toString(), NES::PrintSink::toString(), NES::ZmqSink::toString(), NES::Network::NetworkSink::writeData(), NES::FileSink::writeData(), NES::PrintSink::writeData(), and NES::ZmqSink::writeData().

◆ writeMutex

std::recursive_mutex NES::SinkMedium::writeMutex

protected

Referenced by getNumberOfWrittenOutBuffers(), getNumberOfWrittenOutTuples(), NES::FileSink::writeData(), NES::MonitoringSink::writeData(), NES::PrintSink::writeData(), NES::RawBufferSink::writeData(), and NES::ZmqSink::writeData().

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

/nebulastream/nes-runtime/include/Sinks/Mediums/SinkMedium.hpp
/nebulastream/nes-runtime/src/Sinks/Mediums/SinkMedium.cpp

Public Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ SinkMedium() [1/2]

◆ SinkMedium() [2/2]

Member Function Documentation

◆ getNumberOfWrittenOutBuffers()

◆ getNumberOfWrittenOutTuples()

◆ getOperatorId()

◆ getParentPlanId()

◆ getParentPlanVersion()

◆ getSchemaPtr()

◆ getSharedQueryId()

◆ getSinkFormat()

◆ getSinkMediumType()

◆ isForMigration()

◆ postReconfigurationCallback()

◆ reconfigure()

◆ setMigrationFlag()

◆ setup()

◆ shutdown()

◆ toString()

◆ writeData()

Member Data Documentation

◆ activeProducers

◆ decomposedQueryId

◆ decomposedQueryVersion

◆ migration

◆ nodeEngine

◆ numberOfOrigins

◆ schemaWritten

◆ sentBuffer

◆ sentTuples

◆ sharedQueryId

◆ sinkFormat

◆ writeMutex