QueueingSocketSink.ct

Go to the documentation of this file.

//
// Copyright (c) 2020 Fraunhofer Institute for Applied Information Technology (FIT)
//                    Network Research Group (NET)
//                    Schloss Birlinghoven, 53754 Sankt Augustin, GERMANY
//                    Contact: support@wiback.org
//
// This file is part of the SENF code tree.
// It is licensed under the 3-clause BSD License (aka New BSD License).
// See LICENSE.txt in the top level directory for details or visit
// https://opensource.org/licenses/BSD-3-Clause
//


/** \file
    \brief QueueingSocketSink non-inline template implementation  */

//#include "QueueingSocketSink.ih"

// Custom includes
#include <senf/Utils/Console/ParsedCommand.hh>

#define prefix_
//-/////////////////////////////////////////////////////////////////////////////////////////////////

//-/////////////////////////////////////////////////////////////////////////////////////////////////
// senf::ppi::module::PassiveQueueingSocketSink<Writer>

template <class Writer>
prefix_ senf::ppi::module::PassiveQueueingSocketSink<Writer>::PassiveQueueingSocketSink(Handle const & handle, QueueingAlgorithm::ptr qAlgorithm)
    : handle_(handle), writer_(),
      qAlgo_(qAlgorithm.release()),
      event_(handle_, IOEvent::Write)
{
    namespace fty = console::factory;
    dir.add( "active", qAlgo_->consoleDir());
    dir.add( "set", fty::Command(
            &PassiveQueueingSocketSink<Writer>::setQAlgorithm, this) );
    dir.add( "list", fty::Command(
            &QueueingAlgorithmRegistry::dump, &QueueingAlgorithmRegistry::instance()));
    registerEvent( event_, &PassiveQueueingSocketSink::writable );
    event_.enabled( false);
    noroute(input);
    input.onRequest( &PassiveQueueingSocketSink::write);
    input.throttlingDisc( ThrottlingDiscipline::NONE);
    checkThrottle();
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::handle(Handle const & handle)
{
    handle_ = handle;
    event_.set( handle_, IOEvent::Write);
    qAlgo_->clear();
    checkThrottle();
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::write()
{
    PacketType const & p ( input());
    if (qAlgo_->size() > 0) {
        qAlgo_->enqueue( p);
        return;
    }
    if (! writer_( handle_, p)) {
        if (qAlgo_->enqueue( p) && !event_.enabled()) {
          event_.enabled( true);
        }
    }
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::writable()
{
    if (qAlgo_->peek() > 0) {
        writer_( handle_, static_cast<PacketType const &>(qAlgo_->front()));
        qAlgo_->pop();
    }
    if (qAlgo_->size() == 0) {
        event_.enabled( false);
    }
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::checkThrottle()
{
    if (handle_.valid())
        input.unthrottle();
    else
        input.throttle();
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::qAlgorithm(QueueingAlgorithm::ptr qAlgorithm)
{
//    dir.remove( "active");
    qAlgo_.reset( qAlgorithm.release());
    dir.add( "active", qAlgo_->consoleDir());
    if (event_.enabled())
        event_.enabled( false);
}

template <class Writer>
prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::setQAlgorithm(std::string const & key)
{
    qAlgorithm( QueueingAlgorithmRegistry::instance().createQAlgorithm( key));
}

//-/////////////////////////////////////////////////////////////////////////////////////////////////
#undef prefix_


// Local Variables:
// mode: c++
// fill-column: 100
// comment-column: 40
// c-file-style: "senf"
// indent-tabs-mode: nil
// ispell-local-dictionary: "american"
// compile-command: "scons -u test"
// End::