GlimpseService.cpp

Go to the documentation of this file.

/*
* Copyright Onix Solutions Limited [OnixS]. All rights reserved.
*
* This software owned by Onix Solutions Limited [OnixS] and is protected by copyright law
* and international copyright treaties.
*
* Access to and use of the software is governed by the terms of the applicable OnixS Software
* Services Agreement (the Agreement) and Customer end user license agreements granting
* a non-assignable, non-transferable and non-exclusive license to use the software
* for it's own data processing purposes under the terms defined in the Agreement.
*
* Except as otherwise granted within the terms of the Agreement, copying or reproduction of any part
* of this source code or associated reference material to any other location for further reproduction
* or redistribution, and any amendments to this copyright notice, are expressly prohibited.
*
* Any reproduction or redistribution for sale or hiring of the Software not in accordance with
* the terms of the Agreement is a violation of copyright law.
*/

#include <OnixS/SgxTitan/MarketData/Itch/HandlerSettings.h>

#include <OnixS/HandlerCore/HandlerLogger.h>

#include "HandlerLogger.h"

#include "GlimpseService.Messages.h"
#include "GlimpseService.h"


ONIXS_HANDLER_NAMESPACE_BEGIN

using namespace OnixS::ItchCore;

namespace
{
    DataSource createDataSource(const std::string& sessionId)
    {
        DataSource dataSource;
        dataSource.origin = DataSource::glimpse;
        dataSource.packetMessageCount = 1;
        dataSource.packetMessageNumber = 0;
        dataSource.packetReceptionTime = Timestamp();


        dataSource.messageSeqNum = 0;
        dataSource.session = reinterpret_cast<const Byte*>(sessionId.c_str());

        return dataSource;
    }

    OnixS::ItchCore::SoupBinTCP::ReconnectableSessionSettings createSessionSettings(const HandlerSettings& settings)
    {
        OnixS::ItchCore::SoupBinTCP::ReconnectableSessionSettings sessionSettings;
        sessionSettings.receiveTimeoutMs = 2000;
        sessionSettings.sendTimeoutMs = 900;

        sessionSettings.reconnectIntervalMs = settings.minReconnectingTimeout * 1000;
        sessionSettings.reconnectAttempts = settings.maxConnectingAttempts;
        sessionSettings.resetSeqNumWhenReconnect = true;

        return sessionSettings;
    }


    OnixS::ItchCore::SoupBinTCP::ConnectivitySettings createSessionConnectivitySettings(
            const HandlerSettings& settings,
            const SessionType& session)
    {
        OnixS::ItchCore::SoupBinTCP::ConnectivitySettings connectivitySettings;

        connectivitySettings.connectionA =
                OnixS::ItchCore::SoupBinTCP::ValidatedNetFeedConnection(
                        settings.glimpseFeed.serviceA.address,
                        settings.glimpseFeed.serviceA.port,
                        settings.networkInterfaceForTcpServices);

        connectivitySettings.connectionB =
                OnixS::ItchCore::SoupBinTCP::ValidatedNetFeedConnection(
                        settings.glimpseFeed.serviceB.address,
                        settings.glimpseFeed.serviceB.port,
                        settings.networkInterfaceForTcpServices);


        connectivitySettings.authenticationSettings.userName = settings.glimpseUsername;
        connectivitySettings.authenticationSettings.password = settings.glimpsePassword;
        connectivitySettings.authenticationSettings.session = std::string(reinterpret_cast<const char*>(session));
        connectivitySettings.initialSequenceNumber = 1;

        return connectivitySettings;
    }

}
    
GlimpseService::GlimpseService(
        const HandlerSettings& settings,
        const Logging::LogFacility* parent,
        HandlerCore::Common::HandlerLogger* logger)
        : Logging::LogFacility("GlimpseService", parent, OnixS::Logging::LOG_LEVEL_DEBUG)
    , settings_(settings)
    , sessionSettings_(createSessionSettings(settings_))
    , sessionConnectivitySettings_()
    , logger_(logger)
    , inProgress_(false)
{
}

GlimpseService::~GlimpseService()
{
}

void GlimpseService::invokeOnFailure(const std::string& what)
{
    if (onFailureCallback_)
        onFailureCallback_(what);
}

void GlimpseService::invokeOnWarning(const std::string& what)
{
    if (onWarningCallback_)
        onWarningCallback_(what);
}

void GlimpseService::invokeOnMessage(const DataSource& ds, const IncomingMessage& msg)
{
    if (onMessageCallback_)
        onMessageCallback_(ds, &msg);
}

void GlimpseService::invokeOnRestarted()
{
    if (onRestartedCallback_)
        onRestartedCallback_();
}

void GlimpseService::subscribeOnFailure(OnFailureCallback callback)
{
    onFailureCallback_ = callback;
}

void GlimpseService::subscribeOnWarning(OnWarningCallback callback)
{
    onWarningCallback_ = callback;
}

void GlimpseService::subscribeOnMessage(OnMessageCallback callback)
{
    onMessageCallback_ = callback;
}

void GlimpseService::subscribeOnRestarted(OnRestartedCallback callback)
{
    onRestartedCallback_ = callback;
}

bool GlimpseService::inProgress() const
{
    return inProgress_;
}

void GlimpseService::stop(bool wait)
{
    log(ONIXS_LOG_DEBUG[this] << "Stopping.");
    
    Guard guard(sessionLock_);

    if (session_)
    {
        session_->disconnect(!wait);
    }
}

void GlimpseService::request(const SessionType& session, SequenceNumber requestedSequence)
{
    if (settings_.logSettings & LogSettings::LogPackets)
        log(ONIXS_LOG_INFO[this] << " SnapshotRecoveryRequested");

    sessionConnectivitySettings_ = createSessionConnectivitySettings(settings_, session);

    Guard guard(sessionLock_);

    inProgress_ = true;

    log(ONIXS_LOG_DEBUG[this] << "Requested glimpse sequence: " << requestedSequence);

    session_.reset(
            new OnixS::ItchCore::SoupBinTCP::ReconnectableSession(sessionSettings_, sessionConnectivitySettings_, this, this)
    );

    session_->setLogger(logger_);

    session_->connect();
}

void GlimpseService::onError(OnixS::ItchCore::SoupBinTCP::ReconnectableSession&, const std::string& what)
{
    log(ONIXS_LOG_ERROR[this] << what);

    invokeOnFailure(what);
}

void GlimpseService::onWarning(OnixS::ItchCore::SoupBinTCP::ReconnectableSession&, const std::string& what)
{
    log(ONIXS_LOG_WARN[this] << what);

    invokeOnWarning(what);
}

void GlimpseService::onStateChange(OnixS::ItchCore::SoupBinTCP::ReconnectableSession& session,
                                   OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Enum oldState,
                                   OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Enum newState)
{
    if (OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Disconnected == newState)
        inProgress_ = false;


    if (
        OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Connected == oldState &&
        OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Connecting == newState
        )
    {
        if (settings_.logSettings & LogSettings::LogPackets)
            log(ONIXS_LOG_INFO[this] << " SnapshotRecoveryRestarted");

        log(ONIXS_LOG_WARN[this] << "Reconnect detected, resetting message repository.");
        
        invokeOnRestarted();
    }


    if (OnixS::ItchCore::SoupBinTCP::ReconnectableSessionState::Connected == newState)
    {
        try
        {
            sessionId_ = session.sessionId();
        }
        catch (...)
        {
            log(ONIXS_LOG_WARN[this] << "Error during retrieving session id");
        }
    }
}

void GlimpseService::onData(OnixS::ItchCore::SoupBinTCP::ReconnectableSession& session, const void* data, UInt16 size)
{
    BOOST_ASSERT(data != nullptr);
    BOOST_ASSERT(size != 0);

    try
    {
        processData(session, data, size);
    }
    catch (...)
    {
        log(ONIXS_LOG_WARN[this] << "Exception during message processing.");
    }
}

void GlimpseService::processData(OnixS::ItchCore::SoupBinTCP::ReconnectableSession& session, const void* data, UInt16 size)
{
    const DataSource dataSource = createDataSource(sessionId_);
    const IncomingMessage message = IncomingMessage(static_cast<const unsigned char*>(data), size);
    
    invokeOnMessage(dataSource, message);

    if (static_cast<GlimpseMessageType::Enum>(message.type()) == (GlimpseMessageType::EndOfSnapshot))
    {
        session.disconnect(true);
    }
}


ONIXS_HANDLER_NAMESPACE_END