File content as of revision 129:dc9e37d4bc05:

#include <stdio.h>
#include <ctype.h>
#include "PollThread.h"
#include "logging.h"

bool PollThread::handshake()
{
        int l = snprintf(buf2, sizeof(buf2), "%s", "80\r\n");
        l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
        sock.setBlocking(3000);
        l = sock.sendAndReceive(buf, l, sizeof(buf));
        if (l <= 0) return false;
        const char* p = skipHTTPHeader(buf);
        if (p == NULL) return false;
        size_t i = 0;
        for (; isalnum(*p); ++p, ++i) {
                bayeuxId[i] = *p;
        }
        bayeuxId[i] = 0;
        return bayeuxId[0];
}

bool PollThread::subscribe()
{
        int l = snprintf(buf2, sizeof(buf2), "81,%s,%s\r\n", bayeuxId, chn);
        l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
        sock.setBlocking(3000);
        l = sock.sendAndReceive(buf, l, sizeof(buf));
        if (l <= 0) return false;
        const char *p = skipHTTPHeader(buf);
        if (p == NULL) return false;
        for (; *p && !isgraph(*p); ++p);
        return *p == 0;
}

bool PollThread::connect()
{
        int l = snprintf(buf2, sizeof(buf2), "83,%s\r\n", bayeuxId);
        l = snprintf(buf, sizeof(buf), fmtSmartRest, uri, l, buf2);
        sock.setBlocking(610000); // Timeout after 10m:10s
//        sock.setBlocking(-1);
//        l = sock.sendAndReceive(buf, l, sizeof(buf));
        int c = sock.connect();
        if (c < 0) {
                sock.close();
                return false;
        }
        c = sock.send(buf, l);
        if (c < 0) {
                sock.close();
                return false;
        }
        // Avoid heartbeat message overflows the buffer
        l = 0;
        for (unsigned short i = 0; i < 200; ++i) {
                int l2 = sock.receive(buf+l, sizeof(buf)-l);
                printf("l2: %d\n", l2);
                if (l2 > 0) {
                        printf("poll: %.*s\n", l2, buf+l);
                        l += l2;
                } else if (!sock.is_connected()) {
                        break;
                } else {
                        l = l2;
                        break;
                }
        }
        if (l >= 0)
                buf[l] = 0;
        sock.close();
        return l>0;
}

void PollThread::threadFunc()
{
        unsigned short state = 1;
        while (true) {
                switch (state) {
                case 1: if (!handshake()) {
                                aCritical("Poll: handshake fail!\n");
                                break;
                        }
                case 2: if(!subscribe()) {
                                aCritical("Poll: subscribe fail!\n");
                                state = 1;
                                break;
                        }
                case 3: if(!connect()) {
                                aCritical("Poll: connect fail!\n");
                                state = 1;
                                break;
                        }
                default: parser.parse(buf);
                        if (parser.getBayeuxAdvice() == BA_HANDSHAKE)
                                state = 1;
                        else
                                state = 3;
                }
        }
}