senf/PPI/QueueingSocketSink.hh

   1 // $Id$
   2 //
   3 // Copyright (C) 2010
   4 // Fraunhofer Institute for Open Communication Systems (FOKUS)
   5 //
   6 // The contents of this file are subject to the Fraunhofer FOKUS Public License
   7 // Version 1.0 (the "License"); you may not use this file except in compliance
   8 // with the License. You may obtain a copy of the License at
   9 // http://senf.berlios.de/license.html
  10 //
  11 // The Fraunhofer FOKUS Public License Version 1.0 is based on,
  12 // but modifies the Mozilla Public License Version 1.1.
  13 // See the full license text for the amendments.
  14 //
  15 // Software distributed under the License is distributed on an "AS IS" basis,
  16 // WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
  17 // for the specific language governing rights and limitations under the License.
  18 //
  19 // The Original Code is Fraunhofer FOKUS code.
  20 //
  21 // The Initial Developer of the Original Code is Fraunhofer-Gesellschaft e.V.
  22 // (registered association), Hansastraße 27 c, 80686 Munich, Germany.
  23 // All Rights Reserved.
  24 //
  25 // Contributor(s):
  26 //   Thorsten Horstmann <tho@berlios.de>
  27
  28 /** \file
  29     \brief QueueingSocketSink public header */
  30
  31 #ifndef HH_SENF_PPI_QueueingSocketSink_
  32 #define HH_SENF_PPI_QueueingSocketSink_ 1
  33
  34 // Custom includes
  35 #include <queue>
  36 #include "SocketSink.hh"
  37 #include <senf/Utils/Console/ScopedDirectory.hh>
  38
  39 //#include "QueueingSocketSink.mpp"
  40 //-/////////////////////////////////////////////////////////////////////////////////////////////////
  41
  42 namespace senf {
  43 namespace ppi {
  44
  45     class QueueingAlgorithm
  46         : private boost::noncopyable
  47     {
  48         console::ScopedDirectory<QueueingAlgorithm> dir_;
  49
  50     public:
  51         typedef QueueingAlgorithm * ptr;
  52
  53         virtual ~QueueingAlgorithm() {};
  54
  55         console::DirectoryNode & consoleDir();
  56         Packet dequeue();
  57         bool enqueue(Packet const & packet);
  58         unsigned size();
  59         void clear();
  60
  61     protected:
  62         QueueingAlgorithm();
  63
  64         virtual Packet v_dequeue() = 0;
  65         virtual bool v_enqueue(Packet const & packet) = 0;
  66         virtual unsigned v_size() const = 0;
  67         virtual void v_clear() = 0;
  68     };
  69
  70
  71     namespace detail {
  72         struct QueueingAlgorithmRegistry_EntryBase
  73         {
  74             virtual QueueingAlgorithm::ptr create() const = 0;
  75         };
  76
  77         template <class QAlgorithm>
  78         struct QueueingAlgorithmRegistry_Entry : QueueingAlgorithmRegistry_EntryBase
  79         {
  80             virtual QueueingAlgorithm::ptr create() const;
  81         };
  82     }
  83
  84     class QueueingAlgorithmRegistry
  85         : public senf::singleton<QueueingAlgorithmRegistry>
  86     {
  87         typedef boost::ptr_map<std::string, detail::QueueingAlgorithmRegistry_EntryBase> QAlgoMap;
  88         QAlgoMap qAlgoMap_;
  89
  90         QueueingAlgorithmRegistry() {};
  91     public:
  92         using senf::singleton<QueueingAlgorithmRegistry>::instance;
  93         friend class senf::singleton<QueueingAlgorithmRegistry>;
  94
  95         struct Exception : public senf::Exception {
  96             Exception(std::string const & descr) : senf::Exception(descr) {}
  97         };
  98
  99         template <class QAlgorithm>
 100         struct RegistrationProxy {
 101             RegistrationProxy(std::string const & key);
 102         };
 103
 104         template <class QAlgorithm>
 105         void registerQAlgorithm(std::string key);
 106
 107         QueueingAlgorithm::ptr createQAlgorithm(std::string const & key) const;
 108         void dump(std::ostream & os) const;
 109     };
 110
 111
 112 #   define SENF_PPI_REGISTER_QALGORITHM( key, QAlgorithm )                          \
 113         namespace {                                                                 \
 114             senf::ppi::QueueingAlgorithmRegistry::RegistrationProxy<QAlgorithm>     \
 115                 BOOST_PP_CAT(qAlgorithmRegistration_, __LINE__)( key);              \
 116         }
 117
 118
 119     class FIFOQueueingAlgorithm : public QueueingAlgorithm
 120     {
 121         std::queue<Packet> queue_;
 122         unsigned max_size_;
 123
 124         FIFOQueueingAlgorithm();
 125
 126         virtual Packet v_dequeue();
 127         virtual bool v_enqueue(Packet const & packet);
 128         virtual unsigned v_size() const;
 129         virtual void v_clear();
 130
 131     public:
 132         static QueueingAlgorithm::ptr create();
 133     };
 134
 135
 136 namespace module {
 137
 138     /** \brief QueueingSocketSink
 139
 140         \ingroup io_modules
 141      */
 142     template <class Writer=ConnectedDgramWriter>
 143     class PassiveQueueingSocketSink : public Module
 144     {
 145         SENF_PPI_MODULE(PassiveQueueingSocketSink);
 146
 147     public:
 148         typedef typename Writer::Handle Handle; ///< Handle type requested by writer
 149         typedef typename Writer::PacketType PacketType;
 150
 151         connector::PassiveInput<PacketType> input; ///< Input connector from which data is received
 152         console::ScopedDirectory<PassiveQueueingSocketSink<Writer> > dir;
 153
 154         explicit PassiveQueueingSocketSink(Handle const & handle, QueueingAlgorithm::ptr qAlgorithm);
 155
 156         Writer & writer();              ///< Access the Writer
 157         Handle & handle();              ///< Access handle
 158         void handle(Handle const & handle);
 159                                         ///< Set handle
 160                                         /**< Assigning an empty or in-valid() handle will disable
 161                                              the module until a new valid handle is assigned. */
 162         QueueingAlgorithm & qAlgorithm();
 163         void qAlgorithm(QueueingAlgorithm::ptr qAlgorithm);
 164
 165     private:
 166         void write();
 167         void writable();
 168         void checkThrottle();
 169         void setQAlgorithm(std::string const & key);
 170
 171         Handle handle_;
 172         Writer writer_;
 173         boost::scoped_ptr<QueueingAlgorithm> qAlgo_;
 174         IOEvent event_;
 175     };
 176
 177 }}}
 178
 179 //-/////////////////////////////////////////////////////////////////////////////////////////////////
 180 #include "QueueingSocketSink.cci"
 181 #include "QueueingSocketSink.ct"
 182 #include "QueueingSocketSink.cti"
 183 #endif
 184
 185 \f
 186 // Local Variables:
 187 // mode: c++
 188 // fill-column: 100
 189 // c-file-style: "senf"
 190 // indent-tabs-mode: nil
 191 // ispell-local-dictionary: "american"
 192 // compile-command: "scons -u test"
 193 // comment-column: 40
 194 // End: