PollThread.cpp

00001 #include <stdio.h>
00002 #include <ctype.h>
00003 #include "PollThread.h"
00004 #include "logging.h"
00005 
00006 
00007 bool PollThread::handshake()
00008 {
00009         int l = snprintf(buf2, sizeof(buf2), "%s", "80\r\n");
00010         l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
00011         sock.setBlocking(3000);
00012         l = sock.sendAndReceive(buf, l, sizeof(buf));
00013         if (l <= 0) return false;
00014         const char* p = skipHTTPHeader(buf);
00015         if (p == NULL) return false;
00016         size_t i = 0;
00017         for (; isalnum(*p); ++p, ++i) {
00018                 bayeuxId[i] = *p;
00019         }
00020         bayeuxId[i] = 0;
00021         return bayeuxId[0];
00022 }
00023 
00024 
00025 bool PollThread::subscribe()
00026 {
00027         int l = snprintf(buf2, sizeof(buf2), "81,%s,%s\r\n", bayeuxId, chn);
00028         l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
00029         sock.setBlocking(3000);
00030         l = sock.sendAndReceive(buf, l, sizeof(buf));
00031         if (l <= 0) return false;
00032         const char *p = skipHTTPHeader(buf);
00033         if (p == NULL) return false;
00034         for (; *p && !isgraph(*p); ++p);
00035         return *p == 0;
00036 }
00037 
00038 
00039 bool PollThread::connect()
00040 {
00041         int l = snprintf(buf2, sizeof(buf2), "83,%s\r\n", bayeuxId);
00042         l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
00043         sock.setBlocking(610000); // Timeout after 10m:10s
00044 //        sock.setBlocking(-1);
00045 //        l = sock.sendAndReceive(buf, l, sizeof(buf));
00046         int c = sock.connect();
00047         if (c < 0) {
00048                 sock.close();
00049                 return false;
00050         }
00051         c = sock.send(buf, l);
00052         if (c < 0) {
00053                 sock.close();
00054                 return false;
00055         }
00056         // Avoid heartbeat message overflows the buffer
00057         l = 0;
00058         for (unsigned short i = 0; i < 300; ++i) {
00059                 int l2 = sock.receive(buf+l, sizeof(buf)-l);
00060                 if (l2 > 0) {
00061                         l += l2;
00062                 } else {
00063                         break;
00064                 }
00065         }
00066         sock.close();
00067         buf[l] = 0;
00068         return l>0;
00069 }
00070 
00071 
00072 void PollThread::threadFunc()
00073 {
00074         unsigned short state = 1;
00075         while (true) {
00076                 switch (state) {
00077                 case 1: if (!handshake()) {
00078                                 aCritical("Poll: handshake fail!\n");
00079                                 break;
00080                         }
00081                 case 2: if(!subscribe()) {
00082                                 aCritical("Poll: subscribe fail!\n");
00083                                 state = 1;
00084                                 break;
00085                         }
00086                 case 3: if(!connect()) {
00087                                 aCritical("Poll: connect fail!\n");
00088                                 break;
00089                         }
00090                         parser.parse(buf);
00091                         if (parser.getBayeuxAdvice() == BA_HANDSHAKE)
00092                                 state = 1;
00093                         else
00094                                 state = 3;
00095                 }
00096         }
00097 }