FeedEngineImpl.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 "FeedEngineImpl.h"

#include <system/Thread.h>

#include <boost/foreach.hpp>

ONIXS_HANDLER_NAMESPACE_BEGIN

using namespace std;

FeedEngineImpl::FeedEngineImpl(const FeedEngineSettings & settings, HANDLER_NAMESPACE::FeedEngineListener* listener, FeedEngine* publicFeedEngine)
    : settings_(settings)
    , listener_(listener)
    , publicFeedEngine_(publicFeedEngine)
{
    System::Thread::CpuIndexes cpuIndexes;
    settings.threadAffinity().copyTo(cpuIndexes);

    HandlerCore::MarketData::FE::FeedEngineSettings feSettings;

    BOOST_FOREACH(size_t cpuIndex, cpuIndexes)
    {
        feSettings.threadAffinity().cpus().insert(static_cast<HandlerCore::MarketData::FE::CpuIndex>(cpuIndex));
    }

    feSettings.threadCount(settings.threadCount());
    feSettings.watch(HandlerCore::MarketData::FE::UtcWatch::service());
    feSettings.dataWaitTime(settings.dataWaitTime());
    feSettings.spinBeforeIdleTime(settings.spinBeforeIdleTime());
    feSettings.socketBufferSize(settings.socketBufferSize());

    feedEngine_.reset(new HandlerCore::MarketData::FE::MultithreadedFeedEngine(feSettings, this));
}

FeedEngineImpl::~FeedEngineImpl()
{
}

void FeedEngineImpl::start()
{
    feedEngine_->start();
}

void FeedEngineImpl::onFeedEngineThreadBegin() 
{
    if (listener_)
        listener_->onFeedEngineThreadBegin(*publicFeedEngine_);
}

void FeedEngineImpl::onFeedEngineThreadEnd() 
{
    if (listener_)
        listener_->onFeedEngineThreadEnd(*publicFeedEngine_);
}

void FeedEngineImpl::onFeedEngineThreadIdle(HandlerCore::MarketData::FE::FeedEngineThreadIdle::Reason reason, UInt32& timeout) 
{
    if (listener_)
        listener_->onFeedEngineThreadIdle(*publicFeedEngine_, static_cast<HANDLER_NAMESPACE::FeedEngineThreadIdle::Reason>(reason), timeout);
}

ONIXS_HANDLER_NAMESPACE_END