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();
 
    try {
        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) {
                if ((rBuff + 1) == frame->endBuffer())
                    throw(rogue::GeneralError::create(
                        "StreamUnZip::acceptFrame",
                        "Truncated or corrupt bzip2 stream: decompressor needs more input than available"));
                ++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);
    } catch (...) {
        BZ2_bzDecompressEnd(&strm);
        throw;
    }
 
    const uint32_t totalOut = strm.total_out_lo32;
    BZ2_bzDecompressEnd(&strm);
 
    newFrame->setPayload(totalOut);
    newFrame->setError(frame->getError());
    newFrame->setChannel(frame->getChannel());
    newFrame->setFlags(frame->getFlags());
 
    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
}