Implemented aysnc parser in STG::Conn.

diff --git a/projects/stargazer/plugins/other/radius/conn.cpp b/projects/stargazer/plugins/other/radius/conn.cpp
index 392624a6ef4846dd8d4c424baca068638e4ee4a2..99aa83b6cd59cc4cf69780809e176e43a1185839 100644 (file)
--- a/projects/stargazer/plugins/other/radius/conn.cpp
+++ b/projects/stargazer/plugins/other/radius/conn.cpp
@@ -36,6 +36,8 @@
 #include <cerrno>
 
 #include <unistd.h>
 #include <cerrno>
 
 #include <unistd.h>

+#include <sys/types.h>
+#include <sys/socket.h>

 
 using STG::Conn;
 using STG::Config;
 
 using STG::Conn;
 using STG::Config;


@@ -50,8 +52,8 @@ using STG::JSON::StringGen;
 namespace
 {
 
 namespace
 {
 

-double CONN_TIMEOUT = 5;
-double PING_TIMEOUT = 1;
+double CONN_TIMEOUT = 60;
+double PING_TIMEOUT = 10;

 
 enum Packet
 {
 
 enum Packet
 {


@@ -105,10 +107,12 @@ class StageParser : public EnumParser<Stage>
 class TopParser : public NodeParser
 {
     public:
 class TopParser : public NodeParser
 {
     public:

-        TopParser()
+        typedef void (*Callback) (void* /*data*/);
+        TopParser(Callback callback, void* data)

             : m_packetParser(this, m_packet, m_packetStr),
               m_stageParser(this, m_stage, m_stageStr),
             : m_packetParser(this, m_packet, m_packetStr),
               m_stageParser(this, m_stage, m_stageStr),

-              m_pairsParser(this, m_data)
+              m_pairsParser(this, m_data),
+              m_callback(callback), m_callbackData(data)

         {}
 
         virtual NodeParser* parseStartMap() { return this; }
         {}
 
         virtual NodeParser* parseStartMap() { return this; }


@@ -125,7 +129,7 @@ class TopParser : public NodeParser
 
             return this;
         }
 
             return this;
         }

-        virtual NodeParser* parseEndMap() { return this; }
+        virtual NodeParser* parseEndMap() { m_callback(m_callbackData); return this; }

 
         const std::string& packetStr() const { return m_packetStr; }
         Packet packet() const { return m_packet; }
 
         const std::string& packetStr() const { return m_packetStr; }
         Packet packet() const { return m_packet; }


@@ -143,12 +147,18 @@ class TopParser : public NodeParser
         PacketParser m_packetParser;
         StageParser m_stageParser;
         PairsParser m_pairsParser;
         PacketParser m_packetParser;
         StageParser m_stageParser;
         PairsParser m_pairsParser;

+
+        Callback m_callback;
+        void* m_callbackData;

 };
 
 class ProtoParser : public Parser
 {
     public:
 };
 
 class ProtoParser : public Parser
 {
     public:

-        ProtoParser() : Parser( &m_topParser ) {}
+        ProtoParser(TopParser::Callback callback, void* data)
+            : Parser( &m_topParser ),
+              m_topParser(callback, data)
+        {}

 
         const std::string& packetStr() const { return m_topParser.packetStr(); }
         Packet packet() const { return m_topParser.packet(); }
 
         const std::string& packetStr() const { return m_topParser.packetStr(); }
         Packet packet() const { return m_topParser.packet(); }


@@ -182,6 +192,11 @@ class PacketGen : public Gen
             m_gen.add(key, map);
             return *this;
         }
             m_gen.add(key, map);
             return *this;
         }

+        PacketGen& add(const std::string& key, MapGen& map)
+        {
+            m_gen.add(key, map);
+            return *this;
+        }

     private:
         MapGen m_gen;
         StringGen m_type;
     private:
         MapGen m_gen;
         StringGen m_type;


@@ -213,10 +228,10 @@ class Conn::Impl
         time_t m_lastActivity;
         ProtoParser m_parser;
 
         time_t m_lastActivity;
         ProtoParser m_parser;
 

-        bool process();
-        bool processPing();
-        bool processPong();
-        bool processData();
+        static void process(void* data);
+        void processPing();
+        void processPong();
+        void processData();

         bool answer(const USER& user);
         bool answerNo();
         bool sendPing();
         bool answer(const USER& user);
         bool answerNo();
         bool sendPing();


@@ -262,7 +277,8 @@ Conn::Impl::Impl(USERS& users, PLUGIN_LOGGER& logger, const Config& config, int
       m_remote(remote),
       m_ok(true),
       m_lastPing(time(NULL)),
       m_remote(remote),
       m_ok(true),
       m_lastPing(time(NULL)),

-      m_lastActivity(m_lastPing)
+      m_lastActivity(m_lastPing),
+      m_parser(&Conn::Impl::process, this)

 {
 }
 
 {
 }
 


@@ -281,16 +297,12 @@ bool Conn::Impl::read()
         m_ok = false;
         return false;
     }
         m_ok = false;
         return false;
     }

+    printfd(__FILE__, "Read %d bytes.\n%s\n", res, std::string(buffer.data(), res).c_str());

     m_lastActivity = time(NULL);
     if (res == 0)
     {
     m_lastActivity = time(NULL);
     if (res == 0)
     {

-        if (!m_parser.done())
-        {
-            m_ok = false;
-            m_logger("Failed to read data from '" + m_remote + "': " + strerror(errno));
-            return false;
-        }
-        return process();
+        m_ok = false;
+        return true;

     }
     return m_parser.append(buffer.data(), res);
 }
     }
     return m_parser.append(buffer.data(), res);
 }


@@ -300,48 +312,60 @@ bool Conn::Impl::tick()
     time_t now = time(NULL);
     if (difftime(now, m_lastActivity) > CONN_TIMEOUT)
     {
     time_t now = time(NULL);
     if (difftime(now, m_lastActivity) > CONN_TIMEOUT)
     {

+        int delta = difftime(now, m_lastActivity);
+        printfd(__FILE__, "Connection to '%s' timed out: %d sec.\n", m_remote.c_str(), delta);

         m_logger("Connection to " + m_remote + " timed out.");
         m_ok = false;
         return false;
     }
     if (difftime(now, m_lastPing) > PING_TIMEOUT)
         m_logger("Connection to " + m_remote + " timed out.");
         m_ok = false;
         return false;
     }
     if (difftime(now, m_lastPing) > PING_TIMEOUT)

+    {
+        int delta = difftime(now, m_lastPing);
+        printfd(__FILE__, "Ping timeout: %d sec. Sending ping...\n", delta);

         sendPing();
         sendPing();

+    }

     return true;
 }
 
     return true;
 }
 

-bool Conn::Impl::process()
+void Conn::Impl::process(void* data)

 {
 {

-    switch (m_parser.packet())
+    Impl& impl = *static_cast<Impl*>(data);
+    switch (impl.m_parser.packet())

     {
         case PING:
     {
         case PING:

-            return processPing();
+            impl.processPing();
+            return;

         case PONG:
         case PONG:

-            return processPong();
+            impl.processPong();
+            return;

         case DATA:
         case DATA:

-            return processData();
+            impl.processData();
+            return;

     }
     }

-    m_logger("Received invalid packet type: " + m_parser.packetStr());
-    return false;
+    printfd(__FILE__, "Received invalid packet type: '%s'.\n", impl.m_parser.packetStr().c_str());
+    impl.m_logger("Received invalid packet type: " + impl.m_parser.packetStr());

 }
 
 }
 

-bool Conn::Impl::processPing()
+void Conn::Impl::processPing()

 {
 {

-    return sendPong();
+    printfd(__FILE__, "Got ping. Sending pong...\n");
+    sendPong();

 }
 
 }
 

-bool Conn::Impl::processPong()
+void Conn::Impl::processPong()

 {
 {

+    printfd(__FILE__, "Got pong.\n");

     m_lastActivity = time(NULL);
     m_lastActivity = time(NULL);

-    return true;

 }
 
 }
 

-bool Conn::Impl::processData()
+void Conn::Impl::processData()

 {
 {

+    printfd(__FILE__, "Got data.\n");

     int handle = m_users.OpenSearch();
 
     USER_PTR user = NULL;
     int handle = m_users.OpenSearch();
 
     USER_PTR user = NULL;

-    bool match = true;
-    while (m_users.SearchNext(handle, &user))
+    bool match = false;
+    while (m_users.SearchNext(handle, &user) == 0)

     {
         if (user == NULL)
             continue;
     {
         if (user == NULL)
             continue;


@@ -371,24 +395,23 @@ bool Conn::Impl::processData()
         answerNo();
 
     m_users.CloseSearch(handle);
         answerNo();
 
     m_users.CloseSearch(handle);

-
-    return true;

 }
 
 bool Conn::Impl::answer(const USER& user)
 {
 }
 
 bool Conn::Impl::answer(const USER& user)
 {

-    boost::scoped_ptr<MapGen> reply(new MapGen);
+    printfd(__FILE__, "Got match. Sending answer...\n");
+    MapGen reply;

     for (Config::Pairs::const_iterator it = m_config.reply.begin(); it != m_config.reply.end(); ++it)
     for (Config::Pairs::const_iterator it = m_config.reply.begin(); it != m_config.reply.end(); ++it)

-        reply->add(it->first, new StringGen(user.GetParamValue(it->second)));
+        reply.add(it->first, new StringGen(user.GetParamValue(it->second)));

 
 

-    boost::scoped_ptr<MapGen> modify(new MapGen);
+    MapGen modify;

     for (Config::Pairs::const_iterator it = m_config.modify.begin(); it != m_config.modify.end(); ++it)
     for (Config::Pairs::const_iterator it = m_config.modify.begin(); it != m_config.modify.end(); ++it)

-        modify->add(it->first, new StringGen(user.GetParamValue(it->second)));
+        modify.add(it->first, new StringGen(user.GetParamValue(it->second)));

 
     PacketGen gen("data");
     gen.add("result", "ok")
 
     PacketGen gen("data");
     gen.add("result", "ok")

-       .add("reply", reply.get())
-       .add("modify", modify.get());
+       .add("reply", reply)
+       .add("modify", modify);

 
     m_lastPing = time(NULL);
 
 
     m_lastPing = time(NULL);
 


@@ -397,8 +420,9 @@ bool Conn::Impl::answer(const USER& user)
 
 bool Conn::Impl::answerNo()
 {
 
 bool Conn::Impl::answerNo()
 {

+    printfd(__FILE__, "No match. Sending answer...\n");

     PacketGen gen("data");
     PacketGen gen("data");

-    gen.add("result", "ok");
+    gen.add("result", "no");

 
     m_lastPing = time(NULL);
 
 
     m_lastPing = time(NULL);
 


@@ -425,10 +449,12 @@ bool Conn::Impl::sendPong()
 
 bool Conn::Impl::write(void* data, const char* buf, size_t size)
 {
 
 bool Conn::Impl::write(void* data, const char* buf, size_t size)
 {

+    std::string json(buf, size);
+    printfd(__FILE__, "Writing JSON:\n%s\n", json.c_str());

     Conn::Impl& conn = *static_cast<Conn::Impl*>(data);
     while (size > 0)
     {
     Conn::Impl& conn = *static_cast<Conn::Impl*>(data);
     while (size > 0)
     {

-        ssize_t res = ::write(conn.m_sock, buf, size);
+        ssize_t res = ::send(conn.m_sock, buf, size, MSG_NOSIGNAL);

         if (res < 0)
         {
             conn.m_logger("Failed to write pong to '" + conn.m_remote + "': " + strerror(errno));
         if (res < 0)
         {
             conn.m_logger("Failed to write pong to '" + conn.m_remote + "': " + strerror(errno));