web::http::client::tools::EventSourceT< Client > Class Template Reference

#include <EventSource.h>

Inheritance diagram for web::http::client::tools::EventSourceT< Client >:

Collaboration diagram for web::http::client::tools::EventSourceT< Client >:

Classes
struct	SharedConfig

Public Types
using	MasterRequest = typename Client::MasterRequest
using	Request = typename Client::Request
using	Response = typename Client::Response
using	SocketConnection = typename Client::SocketConnection
using	SocketAddress = typename SocketConnection::SocketAddress
using	Config = typename Client::Config
Public Types inherited from web::http::client::tools::EventSource
enum class	ReadyState : int { CONNECTING = 0 , OPEN = 1 , CLOSED = 2 }

Public Member Functions
	EventSourceT (const EventSourceT &)=delete
EventSourceT &	operator= (const EventSourceT &)=delete
	EventSourceT (EventSourceT &&) noexcept=delete
EventSourceT &	operator= (EventSourceT &&) noexcept=delete
void	close () override
Public Member Functions inherited from web::http::client::tools::EventSource
virtual	~EventSource ()
EventSource *	onMessage (std::function< void(const MessageEvent &)> messageCallback)
EventSource *	addEventListener (const std::string &key, std::function< void(const MessageEvent &)> eventListener)
EventSource *	removeEventListeners (const std::string &type)
EventSource *	onOpen (std::function< void()> onOpen)
EventSource *	onError (std::function< void()> onError)
EventSource::ReadyState	readyState () const
const std::string &	lastEventId () const
uint32_t	retry () const
EventSource *	retry (uint32_t retry)

Protected Member Functions
	EventSourceT ()
void	init (const std::string &scheme, const SocketAddress &socketAddress, const std::string &path)
Protected Member Functions inherited from web::http::client::tools::EventSource
	EventSource ()

Static Private Member Functions
static bool	digits (std::string_view maybeDigitsAsString)
static uint32_t	parseU32 (std::string_view uint32AsString)
static void	deliverMessage (const std::shared_ptr< SharedState > &sharedState, const std::string &evType, const std::string &payload, const std::string &evId)
static void	dispatch (const std::shared_ptr< SharedState > &sharedState)
static bool	processLine (const std::shared_ptr< SharedState > &sharedState, const std::shared_ptr< SharedConfig > &sharedConfig, std::string_view line)
static bool	parse (const std::shared_ptr< SharedState > &sharedState, const std::shared_ptr< SharedConfig > &sharedConfig)

Private Attributes
std::shared_ptr< Client >	client
SocketConnection *	socketConnection = nullptr
std::shared_ptr< SharedConfig >	sharedConfig

Additional Inherited Members
Protected Attributes inherited from web::http::client::tools::EventSource
std::shared_ptr< SharedState >	sharedState

Detailed Description

template<typename Client>
class web::http::client::tools::EventSourceT< Client >

Definition at line 121 of file EventSource.h.

Member Typedef Documentation

Config

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::Config = typename Client::Config

Definition at line 130 of file EventSource.h.

MasterRequest

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::MasterRequest = typename Client::MasterRequest

Definition at line 125 of file EventSource.h.

Request

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::Request = typename Client::Request

Definition at line 126 of file EventSource.h.

Response

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::Response = typename Client::Response

Definition at line 127 of file EventSource.h.

SocketAddress

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::SocketAddress = typename SocketConnection::SocketAddress

Definition at line 129 of file EventSource.h.

SocketConnection

template<typename Client>

using web::http::client::tools::EventSourceT< Client >::SocketConnection = typename Client::SocketConnection

Definition at line 128 of file EventSource.h.

Constructor & Destructor Documentation

EventSourceT() [1/3]

template<typename Client>

web::http::client::tools::EventSourceT< Client >::EventSourceT ( const EventSourceT< Client > & )

delete

EventSourceT() [2/3]

template<typename Client>

web::http::client::tools::EventSourceT< Client >::EventSourceT ( EventSourceT< Client > && )

deletenoexcept

EventSourceT() [3/3]

template<typename Client>

web::http::client::tools::EventSourceT< Client >::EventSourceT ( )

inlineprotected

Definition at line 283 of file EventSource.h.

            : sharedConfig(std::make_shared<SharedConfig>()) {
        }

Member Function Documentation

close()

template<typename Client>

void web::http::client::tools::EventSourceT< Client >::close ( )

inlineoverridevirtual

Implements web::http::client::tools::EventSource.

Definition at line 429 of file EventSource.h.

                              {
            sharedState->ready = ReadyState::CLOSED;
 
            sharedConfig->config->setReconnect(false);
            sharedConfig->config->setRetry(false);
 
            if (socketConnection != nullptr) {
                socketConnection->shutdownWrite(true);
            }
        }

deliverMessage()

template<typename Client>

void web::http::client::tools::EventSourceT< Client >::deliverMessage ( const std::shared_ptr< SharedState > & sharedState, const std::string & evType, const std::string & payload, const std::string & evId )

inlinestaticprivate

Definition at line 160 of file EventSource.h.

                                                          {
            const MessageEvent e{evType, payload, evId, sharedState->origin};
 
            if (auto it = sharedState->onEventListener.find(evType); it != sharedState->onEventListener.end()) {
                for (const auto& onEventListener : it->second) {
                    onEventListener(e);
                }
            }
 
            if (evType == "message") {
                for (const auto& onMessageListener : sharedState->onMessageListener) {
                    onMessageListener(e);
                }
            }
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::dispatch().

Here is the caller graph for this function:

digits()

template<typename Client>

bool web::http::client::tools::EventSourceT< Client >::digits ( std::string_view maybeDigitsAsString )

inlinestaticprivate

Definition at line 137 of file EventSource.h.

                                                               {
            for (const char maybeDigit : maybeDigitsAsString) {
                if (std::isdigit(static_cast<unsigned char>(maybeDigit)) == 0) {
                    return false;
                }
            }
 
            return !maybeDigitsAsString.empty();
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::processLine().

Here is the caller graph for this function:

dispatch()

template<typename Client>

void web::http::client::tools::EventSourceT< Client >::dispatch ( const std::shared_ptr< SharedState > & sharedState )

inlinestaticprivate

Definition at line 179 of file EventSource.h.

                                                                            {
            sharedState->lastId = sharedState->idBuf;
            if (sharedState->data.empty()) {
                sharedState->type.clear();
                return;
            }
            if (!sharedState->data.empty() && sharedState->data.back() == '\n') {
                sharedState->data.pop_back();
            }
 
            const std::string evType = sharedState->type.empty() ? "message" : std::exchange(sharedState->type, std::string{});
            const std::string payload = std::exchange(sharedState->data, std::string{});
            const std::string evId = sharedState->lastId;
 
            deliverMessage(sharedState, evType, payload, evId);
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::parse().

Here is the caller graph for this function:

init()

template<typename Client>

void web::http::client::tools::EventSourceT< Client >::init ( const std::string & scheme, const SocketAddress & socketAddress, const std::string & path )

inlineprotected

Definition at line 287 of file EventSource.h.

                                                                                                      {
            sharedState->path = path;
            sharedState->origin = scheme + "://" + socketAddress.toString(false);
 
            LOG(TRACE) << "Origin: " << sharedState->origin;
            LOG(TRACE) << "  Path: " << sharedState->path;
 
            const std::weak_ptr<EventSourceT> eventSourceWeak = this->weak_from_this();
 
            client = std::make_shared<Client>(
                [eventSourceWeak](SocketConnection* socketConnection) {
                    LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnConnect";
 
                    if (const std::shared_ptr<EventSourceT> eventStream = eventSourceWeak.lock()) {
                        eventStream->socketConnection = socketConnection;
                    }
                },
                [](SocketConnection* socketConnection) {
                    LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnConnected";
                },
                [eventSourceWeak, sharedState = this->sharedState, sharedConfig = this->sharedConfig](SocketConnection* socketConnection) {
                    LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnDisconnect";
 
                    if (const std::shared_ptr<EventSourceT> eventSource = eventSourceWeak.lock()) {
                        eventSource->socketConnection = nullptr;
                    }
 
                    if (sharedState->ready != ReadyState::CLOSED) {
                        if (auto it = sharedState->onEventListener.find("error"); it != sharedState->onEventListener.end()) {
                            EventSource::MessageEvent e{"error", "", sharedState->lastId, sharedState->origin};
 
                            for (auto& onError : it->second) {
                                onError(e);
                            }
                        }
 
                        for (const auto& onError : sharedState->onErrorListener) {
                            onError();
                        }
 
                        sharedState->ready = ReadyState::CONNECTING;
                        sharedState->data.clear();
                        sharedState->type.clear();
                        sharedState->idBuf.clear();
                        sharedState->pending.clear();
 
                        sharedConfig->config->setReconnectTime(sharedState->retry / 1000.);
                        sharedConfig->config->setRetryTimeout(sharedState->retry / 1000.);
                    }
                },
                [eventSourceWeak, sharedState = this->sharedState, sharedConfig = this->sharedConfig](
                    const std::shared_ptr<MasterRequest>& masterRequest) {
                    const std::string connectionName = masterRequest->getSocketContext()->getSocketConnection()->getConnectionName();
 
                    LOG(DEBUG) << connectionName << " EventSource: OnRequestStart";
 
                    if (!sharedState->lastId.empty()) {
                        masterRequest->set("Last-Event-ID", sharedState->lastId);
                    }
 
                    if (!masterRequest->requestEventSource(
                            sharedState->path,
                            [masterRequestWeak = std::weak_ptr<MasterRequest>(masterRequest), sharedState, sharedConfig, connectionName]()
                                -> std::size_t {
                                std::size_t consumed = 0;
 
                                if (const std::shared_ptr<MasterRequest> masterRequest = masterRequestWeak.lock()) {
                                    char buf[16384];
                                    consumed = masterRequest->getSocketContext()->readFromPeer(buf, sizeof(buf));
 
                                    sharedState->pending.append(buf, consumed);
 
                                    if (!parse(sharedState, sharedConfig)) {
                                        masterRequest->getSocketContext()->shutdownWrite(true);
                                    }
                                } else {
                                    LOG(DEBUG) << connectionName << ": server-sent event: server disconnect";
                                }
 
                                return consumed;
                            },
                            [sharedState, sharedConfig, connectionName]() {
                                LOG(DEBUG) << connectionName << ": server-sent event stream start";
 
                                sharedState->ready = ReadyState::OPEN;
 
                                sharedConfig->config->setReconnectTime(sharedState->retry / 1000.);
                                sharedConfig->config->setRetryTimeout(sharedState->retry / 1000.);
 
                                if (auto it = sharedState->onEventListener.find("open"); it != sharedState->onEventListener.end()) {
                                    EventSource::MessageEvent e{"open", "", sharedState->lastId, sharedState->origin};
 
                                    for (auto& onOpen : it->second) {
                                        onOpen(e);
                                    }
                                }
 
                                for (const auto& onOpen : sharedState->onOpenListener) {
                                    onOpen();
                                }
                            },
                            [sharedState, connectionName]() {
                                LOG(DEBUG) << connectionName
                                           << ": not an server-sent event endpoint: " << sharedState->origin + sharedState->path;
                            })) {
                        if (const std::shared_ptr<EventSourceT> eventSource = eventSourceWeak.lock()) {
                            eventSource->socketConnection->close();
                        }
                    }
                },
                [](const std::shared_ptr<Request>& req) {
                    LOG(DEBUG) << req->getSocketContext()->getSocketConnection()->getConnectionName() << " EventSource: OnRequestEnd";
                });
 
            client->getConfig().Remote::setSocketAddress(socketAddress);
            client->getConfig().setReconnect();
            client->getConfig().setRetry();
            client->getConfig().setRetryBase(1);
 
            sharedConfig->config = &client->getConfig();
 
            client->connect([instanceName = client->getConfig().getInstanceName()](
                                const core::socket::SocketAddress& socketAddress,
                                const core::socket::State& state) { // example.com:81 simulate connnect timeout
                switch (state) {
                    case core::socket::State::OK:
                        LOG(DEBUG) << instanceName << ": connected to '" << socketAddress.toString() << "'";
                        break;
                    case core::socket::State::DISABLED:
                        LOG(DEBUG) << instanceName << ": disabled";
                        break;
                    case core::socket::State::ERROR:
                        LOG(DEBUG) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
                        break;
                    case core::socket::State::FATAL:
                        LOG(DEBUG) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
                        break;
                }
            });
        }

Referenced by web::http::legacy::in6::EventSource::EventSource, web::http::legacy::in::EventSource(), web::http::legacy::rc::EventSource::EventSource, web::http::legacy::un::EventSource(), web::http::tls::in6::EventSource(), web::http::tls::in::EventSource(), web::http::tls::rc::EventSource(), and web::http::tls::un::EventSource().

Here is the caller graph for this function:

operator=() [1/2]

template<typename Client>

EventSourceT & web::http::client::tools::EventSourceT< Client >::operator= ( const EventSourceT< Client > & )

delete

operator=() [2/2]

template<typename Client>

EventSourceT & web::http::client::tools::EventSourceT< Client >::operator= ( EventSourceT< Client > && )

deletenoexcept

parse()

template<typename Client>

bool web::http::client::tools::EventSourceT< Client >::parse ( const std::shared_ptr< SharedState > & sharedState, const std::shared_ptr< SharedConfig > & sharedConfig )

inlinestaticprivate

Definition at line 244 of file EventSource.h.

                                                                                                                          {
            static constexpr size_t kMaxLine = 256 * 1024;
 
            bool success = true;
            bool eol = true;
            do {
                const size_t i = sharedState->pending.find_first_of("\r\n");
 
                eol = i != std::string::npos && (sharedState->pending[i] != '\r' || i + 1 < sharedState->pending.size());
 
                if (eol && i <= kMaxLine) {
                    const size_t eolLength =
                        (sharedState->pending[i] == '\r' && i + 1 < sharedState->pending.size() && sharedState->pending[i + 1] == '\n') ? 2
                                                                                                                                        : 1;
 
                    const std::string_view line(sharedState->pending.data(), i);
 
                    if (!line.empty()) {
                        success = processLine(sharedState, sharedConfig, line);
                    } else {
                        dispatch(sharedState);
                    }
 
                    sharedState->pending.erase(0, i + eolLength);
                } else {
                    success = sharedState->pending.size() <= kMaxLine; // Line length
                }
            } while (eol && success);
 
            return success;
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::init().

Here is the caller graph for this function:

parseU32()

template<typename Client>

uint32_t web::http::client::tools::EventSourceT< Client >::parseU32 ( std::string_view uint32AsString )

inlinestaticprivate

Definition at line 147 of file EventSource.h.

                                                                {
            uint64_t uint32AsUint64 = 0;
 
            for (const char digit : uint32AsString) {
                uint32AsUint64 = uint32AsUint64 * 10 + (static_cast<unsigned char>(digit) - '0');
                if (uint32AsUint64 > 0xFFFFFFFFULL) {
                    return 0xFFFFFFFFU;
                }
            }
 
            return static_cast<uint32_t>(uint32AsUint64);
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::processLine().

Here is the caller graph for this function:

processLine()

template<typename Client>

bool web::http::client::tools::EventSourceT< Client >::processLine ( const std::shared_ptr< SharedState > & sharedState, const std::shared_ptr< SharedConfig > & sharedConfig, std::string_view line )

inlinestaticprivate

Definition at line 196 of file EventSource.h.

                                                     {
            bool success = true;
 
            if (!line.empty() && line.front() != ':') {
                std::string_view field;
                std::string_view value;
 
                if (auto p = line.find(':'); p != std::string_view::npos) {
                    field = line.substr(0, p);
                    value = line.substr(p + 1);
                    if (!value.empty() && value.front() == ' ') {
                        value.remove_prefix(1);
                    }
                } else {
                    field = line;
                }
 
                if (field == "data") {
                    static constexpr size_t kMaxDataSize = 1 << 20;
 
                    success = sharedState->data.size() + value.size() + 1 <= kMaxDataSize; // Data length
 
                    if (success) {
                        sharedState->data.append(value);
                        sharedState->data.push_back('\n');
                    }
                } else if (field == "event") {
                    sharedState->type = value;
                } else if (field == "id") {
                    if (value.find('\0') == std::string_view::npos) {
                        sharedState->idBuf = value;
                    }
                } else if (field == "retry") {
                    if (digits(value)) {
                        uint32_t const retry = parseU32(value);
 
                        sharedState->retry = retry;
                        sharedConfig->config->setReconnectTime(retry / 1000.);
                        sharedConfig->config->setRetryTimeout(retry / 1000.);
                    }
                }
            }
 
            return success;
        }

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::parse().

Here is the caller graph for this function:

Member Data Documentation

client

template<typename Client>

std::shared_ptr<Client> web::http::client::tools::EventSourceT< Client >::client

private

Definition at line 441 of file EventSource.h.

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::init().

sharedConfig

template<typename Client>

std::shared_ptr<SharedConfig> web::http::client::tools::EventSourceT< Client >::sharedConfig

private

Definition at line 443 of file EventSource.h.

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::close(), web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::EventSourceT(), and web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::init().

socketConnection

template<typename Client>

SocketConnection* web::http::client::tools::EventSourceT< Client >::socketConnection = nullptr

private

Definition at line 442 of file EventSource.h.

Referenced by web::http::client::tools::EventSourceT< web::http::legacy::in::Client >::close().

---

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

• web/http/client/tools/EventSource.h