GZCSVFileReader.cpp

/*
** libutil++
** $Id: GZCSVFileReader.cpp 1653 2016-02-28 19:54:59Z sella $
** Copyright (c) 2011-2016 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

static const char rcsid[] __attribute__((used)) = "$Id: GZCSVFileReader.cpp 1653 2016-02-28 19:54:59Z sella $";

#include "GZCSVFileReader.h"

#include <fcntl.h>

using namespace sella::file;

GZCSVFileReader::GZCSVFileReader(const std::string & file, char delimiter, char seperator) THROWS(IOException) :
        filename(file),

        delimiter(delimiter),
        seperator(seperator),

        descriptor(::open(file.c_str(), O_RDONLY | O_LARGEFILE))
{
        /* Ensure the file was opened for reading, and zero all buffers and state machines. */
        if (this->descriptor <= 0) {
                THROW(IOException, "failed to open file '%s' for reading", this->filename.c_str());
        }
        memset(&(this->stream), 0, sizeof(this->stream));
        memset(&(this->buffer), 0, sizeof(this->buffer));

        /* Initialize for the reading of GZIP compressed files. */
        if (::inflateInit2(&(this->stream), 16 + MAX_WBITS) != Z_OK) {
                ::close(this->descriptor);

                THROW(IOException, "failed to initilize zlib instance for reading file '%s'", this->filename.c_str());
        }
        this->stream.avail_out = sizeof(this->buffer[1]);
}

GZCSVFileReader::~GZCSVFileReader() NO_EXCEPTIONS {
        if (this->descriptor != 0) {
                ::inflateEnd(&(this->stream));

                this->close();
        }
}

void GZCSVFileReader::read(std::vector<std::vector<std::string>> & data) THROWS(IOException) {
        do {
                if ((data.empty() == false) && data.back().empty()) {
                        data.pop_back();
                }
                data.push_back(std::vector<std::string>());
        } while (this->read(data.back()) == true);

        if ((data.empty() == false) && data.back().empty()) {
                data.pop_back();
        }
}

bool GZCSVFileReader::read(std::vector<std::string> & data) THROWS(IOException) {
        /* Read field entries from the memory buffer, and as necessary fall back to reading from the disk. */
        while (this->read(data, std::string()) == false) {
                /* EMPTY LOOP */
        }
        return data.empty() ? false : true;
}

void GZCSVFileReader::close(bool unlink) NO_EXCEPTIONS {
        ::close(this->descriptor);

        if (unlink == true) {
                ::unlink(this->filename.c_str());
        }
        this->filename.clear();
        this->descriptor = 0;
}

bool GZCSVFileReader::read(std::vector<std::string> & data, std::string prepend) THROWS(IOException) {
        uint8_t * head = this->stream.next_out - (sizeof(this->buffer[1]) - this->stream.avail_out);

        /* Determine if there is decopmressed data available. */
        if (this->stream.avail_out == sizeof(this->buffer[1])) {
                /* There is no decopressed data available, so determine if there is compressed data available in memory. */
                if (this->stream.avail_in == 0) {
                        /* There is no compressed data available, so read data from file into the compressed data buffer. */
                        if ((this->stream.avail_in = ::read(this->descriptor, this->buffer[0], sizeof(this->buffer[0]))) < 0) {
                                THROW(IOException, "failed to read from file '%s'", this->filename.c_str());
                        } else if (this->stream.avail_in == 0) {
                                return true; /* The end of the file was reached, and no more data could be read. */
                        }
                        this->stream.next_in = this->buffer[0];
                }

                /* Decompress data from the compressed buffer into the uncompressed buffer. */
                this->stream.avail_out = sizeof(this->buffer[1]);
                this->stream.next_out = this->buffer[1];

                switch (::inflate(&(this->stream), Z_SYNC_FLUSH)) {
                        case Z_STREAM_ERROR:
                                THROW(IOException, "zlib decompression state was destroyed");

                        case Z_NEED_DICT:
                                THROW(IOException, "zlib decompression failed due to missing dictionary");

                        case Z_DATA_ERROR:
                                THROW(IOException, "zlib decompression failed due to bad data in stream");

                        case Z_MEM_ERROR:
                                THROW(IOException, "zlib decompression failed due to a memory error");
                }
                head = this->stream.next_out - (sizeof(this->buffer[1]) - this->stream.avail_out);
        }

        /* Adjust the limits of the decompressed memory buffer. */
        for (uint8_t * i = head; i != this->stream.next_out; i++) {
                if ((*i) == this->seperator) { /* The end of the record has been reached; store the last field and return success. */
                        data.push_back(prepend.append(std::string(reinterpret_cast<char *>(head), i - head)));
                        this->stream.avail_out += (i - head + 1);

                        return true; /* The record request has been fullfilled. */
                } else if (*(i) == this->delimiter) { /* A record delimiter has been identified. */
                        data.push_back(prepend.append(std::string(reinterpret_cast<char *>(head), i - head)));
                        this->stream.avail_out += (i - head + 1);

                        return false; /* The record request has not yet been fullfilled. */
                }
        }
        this->stream.avail_out = sizeof(this->buffer[1]);

        /* Recursively call this function with a prepend value supplied to account for a field which is too long for the buffers. */
        return this->read(data, std::string(reinterpret_cast<char *>(head), this->stream.next_out - head));
}

/*
** vim: noet ts=3 sw=3
*/