#include <SocketWriter.h>

Inheritance diagram for core::socket::stream::SocketWriter:

Collaboration diagram for core::socket::stream::SocketWriter:

Public Member Functions
	SocketWriter ()=delete
Public Member Functions inherited from core::DescriptorEventReceiver
	DescriptorEventReceiver (const std::string &name, DescriptorEventPublisher &descriptorEventPublisher, const utils::Timeval &timeout=TIMEOUT::DISABLE)
int	getRegisteredFd () const
bool	isEnabled () const
bool	isSuspended () const
void	setTimeout (const utils::Timeval &timeout)
utils::Timeval	getTimeout (const utils::Timeval &currentTime) const
void	checkTimeout (const utils::Timeval &currentTime)
Public Member Functions inherited from core::EventReceiver
	EventReceiver (const std::string &name)
	EventReceiver (EventReceiver &)=delete
	EventReceiver (EventReceiver &&)=delete
EventReceiver &	operator= (EventReceiver &)=delete
EventReceiver &	operator= (EventReceiver &&)=delete
virtual void	destruct ()
void	span ()
void	relax ()
const std::string &	getName () const

Protected Member Functions
	SocketWriter (const std::string &instanceName, const std::function< void(int)> &onStatus, const utils::Timeval &timeout, std::size_t blockSize, const utils::Timeval &terminateTimeout)
std::size_t	getTotalSent () const
std::size_t	getTotalQueued () const
virtual ssize_t	write (const char *chunk, std::size_t chunkLen)
void	setBlockSize (std::size_t writeBlockSize)
void	sendToPeer (const char *chunk, std::size_t chunkLen)
bool	streamToPeer (core::pipe::Source *source)
void	streamEof ()
void	shutdownWrite (const std::function< void()> &onShutdown)
Protected Member Functions inherited from core::eventreceiver::WriteEventReceiver
	WriteEventReceiver (const std::string &name, const utils::Timeval &timeout)
virtual void	writeTimeout ()
Protected Member Functions inherited from core::DescriptorEventReceiver
bool	enable (int fd)
void	disable ()
void	suspend ()
void	resume ()
Protected Member Functions inherited from core::Observer
void	observed ()
void	unObserved ()
virtual void	unobservedEvent ()=0
	Observer ()=default
	Observer (Observer &)=delete
	Observer (Observer &&)=delete
virtual	~Observer ()
Protected Member Functions inherited from core::EventReceiver
virtual	~EventReceiver ()=default
	EventReceiver (const std::string &name)
	EventReceiver (EventReceiver &)=delete
	EventReceiver (EventReceiver &&)=delete
EventReceiver &	operator= (EventReceiver &)=delete
EventReceiver &	operator= (EventReceiver &&)=delete
virtual void	destruct ()
void	span ()
void	relax ()
const std::string &	getName () const

Protected Attributes
bool	markShutdown = false
std::function< void()>	onShutdown
std::vector< char >	writePuffer
bool	shutdownInProgress = false
utils::Timeval	terminateTimeout

Private Member Functions
void	writeEvent () final
void	signalEvent (int sigNum) final
void	doWrite ()
virtual bool	onSignal (int sigNum)=0
virtual void	doWriteShutdown (const std::function< void()> &onShutdown)=0

Private Attributes
std::function< void(int)>	onStatus
core::pipe::Source *	source = nullptr
std::size_t	blockSize = 0
std::size_t	totalQueued = 0
std::size_t	totalSent = 0

Additional Inherited Members
Static Public Member Functions inherited from core::EventReceiver
static void	atNextTick (const std::function< void(void)> &callBack)
Static Protected Member Functions inherited from core::EventReceiver
static void	atNextTick (const std::function< void(void)> &callBack)

Detailed Description

Definition at line 65 of file SocketWriter.h.

Constructor & Destructor Documentation

◆ SocketWriter() [1/2]

core::socket::stream::SocketWriter::SocketWriter ( )

delete

◆ SocketWriter() [2/2]

core::socket::stream::SocketWriter::SocketWriter	(	const std::string &	instanceName,
		const std::function< void(int)> &	onStatus,
		const utils::Timeval &	timeout,
		std::size_t	blockSize,
		const utils::Timeval &	terminateTimeout )

explicitprotected

Definition at line 56 of file SocketWriter.cpp.

        : core::eventreceiver::WriteEventReceiver(instanceName, timeout)
        , onStatus(onStatus)
        , blockSize(blockSize)
        , terminateTimeout(terminateTimeout) {
    }

References blockSize, onStatus, terminateTimeout, utils::Timeval::Timeval(), and core::eventreceiver::WriteEventReceiver::WriteEventReceiver().

Here is the call graph for this function:

Member Function Documentation

◆ doWrite()

void core::socket::stream::SocketWriter::doWrite ( )

private

Definition at line 91 of file SocketWriter.cpp.

                               {
        if (!writePuffer.empty()) {
            const std::size_t writeLen = (writePuffer.size() < blockSize) ? writePuffer.size() : blockSize;
            const ssize_t retWrite = write(writePuffer.data(), writeLen);
 
            if (retWrite > 0) {
                totalSent += static_cast<std::size_t>(retWrite);
                writePuffer.erase(writePuffer.begin(), writePuffer.begin() + retWrite);
 
                if (writePuffer.capacity() > writePuffer.size() * 2) {
                    writePuffer.shrink_to_fit();
                }
 
                if (!isSuspended()) {
                    suspend();
                }
                span();
            } else if ((errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) && isSuspended()) {
                resume();
            } else {
                onStatus(retWrite == 0 ? 0 : errno);
            }
        } else {
            if (!isSuspended()) {
                suspend();
            }
 
            if (markShutdown) {
                LOG(TRACE) << getName() << ": Shutdown restart";
                doWriteShutdown(onShutdown);
            } else if (source != nullptr) {
                source->resume();
            }
        }
    }

References blockSize, doWriteShutdown(), core::EventReceiver::getName(), core::DescriptorEventReceiver::isSuspended(), markShutdown, onShutdown, onStatus, core::DescriptorEventReceiver::resume(), core::pipe::Source::resume(), source, core::EventReceiver::span(), core::DescriptorEventReceiver::suspend(), totalSent, write(), and writePuffer.

Referenced by writeEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ doWriteShutdown()

virtual void core::socket::stream::SocketWriter::doWriteShutdown ( const std::function< void()> & onShutdown )

privatepure virtual

Implemented in core::socket::stream::SocketConnectionT< PhysicalSocketT, core::socket::stream::legacy::SocketReader, core::socket::stream::legacy::SocketWriter, ConfigT >, core::socket::stream::SocketConnectionT< PhysicalSocketT, core::socket::stream::tls::SocketReader, core::socket::stream::tls::SocketWriter, ConfigT >, and core::socket::stream::tls::SocketConnection< PhysicalSocketT, ConfigT >.

Referenced by doWrite(), and shutdownWrite().

Here is the caller graph for this function:

◆ getTotalQueued()

std::size_t core::socket::stream::SocketWriter::getTotalQueued ( ) const

protected

Definition at line 71 of file SocketWriter.cpp.

                                                 {
        return totalQueued;
    }

References totalQueued.

◆ getTotalSent()

std::size_t core::socket::stream::SocketWriter::getTotalSent ( ) const

protected

Definition at line 67 of file SocketWriter.cpp.

                                               {
        return totalSent;
    }

References totalSent.

◆ onSignal()

virtual bool core::socket::stream::SocketWriter::onSignal ( int sigNum )

privatepure virtual

Implemented in core::socket::stream::SocketConnectionT< PhysicalSocketT, core::socket::stream::legacy::SocketReader, core::socket::stream::legacy::SocketWriter, ConfigT >, and core::socket::stream::SocketConnectionT< PhysicalSocketT, core::socket::stream::tls::SocketReader, core::socket::stream::tls::SocketWriter, ConfigT >.

Referenced by signalEvent().

Here is the caller graph for this function:

◆ sendToPeer()

void core::socket::stream::SocketWriter::sendToPeer	(	const char *	chunk,
		std::size_t	chunkLen )

protected

Definition at line 131 of file SocketWriter.cpp.

                                                                       {
        if (!shutdownInProgress && !markShutdown) {
            if (isEnabled()) {
                if (writePuffer.empty()) {
                    resume();
                }
 
                writePuffer.insert(writePuffer.end(), chunk, chunk + chunkLen);
                totalQueued += chunkLen;
 
                if (source != nullptr && writePuffer.size() > 5 * blockSize) {
                    source->suspend();
                }
            } else {
                LOG(WARNING) << getName() << ": Send while not enabled";
            }
        } else {
            LOG(WARNING) << getName() << ": Send while shutdown in progress: ignoring";
        }
    }

References blockSize, core::EventReceiver::getName(), core::DescriptorEventReceiver::isEnabled(), markShutdown, core::DescriptorEventReceiver::resume(), shutdownInProgress, source, core::pipe::Source::suspend(), totalQueued, and writePuffer.

Here is the call graph for this function:

◆ setBlockSize()

void core::socket::stream::SocketWriter::setBlockSize ( std::size_t writeBlockSize )

protected

Definition at line 127 of file SocketWriter.cpp.

                                                            {
        blockSize = writeBlockSize;
    }

References blockSize.

◆ shutdownWrite()

void core::socket::stream::SocketWriter::shutdownWrite ( const std::function< void()> & onShutdown )

protected

Definition at line 181 of file SocketWriter.cpp.

                                                                          {
        if (!shutdownInProgress) {
            shutdownInProgress = true;
 
            SocketWriter::onShutdown = onShutdown;
            if (writePuffer.empty()) {
                LOG(TRACE) << getName() << ": Shutdown start";
                doWriteShutdown(onShutdown);
            } else {
                markShutdown = true;
                LOG(TRACE) << getName() << ": Shutdown delayed due to queued data";
            }
        }
    }

References doWriteShutdown(), core::EventReceiver::getName(), markShutdown, onShutdown, shutdownInProgress, and writePuffer.

Referenced by signalEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ signalEvent()

void core::socket::stream::SocketWriter::signalEvent ( int sigNum )

finalprivatevirtual

Implements core::DescriptorEventReceiver.

Definition at line 79 of file SocketWriter.cpp.

                                             {
        if (onSignal(sigNum)) {
            shutdownWrite([this]() {
                SocketWriter::disable();
            });
        }
    }

References core::DescriptorEventReceiver::disable(), onSignal(), and shutdownWrite().

Here is the call graph for this function:

◆ streamEof()

void core::socket::stream::SocketWriter::streamEof ( )

protected

Definition at line 176 of file SocketWriter.cpp.

                                 {
        LOG(TRACE) << getName() << ": Stream EOF";
        this->source = nullptr;
    }

References core::EventReceiver::getName(), and source.

Here is the call graph for this function:

◆ streamToPeer()

bool core::socket::stream::SocketWriter::streamToPeer ( core::pipe::Source * source )

protected

Definition at line 152 of file SocketWriter.cpp.

                                                          {
        bool success = false;
 
        if (!shutdownInProgress && !markShutdown) {
            if (isEnabled()) {
                success = source != nullptr;
 
                if (success) {
                    LOG(TRACE) << getName() << ": Stream started";
                } else {
                    LOG(WARNING) << getName() << ": Stream source is nullptr";
                }
            } else {
                LOG(WARNING) << getName() << ": Stream while not enabled";
            }
        } else {
            LOG(WARNING) << getName() << ": Stream while shutdown in progress";
        }
 
        this->source = source;
 
        return success;
    }

References core::EventReceiver::getName(), core::DescriptorEventReceiver::isEnabled(), markShutdown, shutdownInProgress, and source.

Here is the call graph for this function:

◆ write()

ssize_t core::socket::stream::SocketWriter::write	(	const char *	chunk,
		std::size_t	chunkLen )

protectedvirtual

Reimplemented in core::socket::stream::tls::SocketWriter.

Definition at line 87 of file SocketWriter.cpp.

                                                                     {
        return core::system::send(this->getRegisteredFd(), chunk, chunkLen, MSG_NOSIGNAL);
    }

References core::DescriptorEventReceiver::getRegisteredFd(), and core::system::send().

Referenced by doWrite(), and core::socket::stream::tls::SocketWriter::write().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ writeEvent()

void core::socket::stream::SocketWriter::writeEvent ( )

finalprivatevirtual

Implements core::eventreceiver::WriteEventReceiver.

Definition at line 75 of file SocketWriter.cpp.

                                  {
        doWrite();
    }

References doWrite().

Here is the call graph for this function:

Member Data Documentation

◆ blockSize

std::size_t core::socket::stream::SocketWriter::blockSize = 0

private

Definition at line 114 of file SocketWriter.h.

Referenced by doWrite(), sendToPeer(), setBlockSize(), and SocketWriter().

◆ markShutdown

bool core::socket::stream::SocketWriter::markShutdown = false

protected

Definition at line 99 of file SocketWriter.h.

Referenced by doWrite(), sendToPeer(), shutdownWrite(), and streamToPeer().

◆ onShutdown

std::function<void()> core::socket::stream::SocketWriter::onShutdown

protected

Definition at line 105 of file SocketWriter.h.

Referenced by doWrite(), and shutdownWrite().

◆ onStatus

std::function<void(int)> core::socket::stream::SocketWriter::onStatus

private

Definition at line 102 of file SocketWriter.h.

Referenced by doWrite(), and SocketWriter().

◆ shutdownInProgress

bool core::socket::stream::SocketWriter::shutdownInProgress = false

protected

Definition at line 109 of file SocketWriter.h.

Referenced by sendToPeer(), shutdownWrite(), and streamToPeer().

◆ source

core::pipe::Source* core::socket::stream::SocketWriter::source = nullptr

private

Definition at line 112 of file SocketWriter.h.

Referenced by doWrite(), sendToPeer(), streamEof(), and streamToPeer().

◆ terminateTimeout

utils::Timeval core::socket::stream::SocketWriter::terminateTimeout

protected

Definition at line 120 of file SocketWriter.h.

Referenced by SocketWriter().

◆ totalQueued

std::size_t core::socket::stream::SocketWriter::totalQueued = 0

private

Definition at line 116 of file SocketWriter.h.

Referenced by getTotalQueued(), and sendToPeer().

◆ totalSent

std::size_t core::socket::stream::SocketWriter::totalSent = 0

private

Definition at line 117 of file SocketWriter.h.

Referenced by doWrite(), and getTotalSent().

◆ writePuffer

std::vector<char> core::socket::stream::SocketWriter::writePuffer

protected

Definition at line 107 of file SocketWriter.h.

Referenced by doWrite(), sendToPeer(), and shutdownWrite().

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

core/socket/stream/SocketWriter.h
core/socket/stream/SocketWriter.cpp

SNode.C

Public Member Functions

Protected Member Functions

Protected Attributes

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ SocketWriter() [1/2]

◆ SocketWriter() [2/2]

Member Function Documentation

◆ doWrite()

◆ doWriteShutdown()

◆ getTotalQueued()

◆ getTotalSent()

◆ onSignal()

◆ sendToPeer()

◆ setBlockSize()

◆ shutdownWrite()

◆ signalEvent()

◆ streamEof()

◆ streamToPeer()

◆ write()

◆ writeEvent()

Member Data Documentation

◆ blockSize

◆ markShutdown

◆ onShutdown

◆ onStatus

◆ shutdownInProgress

◆ source

◆ terminateTimeout

◆ totalQueued

◆ totalSent

◆ writePuffer