#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)

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

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 43 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 35 of file SocketWriter.cpp.

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

Member Function Documentation

◆ doWrite()

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

private

Definition at line 62 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) {
                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();
            }
        }
    }

◆ doWriteShutdown()

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

privatepure virtual

◆ onSignal()

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

privatepure virtual

◆ sendToPeer()

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

protected

Definition at line 101 of file SocketWriter.cpp.

                                                                       {
        if (!shutdownInProgress && !markShutdown) {
            if (isEnabled()) {
                if (writePuffer.empty()) {
                    resume();
                }
 
                writePuffer.insert(writePuffer.end(), chunk, chunk + 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";
        }
    }

◆ setBlockSize()

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

protected

Definition at line 97 of file SocketWriter.cpp.

                                                            {
        blockSize = writeBlockSize;
    }

◆ shutdownWrite()

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

protected

Definition at line 150 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";
            }
        }
    }

◆ signalEvent()

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

finalprivatevirtual

Implements core::DescriptorEventReceiver.

Definition at line 50 of file SocketWriter.cpp.

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

◆ streamEof()

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

protected

Definition at line 145 of file SocketWriter.cpp.

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

◆ streamToPeer()

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

protected

Definition at line 121 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;
    }

◆ 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 58 of file SocketWriter.cpp.

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

◆ writeEvent()

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

finalprivatevirtual

Implements core::eventreceiver::WriteEventReceiver.

Definition at line 46 of file SocketWriter.cpp.

                                  {
        doWrite();
    }

Member Data Documentation

◆ blockSize

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

private

Definition at line 90 of file SocketWriter.h.

◆ markShutdown

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

protected

Definition at line 75 of file SocketWriter.h.

◆ onShutdown

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

protected

Definition at line 81 of file SocketWriter.h.

◆ onStatus

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

private

Definition at line 78 of file SocketWriter.h.

◆ shutdownInProgress

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

protected

Definition at line 85 of file SocketWriter.h.

◆ source

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

private

Definition at line 88 of file SocketWriter.h.

◆ terminateTimeout

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

protected

Definition at line 93 of file SocketWriter.h.

◆ writePuffer

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

protected

Definition at line 83 of file SocketWriter.h.

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

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

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()

◆ onSignal()

◆ sendToPeer()

◆ setBlockSize()

◆ shutdownWrite()

◆ signalEvent()

◆ streamEof()

◆ streamToPeer()

◆ write()

◆ writeEvent()

Member Data Documentation

◆ blockSize

◆ markShutdown

◆ onShutdown

◆ onStatus

◆ shutdownInProgress

◆ source

◆ terminateTimeout

◆ writePuffer