Extractor.cpp

Go to the documentation of this file.

#include <Extractor.h>

#include <iostream>

using namespace std;

Extractor::Extractor(zmq::context_t & _ctx, const PubSubEndpoint & _source, double _wantedFPS):
        ctx(_ctx), source(_source), keepGoing(false) {
    if (_wantedFPS <= 0) {
        wantedFrameTime = 0;
    } else {
        wantedFrameTime = (uint64_t)(1e6/_wantedFPS);
    }
}

Extractor::~Extractor() {

}

void Extractor::startServing() {
    keepGoing = true;
    extractorThread = thread(&Extractor::extract, this);
}

void Extractor::stopServing() {
    cout << "stop serving" << endl;
    keepGoing = false;
    if(extractorThread.joinable()) {
        extractorThread.join();
    }
    cout << "extractor thread joined" << endl;
}

void Extractor::extractSetup() {

}

void Extractor::extract() {
    extractSetup();
    {
        zmq::socket_t sourceSocket(ctx, ZMQ_SUB);
        int linger =  0;
        sourceSocket.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
        sourceSocket.connect(source.endpoint.c_str());
        sourceSocket.connect("inproc://control");
        sourceSocket.setsockopt(ZMQ_SUBSCRIBE, source.topic.c_str(), source.topic.length());
        sourceSocket.setsockopt(ZMQ_SUBSCRIBE, "control", 7);

        int64_t more;
        size_t more_size = sizeof(more);
        cout << "starting extractor" << endl;
        
        zmq::pollitem_t items [] = {
            {(void*)sourceSocket, 0, ZMQ_POLLIN, 0}
        };
        while(keepGoing) {
            zmq::poll(&items[0], 1, 100);
            if(!(items[0].revents & ZMQ_POLLIN)) {
                continue;
            }
            zmq::message_t sourceTopic;
            sourceSocket.recv(&sourceTopic);
            if(strncmp((char*)sourceTopic.data(), source.topic.c_str(), source.topic.length()) == 0) {
                // got right data
                sourceSocket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
                if(!(more & 0xffffffff)) {
                    cout << "invalid message from topic " << source.topic << "!!!" << endl;
                    continue;
                }
                zmq::message_t sourceDimensions;
                sourceSocket.recv(&sourceDimensions);
                uint32_t width = readNetworkUInt32(((char*)sourceDimensions.data()) + 0);
                uint32_t height = readNetworkUInt32(((char*)sourceDimensions.data()) + 4);
                uint32_t bytesPerPixel = readNetworkUInt32(((char*)sourceDimensions.data()) + 8);
                uint64_t grabHighResTime = readNetworkUInt64(((char*)sourceDimensions.data()) + 12);
                uint64_t grabWallTime = readNetworkUInt64(((char*)sourceDimensions.data()) + 20);

                sourceSocket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
                if(!(more & 0xffffffff)) {
                    cout << "invalid message from topic " << source.topic << "!!!" << endl;
                    continue;
                }
                zmq::message_t * sourceImage = new zmq::message_t;
                sourceSocket.recv(sourceImage);
                
                if (grabHighResTime >= nextFrameTime) {
                    nextFrameTime += wantedFrameTime;
                    if (grabHighResTime > nextFrameTime) {
                        nextFrameTime = grabHighResTime + wantedFrameTime;
                    }
                    extractImpl(width, height, bytesPerPixel, grabHighResTime, grabWallTime, sourceImage);
                } else {
                    delete sourceImage;
                }
                
            } else if(strncmp((char*)sourceTopic.data(), "control", 7) == 0) {
                sourceSocket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
                if(!(more & 0xffffffff)) {
                    cout << "invalid message from topic " << source.topic << "!!!" << endl;
                    continue;
                }
                zmq::message_t controlMessage;
                sourceSocket.recv(&controlMessage);
                if(strncmp((char*)controlMessage.data(), "stop", 4) == 0) {
                    cout << "stoping extractor" << endl;
                    break;
                }
            } else {
                // got some unexpected data
            }
            // flush remaining multipart data if any
            sourceSocket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
            while((more & 0xffffffff)) { /* this is a hack, ZMQ should either return 0 or 1 but more_size gets strangely
                                            reset to 4 which results in garbage in the upper 4 bytes */
                zmq::message_t dummy;
                sourceSocket.recv(&dummy);
                sourceSocket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
            }
        }
    }
    extractTeardown();
}

void Extractor::extractTeardown() {

}