StreamUnZip.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/utilities/StreamUnZip.h"
 
#include <bzlib.h>
#include <inttypes.h>
#include <stdarg.h>
#include <unistd.h>
 
#include <memory>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/interfaces/stream/Buffer.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/FrameLock.h"
#include "rogue/interfaces/stream/Master.h"
#include "rogue/interfaces/stream/Slave.h"
 
namespace ris = rogue::interfaces::stream;
namespace ru  = rogue::utilities;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
ru::StreamUnZipPtr ru::StreamUnZip::create() {
    ru::StreamUnZipPtr p = std::make_shared<ru::StreamUnZip>();
    return (p);
}
 
ru::StreamUnZip::StreamUnZip() {}
 
ru::StreamUnZip::~StreamUnZip() {}
 
void ru::StreamUnZip::acceptFrame(ris::FramePtr frame) {
    ris::Frame::BufferIterator rBuff;
    ris::Frame::BufferIterator wBuff;
    int32_t ret;
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr lock = frame->lock();
 
    // First request a new frame of the same size
    ris::FramePtr newFrame = this->reqFrame(frame->getPayload(), true);
 
    // Setup compression
    bz_stream strm;
    strm.bzalloc = NULL;
    strm.bzfree  = NULL;
    strm.opaque  = NULL;
 
    if ((ret = BZ2_bzDecompressInit(&strm, 0, 0)) != BZ_OK)
        throw(rogue::GeneralError::create("StreamUnZip::acceptFrame",
                                          "Error initializing decompressor. ret=%" PRIi32,
                                          ret));
 
    // Setup decompression pointers
    rBuff         = frame->beginBuffer();
    strm.next_in  = reinterpret_cast<char*>((*rBuff)->begin());
    strm.avail_in = (*rBuff)->getPayload();
 
    wBuff          = newFrame->beginBuffer();
    strm.next_out  = reinterpret_cast<char*>((*wBuff)->begin());
    strm.avail_out = (*wBuff)->getAvailable();
 
    // Use the iterators to move data
    do {
        ret = BZ2_bzDecompress(&strm);
 
        if ((ret != BZ_STREAM_END) && (ret != BZ_OK))
            throw(rogue::GeneralError::create("StreamUnZip::acceptFrame", "Decompression runtime error %" PRIi32, ret));
 
        if (ret == BZ_STREAM_END) break;
 
        // Update read buffer if necessary
        if (strm.avail_in == 0) {
            ++rBuff;
            strm.next_in  = reinterpret_cast<char*>((*rBuff)->begin());
            strm.avail_in = (*rBuff)->getPayload();
        }
 
        // Update write buffer if necessary
        if (strm.avail_out == 0) {
            // We ran out of room, double the frame size
            if ((wBuff + 1) == newFrame->endBuffer()) {
                ris::FramePtr tmpFrame = this->reqFrame(frame->getPayload(), true);
                wBuff                  = newFrame->appendFrame(tmpFrame);
            } else {
                ++wBuff;
            }
            strm.next_out  = reinterpret_cast<char*>((*wBuff)->begin());
            strm.avail_out = (*wBuff)->getAvailable();
        }
    } while (1);
 
    newFrame->setPayload(strm.total_out_lo32);
    newFrame->setError(frame->getError());
    newFrame->setChannel(frame->getChannel());
    newFrame->setFlags(frame->getFlags());
 
    BZ2_bzDecompressEnd(&strm);
 
    this->sendFrame(newFrame);
}
 
ris::FramePtr ru::StreamUnZip::acceptReq(uint32_t size, bool zeroCopyEn) {
    return (this->reqFrame(size, zeroCopyEn));
}
 
void ru::StreamUnZip::setup_python() {
#ifndef NO_PYTHON
 
    bp::class_<ru::StreamUnZip, ru::StreamUnZipPtr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>(
        "StreamUnZip",
        bp::init<>());
 
    bp::implicitly_convertible<ru::StreamUnZipPtr, ris::SlavePtr>();
    bp::implicitly_convertible<ru::StreamUnZipPtr, ris::MasterPtr>();
#endif
}