senf/PPI/QueueingSocketSink.ct

   1 // $Id$
   2 //
   3 // Copyright (C) 2010
   4 // Fraunhofer Institute for Open Communication Systems (FOKUS)
   5 // Competence Center NETwork research (NET), St. Augustin, GERMANY
   6 //     Thorsten Horstmann <tho@berlios.de>
   7 //
   8 // This program is free software; you can redistribute it and/or modify
   9 // it under the terms of the GNU General Public License as published by
  10 // the Free Software Foundation; either version 2 of the License, or
  11 // (at your option) any later version.
  12 //
  13 // This program is distributed in the hope that it will be useful,
  14 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 // GNU General Public License for more details.
  17 //
  18 // You should have received a copy of the GNU General Public License
  19 // along with this program; if not, write to the
  20 // Free Software Foundation, Inc.,
  21 // 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  22
  23 /** \file
  24     \brief QueueingSocketSink non-inline template implementation  */
  25
  26 //#include "QueueingSocketSink.ih"
  27
  28 // Custom includes
  29 #include <senf/Utils/Console/ParsedCommand.hh>
  30
  31 #define prefix_
  32 //-/////////////////////////////////////////////////////////////////////////////////////////////////
  33
  34 //-/////////////////////////////////////////////////////////////////////////////////////////////////
  35 // senf::ppi::QueueingAlgorithmRegistry
  36
  37 template <class QAlgorithm>
  38 prefix_ void senf::ppi::QueueingAlgorithmRegistry::registerQAlgorithm(std::string key)
  39 {
  40     if (qAlgoMap_.find( key) == qAlgoMap_.end() )
  41         qAlgoMap_.insert(key, new detail::QueueingAlgorithmRegistry_Entry<QAlgorithm>() );
  42     else
  43         throw Exception("Duplicated QAlgorithm Registration ") << key;
  44 }
  45
  46 //-/////////////////////////////////////////////////////////////////////////////////////////////////
  47 // senf::ppi::detail::QueueingAlgorithmRegistry_Entry<QAlgorithm>
  48
  49 template <class QAlgorithm>
  50 prefix_ senf::ppi::QueueingAlgorithm::ptr senf::ppi::detail::QueueingAlgorithmRegistry_Entry<QAlgorithm>::create()
  51     const
  52 {
  53     return QAlgorithm::create();
  54 }
  55
  56 //-/////////////////////////////////////////////////////////////////////////////////////////////////
  57 // senf::ppi::module::PassiveQueueingSocketSink<Writer>
  58
  59 template <class Writer>
  60 prefix_ senf::ppi::module::PassiveQueueingSocketSink<Writer>::PassiveQueueingSocketSink(Handle const & handle, QueueingAlgorithm::ptr qAlgorithm)
  61     : dir( this),
  62       handle_( handle), writer_( ),
  63       qAlgo_( qAlgorithm),
  64       event_( handle_, IOEvent::Write)
  65 {
  66     namespace fty = console::factory;
  67     dir.add( "active", qAlgo_->consoleDir());
  68     dir.add( "set", fty::Command(
  69             &PassiveQueueingSocketSink<Writer>::setQAlgorithm, this) );
  70     dir.add( "list", fty::Command(
  71             &QueueingAlgorithmRegistry::dump, &QueueingAlgorithmRegistry::instance()));
  72     registerEvent( event_, &PassiveQueueingSocketSink::writable );
  73     event_.enabled( false);
  74     noroute(input);
  75     input.onRequest( &PassiveQueueingSocketSink::write);
  76     input.qdisc( QueueingDiscipline::NONE);
  77     checkThrottle();
  78 }
  79
  80 template <class Writer>
  81 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::handle(Handle const & handle)
  82 {
  83     handle_ = handle;
  84     event_.set( handle_, IOEvent::Write);
  85     qAlgo_->clear();
  86     checkThrottle();
  87 }
  88
  89 template <class Writer>
  90 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::write()
  91 {
  92     PacketType p ( input());
  93     if (qAlgo_->size() > 0) {
  94         qAlgo_->enqueue( p);
  95         return;
  96     }
  97     if (! writer_( handle_, p)) {
  98         if (qAlgo_->enqueue( p) && !event_.enabled()) {
  99           event_.enabled( true);
 100         }
 101     }
 102 }
 103
 104 template <class Writer>
 105 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::writable()
 106 {
 107     PacketType p (qAlgo_->dequeue());
 108     if (p)
 109         writer_( handle_, p);
 110     if (qAlgo_->size() == 0) {
 111         event_.enabled( false);
 112     }
 113 }
 114
 115 template <class Writer>
 116 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::checkThrottle()
 117 {
 118     if (handle_.valid())
 119         input.unthrottle();
 120     else
 121         input.throttle();
 122 }
 123
 124 template <class Writer>
 125 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::qAlgorithm(QueueingAlgorithm::ptr qAlgorithm)
 126 {
 127 //    dir.remove( "active");
 128     qAlgo_.reset( qAlgorithm);
 129     dir.add( "active", qAlgo_->consoleDir());
 130     if (event_.enabled())
 131         event_.enabled( false);
 132 }
 133
 134 template <class Writer>
 135 prefix_ void senf::ppi::module::PassiveQueueingSocketSink<Writer>::setQAlgorithm(std::string const & key)
 136 {
 137     qAlgorithm( QueueingAlgorithmRegistry::instance().createQAlgorithm( key));
 138 }
 139
 140 //-/////////////////////////////////////////////////////////////////////////////////////////////////
 141 #undef prefix_
 142
 143 \f
 144 // Local Variables:
 145 // mode: c++
 146 // fill-column: 100
 147 // comment-column: 40
 148 // c-file-style: "senf"
 149 // indent-tabs-mode: nil
 150 // ispell-local-dictionary: "american"
 151 // compile-command: "scons -u test"
 152 // End::