X-Git-Url: http://g0dil.de/git?a=blobdiff_plain;f=Examples%2FRateStuffer%2Fratestuffer.cc;h=6f6a63423d70a59577bc9455ad077b78299ff3b1;hb=20c3dbf64b3beed10655dceaee5791680fb25383;hp=ea294273aa43872843dd93a0d7c495f67ab7e97f;hpb=f0d8247a8343da19c00452a290e7aeeb8944652c;p=senf.git diff --git a/Examples/RateStuffer/ratestuffer.cc b/Examples/RateStuffer/ratestuffer.cc index ea29427..6f6a634 100644 --- a/Examples/RateStuffer/ratestuffer.cc +++ b/Examples/RateStuffer/ratestuffer.cc @@ -1,8 +1,8 @@ // $Id$ // // Copyright (C) 2007 -// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS) -// Kompetenzzentrum fuer Satelitenkommunikation (SatCom) +// Fraunhofer Institute for Open Communication Systems (FOKUS) +// Competence Center NETwork research (NET), St. Augustin, GERMANY // Stefan Bund // // This program is free software; you can redistribute it and/or modify @@ -27,16 +27,8 @@ //#include "ppitest.ih" // Custom includes -#include "Socket/Protocols/INet/UDPSocketHandle.hh" -#include "Socket/Protocols/INet/ConnectedUDPSocketHandle.hh" -#include "Socket/Protocols/INet/INetAddressing.hh" -#include "PPI/SocketReader.hh" -#include "PPI/SocketWriter.hh" -#include "PPI/Module.hh" -#include "PPI/IntervalTimer.hh" -#include "PPI/Joins.hh" -#include "PPI/PassiveQueue.hh" -#include "PPI/Setup.hh" +#include +#include //#include "ppitest.mpp" #define prefix_ @@ -46,82 +38,107 @@ namespace module = senf::ppi::module; namespace connector = senf::ppi::connector; namespace ppi = senf::ppi; -namespace { +// //////////////////////////////////////////////////////////////////////// +// RateFilter - class RateFilter - : public module::Module - { - SENF_PPI_MODULE(RateFilter); - public: +class RateFilter + : public module::Module +{ + SENF_PPI_MODULE(RateFilter); +public: - connector::ActiveInput input; - connector::ActiveOutput output; + connector::ActiveInput<> input; + connector::ActiveOutput<> output; - RateFilter(senf::ClockService::clock_type interval) : timer(interval) { - route(input,output); - route(input,timer); - registerEvent(&RateFilter::timeout, timer); - } + RateFilter(senf::ClockService::clock_type interval); - private: - void timeout() { - output(input()); - } +private: + void timeout(); - ppi::IntervalTimer timer; - }; + ppi::IntervalTimer timer; +}; - class CopyPacketGenerator - : public module::Module - { - SENF_PPI_MODULE(CopyPacketGenerator); - public: +RateFilter::RateFilter(senf::ClockService::clock_type interval) + : timer(interval) +{ + route(input,timer); + route(timer,output); + registerEvent(timer, &RateFilter::timeout); +} + +void RateFilter::timeout() +{ + output(input()); +} - connector::PassiveOutput output; +// //////////////////////////////////////////////////////////////////////// - CopyPacketGenerator(senf::Packet p) : packet(p) { - noroute(output); - output.onRequest(&CopyPacketGenerator::request); - } +class RateStuffer +{ + module::ThrottleBarrier barrier; + module::PassiveQueue queue; + module::CloneSource generator; + module::PriorityJoin join; + RateFilter rateFilter; + +public: + connector::PassiveInput<> & input; + connector::ActiveOutput<> & output; + + RateStuffer(senf::ClockService::clock_type interval, + senf::Packet packet, + unsigned high = 1, + unsigned low = 0) + : barrier (), + queue (), + generator ( packet ), + join (), + rateFilter ( interval ), + input ( barrier.input ), + output ( rateFilter.output ) + { + ppi::connect( barrier, queue ); + ppi::connect( queue, join ); + ppi::connect( generator, join ); + ppi::connect( join, rateFilter ); - private: - void request() { - output(packet); - } + queue.qdisc(ppi::ThresholdQueueing(high,low)); + } +}; - senf::Packet packet; - }; -} +// //////////////////////////////////////////////////////////////////////// +// //////////////////////////////////////////////////////////////////////// // Module setup: // // 'O' = active connector // '>' or '<' = input connector // -// [ udpReader ] O--> [ queue ] -->O [ ] -// [ join ] -->O [ rateFilter] O--> [ udpWriter ] -// [ generator ] -->O [ ] +// +----------------------------------------------------+ +// | stuffer | +// | | +// [ udpSource ] O-->:---> [ queue ] -->O [ ] | +// | [ join ] -->O [ rateFilter] O-->:O--> [ udpSink ] +// | [ generator ] -->O [ ] | +// | | +// +----------------------------------------------------+ int main(int argc, char * argv[]) { - senf::UDPv4ClientSocketHandle inputSocket; - inputSocket.bind(senf::INet4SocketAddress("0.0.0.0:44344")); + senf::UDPv4ClientSocketHandle inputSocket( + senf::INet4SocketAddress("0.0.0.0:44344")); senf::ConnectedUDPv4ClientSocketHandle outputSocket( senf::INet4SocketAddress("localhost:44345")); - module::ActiveSocketReader<> udpReader (inputSocket); - module::PassiveQueue queue; - CopyPacketGenerator generator (senf::DataPacket::create(std::string("\n"))); - module::PriorityJoin join; - RateFilter rateFilter (1000000000ul); - module::PassiveSocketWriter<> udpWriter (outputSocket); - - ppi::connect( udpReader, queue ); - ppi::connect( queue, join ); - ppi::connect( generator, join ); - ppi::connect( join, rateFilter ); - ppi::connect( rateFilter, udpWriter ); + module::ActiveSocketSource<> udpSource ( inputSocket ); + RateStuffer stuffer ( 1000000000ul, + senf::DataPacket::create(std::string("\n")), + 2u, 1u ); + module::PassiveSocketSink<> udpSink ( outputSocket ); + + ppi::connect( udpSource, stuffer ); + ppi::connect( stuffer, udpSink ); ppi::run();