example/comm/rrClient/main.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 /**
  10    Realiza peticiones automaticas sobre el servidor de operaciones aritmeticas.
  11
  12    Mediante un servicio de reparto por RoundRobin.
  13
  14    El servidor de este cliente: server.p rserver.p
  15 */
  16
  17 #include <iostream>
  18
  19 #include <string.h>
  20 #include <unistd.h>
  21
  22 #include <anna/core/core.hpp>
  23 #include <anna/app/functions.hpp>
  24 #include <anna/comm/comm.hpp>
  25
  26 #include <anna/timex/Engine.hpp>
  27 #include <anna/timex/Clock.hpp>
  28
  29 #include <anna/test/Response.hpp>
  30 #include <anna/test/Request.hpp>
  31
  32 class Sender : public anna::timex::Clock {
  33 public:
  34    Sender () : Clock ("Sender", (Millisecond)1000),
  35       a_messageBySecond (0),
  36       a_nquarter (0),
  37       a_requests ("Request"),
  38       a_errorCounter (0),
  39       a_txMessageCounter (0)
  40    {;}
  41
  42    void setMessageBySecond (const int messageBySecond) throw () { a_messageBySecond = messageBySecond; }
  43
  44    int getTxMessageCounter () const throw () { return a_txMessageCounter; }
  45
  46 private:
  47    int a_messageBySecond;
  48    int a_nquarter;
  49    int a_errorCounter;
  50    int a_txMessageCounter;
  51    ThreadData <test::Request> a_requests;
  52
  53    bool tick () throw (RuntimeException);
  54 };
  55
  56 class MyCommunicator : public Communicator {
  57 public:
  58    MyCommunicator () : Communicator (), a_avgResponseTime (0), a_rxMessageCounter (0), a_responses ("Response")  {;}
  59
  60 private:
  61    ThreadData <test::Response> a_responses;
  62    int a_avgResponseTime;
  63    int a_rxMessageCounter;
  64
  65    void eventReceiveMessage (ClientSocket &, const Message&) throw (RuntimeException);
  66 /*
  67    void eventBreakConnection (const ClientSocket&) throw ();
  68
  69    void eventBreakConnection (Server* server) throw () {
  70       comm::Communicator::eventBreakConnection (server);
  71    }
  72 */
  73    void eventBreakConnection (const Service* service) throw ();
  74
  75    static bool isOk (const test::Response& response) throw ();
  76 };
  77
  78 class RRClient : public anna::comm::Application {
  79 public:
  80    RRClient ();
  81
  82    Service* getService () const throw () { return a_service; }
  83    const Sender* getSender () const throw () { return &a_sender; }
  84
  85 private:
  86    MyCommunicator a_communicator;
  87    anna::timex::Engine a_timeController;
  88    Sender a_sender;
  89    Service* a_service;
  90
  91    void initialize () throw (RuntimeException);
  92    void run () throw (RuntimeException);
  93 };
  94
  95 using namespace std;
  96
  97 int main (int argc, const char** argv)
  98 {
  99    CommandLine& commandLine (CommandLine::instantiate ());
 100    RRClient app;
 101
 102    srand (time (NULL));
 103
 104    try {
 105       commandLine.initialize (argv, argc);
 106       commandLine.verify ();
 107
 108       Logger::setLevel (Logger::Information);
 109       string traceFile ("client.");
 110       traceFile += anna::functions::asString ((int) getpid ());
 111       traceFile += ".trace";
 112       Logger::initialize ("arithmeticClient", new TraceWriter (traceFile.c_str (),4096000));
 113
 114       app.start ();
 115    }
 116    catch (Exception& ex) {
 117       cout << ex.asString () << endl;
 118    }
 119
 120    return 0;
 121 }
 122
 123 RRClient::RRClient () :
 124    Application ("arithmeticClient", "Cliente de operaciones aritmeticas", "1.0"),
 125    a_communicator (),
 126    a_timeController ((Millisecond)1000, (Millisecond)250)
 127 {
 128    CommandLine& commandLine (CommandLine::instantiate ());
 129
 130    commandLine.add ("p", CommandLine::Argument::Mandatory, "Puertos de los servidores (separados por comas)");
 131    commandLine.add ("a", CommandLine::Argument::Mandatory, "Direccin IP Puerto en el que el servidor atiende respuestas.");
 132    commandLine.add ("n", CommandLine::Argument::Mandatory, "Numero de mensajes por segundo");
 133    commandLine.add ("trace", CommandLine::Argument::Optional, "Nivel de trazas (debug,warning, error,...)");
 134 }
 135
 136 void RRClient::initialize ()
 137    throw (RuntimeException)
 138 {
 139    CommandLine& cl (CommandLine::instantiate ());
 140
 141    if (cl.exists ("trace"))
 142       Logger::setLevel (Logger::asLevel (cl.getValue ("trace")));
 143
 144    a_sender.setMessageBySecond (cl.getIntegerValue ("n"));
 145
 146    Network& network = Network::instantiate ();
 147    Tokenizer ports (cl.getValue ("p"), ",");
 148    int port;
 149
 150    a_service = new comm::RoundRobinDelivery ("Service_Arithmetic", true);
 151
 152    for (Tokenizer::const_iterator ii = ports.begin (), maxii = ports.end (); ii != maxii; ii ++) {
 153       port = atoi (Tokenizer::data (ii));
 154       a_service->attach (network.createServer (cl.getValue ("a"), port, true));
 155    }
 156
 157    a_communicator.attach (a_service);
 158 }
 159
 160 void RRClient::run ()
 161    throw (RuntimeException)
 162 {
 163    a_timeController.activate (a_sender);
 164
 165    a_communicator.accept ();
 166 }
 167
 168 void MyCommunicator::eventReceiveMessage (ClientSocket&, const Message& message)
 169    throw (RuntimeException)
 170 {
 171   LOGMETHOD (TraceMethod tm ("MyCommunicator", "eventReceiveMessage", ANNA_FILE_LOCATION));
 172
 173    test::Response& response = a_responses.get ();
 174    response.decode (message.getBody ());
 175
 176    const anna::Millisecond now = anna::functions::millisecond ();
 177    const int delay =  now - (Millisecond) response.initTime;
 178
 179    if (delay > 0 && isOk (response) == true) {
 180       a_rxMessageCounter ++;
 181       a_avgResponseTime += delay;
 182
 183       LOGINFORMATION (
 184          string msg = anna::functions::asString (
 185             "%d %c %d = %d", response.x, response.op, response.y, response.result
 186          );
 187          msg += anna::functions::asText (" | Delay: ", delay);
 188          Logger::information (msg, ANNA_FILE_LOCATION);
 189       );
 190    }
 191    else {
 192       LOGWARNING (
 193          string msg = anna::functions::asString (
 194             "Flip: %d %c %d = %d", response.x, response.op, response.y, response.result
 195          );
 196          msg += anna::functions::asText (" | Message: ", message.getBody ());
 197          msg += anna::functions::asText (" | Delay: ", delay);
 198          Logger::warning (msg, ANNA_FILE_LOCATION);
 199       );
 200    }
 201 }
 202
 203 void MyCommunicator::eventBreakConnection (const Service* service)
 204    throw ()
 205 {
 206   LOGMETHOD (TraceMethod tm ("MyCommunicator", "eventBreakConnection", ANNA_FILE_LOCATION));
 207
 208    if (a_rxMessageCounter == 0)
 209       return;
 210
 211    comm::Communicator::eventBreakConnection (service);
 212
 213    if (service->isAvailable () == true) {
 214       LOGNOTICE (Logger::notice (service->asString (), ANNA_FILE_LOCATION));
 215       return;
 216    }
 217
 218    LOGNOTICE (
 219       RRClient& app = static_cast <RRClient&> (anna::app::functions::getApp ());
 220       string msg ("Tiempo medio respuesta: ");
 221       msg += anna::functions::asString (a_avgResponseTime / a_rxMessageCounter);
 222       msg += " ms";
 223       msg += anna::functions::asText (" | Rx: ", a_rxMessageCounter);
 224       msg += anna::functions::asText (" | Tx: ", app.getSender ()->getTxMessageCounter ());
 225       Logger::notice (msg, ANNA_FILE_LOCATION);
 226
 227       cout << msg << endl << endl;
 228    );
 229    requestStop ();
 230 }
 231
 232 bool MyCommunicator::isOk (const test::Response& response)
 233    throw ()
 234 {
 235    if (response.op != '+' && response.op != '-' && response.op != '*' && response.op != '/')
 236       return false;
 237
 238    int result = 0;
 239
 240    switch (response.op) {
 241       case '+':
 242          result = response.x + response.y;
 243          break;
 244       case '-':
 245          result = response.x - response.y;
 246          break;
 247       case '*':
 248          result = response.x * response.y;
 249          break;
 250       case '/':
 251          result = (response.y != 0) ? (response.x / response.y): 0;
 252          break;
 253    }
 254
 255    return result == response.result;
 256 }
 257
 258 bool Sender::tick ()
 259    throw (RuntimeException)
 260 {
 261   LOGMETHOD (TraceMethod tm (Logger::Local7, "Sender", "tick", ANNA_FILE_LOCATION));
 262
 263    Service* service = static_cast <RRClient&> (anna::comm::functions::getApp ()).getService ();
 264    Communicator* communicator = anna::app::functions::component <Communicator> (ANNA_FILE_LOCATION);
 265
 266    if (a_messageBySecond == 0)
 267       throw RuntimeException ("Hay que indicar el numero de mensajes por segundo", ANNA_FILE_LOCATION);
 268
 269
 270    if (a_errorCounter > 100) {
 271       communicator->requestStop ();
 272       Logger::warning ("Terminado por errores continuos en la conexion", ANNA_FILE_LOCATION);
 273       return false;
 274    }
 275
 276    test::Request& request = a_requests.get ();
 277
 278    for (int n = 0; n < a_messageBySecond && communicator->hasRequestedStop () == false; n ++) {
 279       request.op = '+';
 280       request.x = rand () % 1000;
 281       request.y = rand () % 1000;
 282       request.initTime = anna::functions::millisecond ();
 283
 284       try {
 285          service->send (request);
 286          a_txMessageCounter ++;
 287       }
 288       catch (RuntimeException& ex) {
 289          a_errorCounter ++;
 290          ex.trace ();
 291          break;
 292       }
 293    }
 294
 295    return true;
 296 }
 297