#include <tsReactiveStream.h>

Inheritance diagram for ts::ReactiveStream:

[legend]

Collaboration diagram for ts::ReactiveStream:

[legend]

Public Member Functions
	ReactiveStream (Reactor &reactor, Device &device, StreamInterface &stream)
virtual void	cancelReadWriteStream (bool silent=false)
Device &	device ()
bool	muteReport (bool mute)
Reactor &	reactor ()
virtual Report &	report () const override
Report *	setReport (Report *report)
ReporterBase *	setReport (ReporterBase *delegate)
bool	signalQueuedOperations ()
virtual bool	startReadStream (ReactiveStreamHandlerInterface *handler, size_t buffer_size=Device::DEFAULT_RECEIVE_BUFFER_SIZE, const ObjectPtr &user_data=ObjectPtr())
virtual bool	startWriteStream (ReactiveStreamHandlerInterface handler, const void data, size_t size, const ObjectPtr &user_data=ObjectPtr())
StreamInterface &	stream ()

Static Public Member Functions
static int	SilentLevel (bool silent, int default_severity=Severity::Error)

Protected Types
using	IOQueue = std::list<std::shared_ptr<IOSB>>
using	IOSB = Device::IOSB
	IOSB shortcut fpr subclasses.
using	IOSet = std::set<std::shared_ptr<IOSB>>

Protected Member Functions
bool	activateAsynchronousIO ()
bool	activateReadReady ()
bool	activateWriteReady ()
bool	cancelAndWaitAsynchronousIO (Device::IOSB &iosb, bool silent)
void	cancelAsynchronousIO (bool silent)
template<class REQUEST> requires std::derived_from<REQUEST, ts::Object>
void	cancelQueue (IOQueue &inqueue, IOQueue &outqueue)
bool	createSignalQueuedOperations ()
void	deactivateAll (bool silent)
void	deactivateAsynchronousIO (bool silent)
void	deactivateQueuedOperations (bool silent)
void	deactivateReadReady (bool silent)
void	deactivateWriteReady (bool silent)
bool	enqueueCompletedIO (const std::shared_ptr< IOSB > &iosb, bool in_reactor)
virtual void	handleAsynchronousIO (Reactor &, EventId, IOSB &, size_t) override
virtual void	handleBroadcastEvent (Reactor &reactor, int error_code, const ObjectPtr &user_data)
virtual void	handleProcessTermination (Reactor &reactor, EventId id, int pid)
virtual void	handleReadReady (Reactor &, EventId, int) override
virtual void	handleTimer (Reactor &reactor, EventId id)
virtual void	handleUserEvent (Reactor &, EventId) override
virtual void	handleWriteReady (Reactor &, EventId, int) override
virtual bool	hasPendingIO ()
virtual bool	needsWriteReady ()
virtual int	processCloseWriteStream (bool silent)
virtual void	processQueuedOperations () override
void	processReceiveBuffer (ByteBlockPtr &data, ReactiveInputControl &control, ReactiveStreamHandlerInterface *handler, int error_code, const ObjectPtr &user_data)
std::shared_ptr< IOSB >	removeFromQueue (IOQueue &queue, IOSB *iosb)
virtual bool	startCloseWriteStream (ReactiveStreamHandlerInterface *handler, bool silent, const ObjectPtr &user_data)
virtual bool	tryCompletedIO (const std::shared_ptr< IOSB > &iosb)
bool	uncheckedSignalQueuedOperations ()

Detailed Description

Using non-blocking devices with StreamInterface in a Reactor environment.

The class ReactiveStream is a wrapper around a stream device class to handle reactive I/O. Typically, the device class is a subclass of Device which implements StreamInterface.

The stream device is a separate object. It is initialized and configured by the application. The application shall not directly call writeStream() or readStream() on this device and delegate these operations to startWriteStream() and startReadStream() in class ReactiveStream.

Member Typedef Documentation

◆ IOQueue

using ts::ReactiveDevice::IOQueue = std::list<std::shared_ptr<IOSB>>

protectedinherited

Queues of I/O requests are queues of shared_ptr to IOSB. This is typically used with immediate I/O where we must process requests in order. Send and receive requests are structures which are stored in the react_data of the IOSB.

◆ IOSet

using ts::ReactiveDevice::IOSet = std::set<std::shared_ptr<IOSB>>

protectedinherited

Unordered set of I/O requests, set of shared_ptr to IOSB. This is typically used with asynchronous I/O. The ordering is enforced because I/O are started in order of calls from applications. The completion processing is likely the same, but driven by the system I/O Completion Ports and we must not assume any order. Send and receive requests are structures which are stored in the react_data of the IOSB.

Constructor & Destructor Documentation

◆ ReactiveStream()

ts::ReactiveStream::ReactiveStream	(	Reactor &	reactor,
		Device &	device,
		StreamInterface &	stream )

Constructor.

Parameters

[in,out]	reactor	Associated reactor. The reactor object must remain valid as long as this object is valid.
[in,out]	device	Associated non-blocking device. The device object must remain valid as long as this object is valid.
[in,out]	stream	Associated stream interface. Typically, the device class is a subclass of Device which implements StreamInterface. Therefore, device and stream are two views of the same object. That device object must remain valid as long as this instance of ReactiveStream exists.

Member Function Documentation

◆ stream()

StreamInterface & ts::ReactiveStream::stream ( )

inline

Get a reference to the associated stream. Typically, device() and stream() return two views of the same object.

Returns: A reference to the associated stream.

◆ startReadStream()

virtual bool ts::ReactiveStream::startReadStream	(	ReactiveStreamHandlerInterface *	handler,
		size_t	buffer_size = Device::DEFAULT_RECEIVE_BUFFER_SIZE,
		const ObjectPtr &	user_data = ObjectPtr() )

virtual

Start the operation of reading data from the stream. Reading operation is permanent and handler is called whenever incoming data are available.

Parameters

[in]	handler	Handler class to call each time data are received. The method handleReadStream() will be called for each new chunk of data. Cannot be null. If a previous receive handler was registered, it is replaced.
[in]	buffer_size	Size of input buffers to receive data.
[in]	user_data	A shared pointer which will be passed unmodified to handler.

Returns: True on success, false on error. Success means that the I/O was successfully started. The final status of the I/O will be transmitted in the handler.

Reimplemented in ts::ReactiveTLSConnection.

◆ startWriteStream()

virtual bool ts::ReactiveStream::startWriteStream	(	ReactiveStreamHandlerInterface *	handler,
		const void *	data,
		size_t	size,
		const ObjectPtr &	user_data = ObjectPtr() )

virtual

Start the operation of writing data to the stream.

Parameters

[in]	handler	Handler class to call when the send operation completes. The method handleWriteStream() will be called. If nullptr, no handler is called.
[in]	data	Address of the data to write. The corresponding memory area must remain valid until the completion or cancelation of the send operation.
[in]	size	Size in bytes of the data to send.
[in]	user_data	A shared pointer which will be passed unmodified to handler.

Returns: True on success, false on error. Success means that the I/O was successfully started. The final status of the I/O will be transmitted in the handler.

Reimplemented in ts::ReactiveTLSConnection.

◆ cancelReadWriteStream()

virtual void ts::ReactiveStream::cancelReadWriteStream ( bool silent = false )

virtual

Cancel any pending read or write operation on this stream. If a repeated read operation is in progress, the repetition is canceled as well.

Parameters

[in] silent If true, do not report errors through the logger.

Reimplemented in ts::ReactiveTLSConnection.

◆ hasPendingIO()

virtual bool ts::ReactiveStream::hasPendingIO ( )

protectedvirtual

Check if some I/O operations are in progress.

Returns: True if there is any pending operation or permanent read operation in progress. False otherwise.

◆ needsWriteReady()

virtual bool ts::ReactiveStream::needsWriteReady ( )

protectedvirtual

With immediate I/O, check if we need to be notified of write-ready conditions. If overriden by a subclass, the method must also call its superclass counterpart.

Returns: True if immediate write-ready notifications are required. False otherwise.

◆ startCloseWriteStream()

virtual bool ts::ReactiveStream::startCloseWriteStream	(	ReactiveStreamHandlerInterface *	handler,
		bool	silent,
		const ObjectPtr &	user_data )

protectedvirtual

Start closing the write direction of the stream. The request is enqueued and will be processed by calling processCloseWriteStream() after all previous send requests are completed. This feature can be used by subclasses for which "closing the write direction of the stream" means something (e.g. true for TCP socket, meaningless for files).

Parameters

[in]	handler	Handler class to call when the close-writer operation completes. The method handleWriteStream() will be called with its parameter error_code containing SYS_EOF. If nullptr, no handler is called.
[in]	silent	If true, do not report errors through the logger.
[in]	user_data	A shared pointer which will be passed unmodified to handler.

Returns: True on success, false on error.

◆ processCloseWriteStream()

virtual int ts::ReactiveStream::processCloseWriteStream ( bool silent )

protectedvirtual

Implement the close of the write direction of the stream. Must be implemented by subclasses which use startCloseWriteStream(). The default implementation does nothing. This method is always called in the context of a Reactor handler, never as an indirect call from the application.

Parameters

[in] silent If true, do not report errors through the logger.

Returns: A system-specific error code. SYS_SUCCESS in case of success.

Reimplemented in ts::ReactiveTCPConnection.

◆ enqueueCompletedIO()

bool ts::ReactiveStream::enqueueCompletedIO	(	const std::shared_ptr< IOSB > &	iosb,
		bool	in_reactor )

protected

Enqueue an IOSB in the completed I/O queue. This method can be called by subclasses which need to process specific I/O completion, other than read or write on the stream.

Parameters

[in]	iosb	Shared pointer to an IOSB. Specific data can be saved in iosb->react_data.
[in]	in_reactor	Set to true if we are in a reactor context, meaning directly called from a reactor handler, not from the application.

Returns: True on success, false on error.

◆ tryCompletedIO()

virtual bool ts::ReactiveStream::tryCompletedIO ( const std::shared_ptr< IOSB > & iosb )

protectedvirtual

Try to interpret an IOSB as a valid completed I/O request and process the completion. If overriden by a subclass, the method must also call its superclass counterpart.

Parameters

[in] iosb Shared pointer to an IOSB. Specific data can be saved in iosb->react_data.

Returns: True if the iosb contained a recognized I/O completion request. False if the request is unknown.

Reimplemented in ts::ReactiveTCPConnection.

◆ processReceiveBuffer()

void ts::ReactiveStream::processReceiveBuffer	(	ByteBlockPtr &	data,
		ReactiveInputControl &	control,
		ReactiveStreamHandlerInterface *	handler,
		int	error_code,
		const ObjectPtr &	user_data )

protected

Invoke the receive handler as many times as possible on a data buffer.

Parameters

[in,out]	data	Data buffer containing the received data. On output, data which are processed by the handler are removed. The buffer may be reallocated if all data were processed.
[in,out]	control	Input control. On input, this is the previously returned value from the handler. Modified by the handler.
[in]	handler	Application handler.
[in]	error_code	Receive error code. If not success, the handler is called exactly once.
[in]	user_data	User data for the handler.

◆ processQueuedOperations()

virtual void ts::ReactiveStream::processQueuedOperations ( )

overrideprotectedvirtual

This virtual method processes operations in the context of a Reactor handler. This is dedicated to operations which must be serialized from an application perspective. These operations are typically queued when triggered from a method which is called by the application. When the reactor processes events, we are sure that the application is not executing a handler. The default implementation does nothing. A subclass should override it if it calls signalQueuedOperations().

Reimplemented from ts::ReactiveBase.

Reimplemented in ts::ReactiveTCPConnection, and ts::ReactiveTLSConnection.

◆ handleReadReady()

virtual void ts::ReactiveStream::handleReadReady	(	Reactor &	reactor,
		EventId	id,
		int	error_code )

overrideprotectedvirtual

Handle a read-ready event in a Reactor. This handler is only invoked in the immediate I/O model.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the event which was signaled.
[in]	error_code	System-specific error code, zero on success, SYS_ERROR in case of unknown error.

Reimplemented from ts::ReactorHandlerInterface.

◆ handleWriteReady()

virtual void ts::ReactiveStream::handleWriteReady	(	Reactor &	reactor,
		EventId	id,
		int	error_code )

overrideprotectedvirtual

Handle a write-ready event in a Reactor. This handler is only invoked in the immediate I/O model.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the event which was signaled.
[in]	error_code	System-specific error code, zero on success, SYS_ERROR in case of unknown error.

Reimplemented from ts::ReactorHandlerInterface.

Reimplemented in ts::ReactiveTCPConnection.

◆ handleAsynchronousIO()

virtual void ts::ReactiveStream::handleAsynchronousIO	(	Reactor &	reactor,
		EventId	id,
		IOSB &	iosb,
		size_t	io_size )

overrideprotectedvirtual

Handle an asynchronous I/O completion event in a Reactor. This handler is only invoked in the asynchronous I/O model.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the event which was signaled.
[in,out]	iosb	IOSB structure which was used when the asynchronous I/O was started. A system-specific error code is in iosb, SYS_CANCELED if the I/O was canceled before completion.
[in]	io_size	Size of the I/O in bytes.

Reimplemented from ts::ReactorHandlerInterface.

Reimplemented in ts::ReactiveTCPConnection.

◆ device()

Device & ts::ReactiveDevice::device ( )

inlineinherited

Get a reference to the associated non-blocking device.

Returns: A reference to the associated non-blocking device.

◆ removeFromQueue()

std::shared_ptr< IOSB > ts::ReactiveDevice::removeFromQueue	(	IOQueue &	queue,
		IOSB *	iosb )

protectedinherited

Search and remove a shared_ptr to IOSB, based on an IOSB address. Search from the front (end) of the queue since a completed I/O is likely on the front.

Parameters

[in,out]	queue	The queue from which to remove iosb.
[in]	iosb	Standard pointer to an IOSB to search and remove.

Returns: The removed shared_ptr to IOSB, or a null pointer if iosb is not found.

◆ cancelQueue()

template<class REQUEST>
requires std::derived_from<REQUEST, ts::Object>

void ts::ReactiveDevice::cancelQueue	(	IOQueue &	inqueue,
		IOQueue &	outqueue )

protectedinherited

Transfer all requests from one queue to another and mark all I/O as canceled.

Template Parameters

REQUEST The subclass of Object which is set in react_data of all requests in inqueue.

Parameters

[in,out]	inqueue	The queue from which all requests are removed.
[in,out]	outqueue	The queue which receives all canceled requests.

◆ activateReadReady()

bool ts::ReactiveDevice::activateReadReady ( )

protectedinherited

Activate read-ready notification for immediate I/O.

Returns: True on success, false on error.

◆ deactivateReadReady()

void ts::ReactiveDevice::deactivateReadReady ( bool silent )

protectedinherited

Deactivate read-ready notification for immediate I/O.

Parameters

[in] silent If true, do not report errors through the logger.

◆ activateWriteReady()

bool ts::ReactiveDevice::activateWriteReady ( )

protectedinherited

Activate write-ready notification for immediate I/O.

Returns: True on success, false on error.

◆ deactivateWriteReady()

void ts::ReactiveDevice::deactivateWriteReady ( bool silent )

protectedinherited

Deactivate write-ready notification for immediate I/O.

Parameters

[in] silent If true, do not report errors through the logger.

◆ activateAsynchronousIO()

bool ts::ReactiveDevice::activateAsynchronousIO ( )

protectedinherited

Activate notification for asynchronous I/O.

Returns: True on success, false on error.

◆ deactivateAsynchronousIO()

void ts::ReactiveDevice::deactivateAsynchronousIO ( bool silent )

protectedinherited

Deactivate notification for asynchronous I/O.

Parameters

[in] silent If true, do not report errors through the logger.

◆ cancelAsynchronousIO()

void ts::ReactiveDevice::cancelAsynchronousIO ( bool silent )

protectedinherited

Cancel all asynchronous I/O in progress. The cancelation occurs in the background and end of canceled asynchronous I/O will be notified.

Parameters

[in] silent If true, do not report errors through the logger.

◆ cancelAndWaitAsynchronousIO()

bool ts::ReactiveDevice::cancelAndWaitAsynchronousIO	(	Device::IOSB &	iosb,
		bool	silent )

protectedinherited

Cancel one specific pending asynchronous I/O and wait for its completion. Warning: This is a blocking call. It shall be used in case of trouble only.

Parameters

[in,out]	iosb	The asynchronous I/O status block.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ deactivateAll()

void ts::ReactiveDevice::deactivateAll ( bool silent )

protectedinherited

Deactivate all registrations for immediate and asynchronous I/O.

Parameters

[in] silent If true, do not report errors through the logger.

◆ reactor()

Reactor & ts::ReactiveBase::reactor ( )

inlineinherited

Get a reference to the associated reactor.

Returns: A reference to the associated reactor.

◆ signalQueuedOperations()

bool ts::ReactiveBase::signalQueuedOperations ( )

inherited

Trigger the execution of processQueuedOperations() in the context of a Reactor handler. Create if necessary and then signal a dedicated user event.

Returns: True on success, false on error.

◆ deactivateQueuedOperations()

void ts::ReactiveBase::deactivateQueuedOperations ( bool silent )

protectedinherited

Deactivate the execution of processQueuedOperations() in the context of a Reactor handler. Deactivate and delete the dedicated user event.

Parameters

[in] silent If true, do not report errors through the logger.

◆ createSignalQueuedOperations()

bool ts::ReactiveBase::createSignalQueuedOperations ( )

protectedinherited

Create, if necessary, the dedicated user event for signalQueuedOperations(). Useless if signalQueuedOperations() is used. Only required with use of uncheckedSignalQueuedOperations().

Returns: True on success, false on error.

◆ uncheckedSignalQueuedOperations()

bool ts::ReactiveBase::uncheckedSignalQueuedOperations ( )

protectedinherited

Trigger the execution of processQueuedOperations() from another thread. The event must have been previously created, either using createSignalQueuedOperations() or signalQueuedOperations().

Returns: True on success, false on error.

◆ handleUserEvent()

virtual void ts::ReactiveBase::handleUserEvent	(	Reactor &	reactor,
		EventId	id )

overrideprotectedvirtualinherited

Handle a user-defined event in a Reactor.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the event which was signaled.

Reimplemented from ts::ReactorHandlerInterface.

◆ report()

virtual Report & ts::ReporterBase::report ( ) const

overridevirtualinherited

Access the Report which is associated with this object. Can be called from another thread only if the Report object is thread-safe.

Returns: A reference to the associated report.

Implements ts::ReporterInterface.

◆ setReport() [1/2]

Report * ts::ReporterBase::setReport ( Report * report )

inherited

Associate this object with another Report to log errors.

Parameters

[in] report Where to report errors. The report object must remain valid as long as this object exists or setReport() is used with another Report object. If report is null, log messages are discarded.

Returns: The address of the previous Report object or a null pointer if there was none.

◆ setReport() [2/2]

ReporterBase * ts::ReporterBase::setReport ( ReporterBase * delegate )

inherited

Associate this object with another ReporterBase to log errors.

Parameters

[in] delegate Use the report of another ReporterBase. If delegate is null, the previous explicit Report is used..

Returns: The address of the previous ReporterBase object or a null pointer if there was none.

◆ muteReport()

bool ts::ReporterBase::muteReport ( bool mute )

inherited

Temporarily mute the associated report.

Parameters

[in] mute It true, report() will return a null report (log messages are discarded), until muteReport() is invoked again with mute set to false.

Returns: Previous state of the mute field.

◆ SilentLevel()

int ts::ReporterBase::SilentLevel	(	bool	silent,
		int	default_severity = Severity::Error )

inlinestaticinherited

Compute a log severity level from a "silent" parameter. Some subclass methods have a "silent" parameter to avoid reporting errors which may be insignificant, typically when closing a device after an error, in which case the close operation may produce other errors if the previous error left the device in an inconsistent state. While those errors should not be displayed as errors, we still display them at debug level.

Parameters

[in]	silent	If true, do not report errors, report debug messages instead.
[in]	default_severity	Default severity, in non-silent mode (error by default).

Returns: Error when silent is false, Debug otherwise.

◆ handleTimer()

virtual void ts::ReactorHandlerInterface::handleTimer	(	Reactor &	reactor,
		EventId	id )

virtualinherited

Handle a timer in a Reactor.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the timer which expires.

◆ handleBroadcastEvent()

virtual void ts::ReactorHandlerInterface::handleBroadcastEvent	(	Reactor &	reactor,
		int	error_code,
		const ObjectPtr &	user_data )

virtualinherited

Handle a broadcast event in a Reactor. A broadcast event is sent to all currently registered events in the reactor.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	error_code	Application-specific error code which was passed to Reactor::signalBroadcastEvent().
[in]	user_data	The user-data shared pointer which was passed to Reactor::signalBroadcastEvent().

◆ handleProcessTermination()

virtual void ts::ReactorHandlerInterface::handleProcessTermination	(	Reactor &	reactor,
		EventId	id,
		int	pid )

virtualinherited

Handle a process termination event in a Reactor. This handler is invoked when the process is no longer there. It is possible that the process was already terminated for a while. It is even possible that the process never really started. There is no portable way to get the process termination status.

Parameters

[in,out]	reactor	Reactor into which the handler is invoked.
[in]	id	Id of the event which was signaled.
[in]	pid	Process id of the terminated process.

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

tsReactiveStream.h

Public Member Functions

Static Public Member Functions

Protected Types

Protected Member Functions

Detailed Description

Member Typedef Documentation

◆ IOQueue

◆ IOSet

Constructor & Destructor Documentation

◆ ReactiveStream()

Member Function Documentation

◆ stream()

◆ startReadStream()

◆ startWriteStream()

◆ cancelReadWriteStream()

◆ hasPendingIO()

◆ needsWriteReady()

◆ startCloseWriteStream()

◆ processCloseWriteStream()

◆ enqueueCompletedIO()

◆ tryCompletedIO()

◆ processReceiveBuffer()

◆ processQueuedOperations()

◆ handleReadReady()

◆ handleWriteReady()

◆ handleAsynchronousIO()

◆ device()

◆ removeFromQueue()

◆ cancelQueue()

◆ activateReadReady()

◆ deactivateReadReady()

◆ activateWriteReady()

◆ deactivateWriteReady()

◆ activateAsynchronousIO()

◆ deactivateAsynchronousIO()

◆ cancelAsynchronousIO()

◆ cancelAndWaitAsynchronousIO()

◆ deactivateAll()

◆ reactor()

◆ signalQueuedOperations()

◆ deactivateQueuedOperations()

◆ createSignalQueuedOperations()

◆ uncheckedSignalQueuedOperations()

◆ handleUserEvent()

◆ report()

◆ setReport() [1/2]

◆ setReport() [2/2]

◆ muteReport()

◆ SilentLevel()

◆ handleTimer()

◆ handleBroadcastEvent()

◆ handleProcessTermination()