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	Config

Public Types
using	MasterRequest = typename Client::MasterRequest
using	Request = typename Client::Request
using	Response = typename Client::Response
using	SocketConnection = typename Client::SocketConnection
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 &url)
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 void	processLine (const std::shared_ptr< SharedState > &sharedState, const std::shared_ptr< Config > &sharedConfig, std::string_view line)
static void	parse (const std::shared_ptr< SharedState > &sharedState, const std::shared_ptr< Config > &sharedConfig)

Private Attributes
std::shared_ptr< Client >	client
SocketConnection *	socketConnection = nullptr
std::shared_ptr< Config >	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 124 of file EventSource.h.

Member Typedef Documentation

MasterRequest

template<typename Client >

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

Definition at line 128 of file EventSource.h.

Request

template<typename Client >

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

Definition at line 129 of file EventSource.h.

Response

template<typename Client >

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

Definition at line 130 of file EventSource.h.

SocketConnection

template<typename Client >

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

Definition at line 131 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 276 of file EventSource.h.

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

References web::http::client::tools::EventSourceT< Client >::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 415 of file EventSource.h.

                              {
            sharedConfig->config->setReconnect(false);
            sharedConfig->config->setRetry(false);
 
            if (socketConnection != nullptr) {
                socketConnection->shutdownRead();
            }
        }

References web::http::client::tools::EventSourceT< Client >::sharedConfig, and web::http::client::tools::EventSourceT< Client >::socketConnection.

deliverMessage()

template<typename Client >

static 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 161 of file EventSource.h.

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

References web::http::client::tools::EventSource::SharedState::onEventListener, web::http::client::tools::EventSource::SharedState::onMessageListener, and web::http::client::tools::EventSource::SharedState::origin.

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

Here is the caller graph for this function:

digits()

template<typename Client >

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

inlinestaticprivate

Definition at line 138 of file EventSource.h.

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

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

Here is the caller graph for this function:

dispatch()

template<typename Client >

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

inlinestaticprivate

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

References web::http::client::tools::EventSource::SharedState::data, web::http::client::tools::EventSourceT< Client >::deliverMessage(), web::http::client::tools::EventSource::SharedState::idBuf, web::http::client::tools::EventSource::SharedState::lastId, and web::http::client::tools::EventSource::SharedState::type.

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

Here is the call graph for this function:

Here is the caller graph for this function:

init()

template<typename Client >

void web::http::client::tools::EventSourceT< Client >::init ( const std::string &  url )

inlineprotected

Definition at line 280 of file EventSource.h.

                                        {
            static const std::regex re(R"(^((https?)://)?([^/:]+)(?::(\d+))?(/.*)?$)");
 
            std::smatch match;
            if (std::regex_match(url, match, re)) {
                // capture URL parts for origin/ports
                sharedState->scheme = (match[2].matched ? match[2].str() : "http");
                sharedState->host = match[3].str();
                const bool isHttps = (sharedState->scheme == "https");
                sharedState->port =
                    match[4].matched ? static_cast<uint16_t>(std::stoi(match[4].str())) : static_cast<uint16_t>(isHttps ? 443 : 80);
                sharedState->origin = sharedState->scheme + std::string("://") + sharedState->host +
                                      (((isHttps && sharedState->port == 443) || (!isHttps && sharedState->port == 80))
                                           ? ""
                                           : (":" + std::to_string(sharedState->port)));
 
                sharedState->path = (match[5].matched ? match[5].str() : "/");
 
                LOG(TRACE) << "Full protocol: " << match[1];
                LOG(TRACE) << "     protocol: " << sharedState->scheme;
                LOG(TRACE) << "         Host: " << sharedState->host;
                LOG(TRACE) << "         Port: " << sharedState->port;
                LOG(TRACE) << "         Path: " << sharedState->path;
                LOG(TRACE) << "       Origin: " << sharedState->origin;
 
                const std::weak_ptr<EventSourceT> eventStreamWeak = this->weak_from_this();
 
                client = std::make_shared<Client>(
                    [eventStreamWeak](SocketConnection* socketConnection) {
                        LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnConnect";
 
                        if (const std::shared_ptr<EventSourceT> eventStream = eventStreamWeak.lock()) {
                            eventStream->socketConnection = socketConnection;
                        }
                    },
                    [state = this->sharedState](SocketConnection* socketConnection) {
                        LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnConnected";
                    },
                    [eventStreamWeak, sharedState = this->sharedState, sharedConfig = this->sharedConfig](
                        SocketConnection* socketConnection) {
                        LOG(DEBUG) << socketConnection->getConnectionName() << " EventSource: OnDisconnect";
 
                        if (const std::shared_ptr<EventSourceT> eventStream = eventStreamWeak.lock()) {
                            eventStream->socketConnection = nullptr;
                        }
 
                        if (sharedConfig->config->getReconnect()) {
                            for (const auto& onError : sharedState->onErrorListener) {
                                onError();
                            }
                            sharedState->ready = ReadyState::CONNECTING;
 
                            sharedConfig->config->setReconnectTime(sharedState->retry / 1000.);
                            sharedConfig->config->setRetryTimeout(sharedState->retry / 1000.);
                        } else {
                            sharedState->ready = ReadyState::CLOSED;
                        }
                    },
                    [url = match[5].matched ? match[5].str() : "/", 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";
 
                        masterRequest->requestEventSource(
                            url,
                            [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));
 
                                    if (consumed > 0) {
                                        sharedState->pending.append(buf, consumed);
                                        parse(sharedState, sharedConfig);
                                    }
                                } 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.);
 
                                for (const auto& onOpen : sharedState->onOpenListener) {
                                    onOpen();
                                }
                            },
                            [sharedState, connectionName]() {
                                LOG(DEBUG) << connectionName
                                           << ": not an server-sent event endpoint: " << sharedState->origin + sharedState->path;
                            });
                    },
                    [](const std::shared_ptr<Request>& req) {
                        LOG(DEBUG) << req->getSocketContext()->getSocketConnection()->getConnectionName() << " EventSource: OnRequestEnd";
                    });
 
                client->getConfig().Remote::setHost(sharedState->host);
                client->getConfig().Remote::setPort(sharedState->port);
                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;
                    }
                });
            }
        }

References web::http::client::tools::EventSourceT< Client >::client, web::http::client::tools::EventSource::CLOSED, web::http::client::tools::EventSource::CONNECTING, core::socket::State::DISABLED, core::socket::State::ERROR, core::socket::State::FATAL, web::http::client::tools::EventSource::SharedState::host, core::socket::State::OK, web::http::client::tools::EventSource::OPEN, web::http::client::tools::EventSource::SharedState::origin, web::http::client::tools::EventSourceT< Client >::parse(), web::http::client::tools::EventSource::SharedState::path, web::http::client::tools::EventSource::SharedState::port, web::http::client::tools::EventSource::SharedState::scheme, web::http::client::tools::EventSourceT< Client >::sharedConfig, web::http::client::tools::EventSource::sharedState, core::socket::SocketAddress::toString(), and core::socket::State::what().

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

Here is the call graph for this function:

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 >

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

inlinestaticprivate

Definition at line 242 of file EventSource.h.

                                                                                                                    {
            size_t start = 0;
            for (;;) {
                const size_t i = sharedState->pending.find_first_of("\r\n", start);
                if (i == std::string::npos) {
                    if (start != 0U) {
                        sharedState->pending.erase(0, start);
                    }
                    break;
                }
 
                if (sharedState->pending[i] == '\r' && i + 1 >= sharedState->pending.size()) {
                    if (start != 0U) {
                        sharedState->pending.erase(0, start);
                    }
                    break;
                }
 
                const size_t eol =
                    (sharedState->pending[i] == '\r' && i + 1 < sharedState->pending.size() && sharedState->pending[i + 1] == '\n') ? 2 : 1;
 
                const std::string_view line(&sharedState->pending[start], i - start);
                start = i + eol;
                processLine(sharedState, sharedConfig, line);
            }
        }

References web::http::client::tools::EventSource::SharedState::pending, and web::http::client::tools::EventSourceT< Client >::processLine().

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

Here is the call graph for this function:

Here is the caller graph for this function:

parseU32()

template<typename Client >

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

inlinestaticprivate

Definition at line 148 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< Client >::processLine().

Here is the caller graph for this function:

processLine()

template<typename Client >

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

inlinestaticprivate

Definition at line 199 of file EventSource.h.

                                                                                                                                   {
            if (line.empty()) {
                dispatch(sharedState);
                return;
            }
            if (line.front() == ':') {
                return;
            }
 
            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") {
                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.);
                }
            }
            // else ignore
        }

References web::http::client::tools::EventSource::SharedState::data, web::http::client::tools::EventSourceT< Client >::digits(), web::http::client::tools::EventSourceT< Client >::dispatch(), web::http::client::tools::EventSource::SharedState::idBuf, web::http::client::tools::EventSourceT< Client >::parseU32(), web::http::client::tools::EventSource::SharedState::retry, and web::http::client::tools::EventSource::SharedState::type.

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

Here is the call graph for this function:

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 425 of file EventSource.h.

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

sharedConfig

template<typename Client >

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

private

Definition at line 427 of file EventSource.h.

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

socketConnection

template<typename Client >

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

private

Definition at line 426 of file EventSource.h.

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

---

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

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