diff options
| -rw-r--r-- | Makefile | 1 | ||||
| -rw-r--r-- | includes/ZHandle.h | 37 | ||||
| -rw-r--r-- | src/ZHandle.cc | 140 |
3 files changed, 178 insertions, 0 deletions
@@ -129,6 +129,7 @@ Output.cc \ Socket.cc \ Buffer.cc \ BStream.cc \ +ZHandle.cc \ \ Task.cc \ TaskMan.cc \ diff --git a/includes/ZHandle.h b/includes/ZHandle.h new file mode 100644 index 0000000..63da59a --- /dev/null +++ b/includes/ZHandle.h @@ -0,0 +1,37 @@ +#pragma once + +#include <zlib.h> +#include <Handle.h> + +namespace Balau { + +class ZStream : public Handle { + public: + typedef enum { + ZLIB, + GZIP, + RAW, + } header_t; + ZStream(const IO<Handle> & h, int level = Z_BEST_COMPRESSION, header_t header = ZLIB); + virtual void close() throw (GeneralException); + virtual bool isClosed(); + virtual bool isEOF(); + virtual bool canRead(); + virtual bool canWrite(); + virtual const char * getName(); + virtual ssize_t read(void * buf, size_t count) throw (GeneralException); + virtual ssize_t write(const void * buf, size_t count) throw (GeneralException); + virtual off_t getSize(); + void detach() { m_detached = true; } + void flush() { doFlush(false); } + private: + void finish() { doFlush(true); } + void doFlush(bool finish); + IO<Handle> m_h; + z_stream m_zin, m_zout; + bool m_detached, m_closed, m_eof; + String m_name; + uint8_t * m_in; +}; + +}; diff --git a/src/ZHandle.cc b/src/ZHandle.cc new file mode 100644 index 0000000..7e1d139 --- /dev/null +++ b/src/ZHandle.cc @@ -0,0 +1,140 @@ +#include "ZHandle.h" +#include "Task.h" + +Balau::ZStream::ZStream(const IO<Handle> & h, int level, header_t header) : m_h(h), m_detached(false), m_closed(false), m_eof(false), m_in(NULL) { + m_zin.zalloc = m_zout.zalloc = NULL; + m_zin.zfree = m_zout.zfree = NULL; + m_zin.opaque = m_zout.opaque = NULL; + m_zin.next_in = NULL; + m_zin.avail_in = 0; + int window = 0; + switch (header) { + case ZLIB: window = 15; break; + case GZIP: window = 31; break; + case RAW: window = -15; break; + } + int r; + r = inflateInit2(&m_zin, window); + EAssert(r == Z_OK, "inflateInit2 returned %i", r); + r = deflateInit2(&m_zout, level, Z_DEFLATED, window, 9, Z_DEFAULT_STRATEGY); + EAssert(r == Z_OK, "deflateInit2 returned %i", r); + m_name.set("ZStream(%s)", m_h->getName()); +} + +void Balau::ZStream::close() throw (GeneralException) { + if (m_in) { + free(m_in); + m_in = NULL; + } + if (m_h->canWrite()) + finish(); + inflateEnd(&m_zin); + deflateEnd(&m_zout); + if (!m_detached) + m_h->close(); + m_closed = true; +} + +bool Balau::ZStream::isClosed() { + return m_closed; +} + +bool Balau::ZStream::isEOF() { + if (m_closed || m_eof) + return true; + return m_h->isEOF(); +} + +bool Balau::ZStream::canRead() { + return m_h->canRead(); +} + +bool Balau::ZStream::canWrite() { + return m_h->canWrite(); +} + +const char * Balau::ZStream::getName() { + return m_name.to_charp(); +} + +static const int BLOCK_SIZE = 1024; + +ssize_t Balau::ZStream::read(void * buf, size_t count) throw (GeneralException) { + if (m_closed || m_eof) + return 0; + + AAssert(m_h->canRead(), "Can't call ZStream::read on a non-readable handle."); + + size_t readTotal = 0; + m_zin.next_out = (Bytef *) buf; + m_zin.avail_out = count; + if (!m_in) { + m_zin.next_in = m_in = (uint8_t *) malloc(BLOCK_SIZE); + m_zin.avail_in = 0; + } + while ((count != 0) && !m_h->isClosed() && !m_h->isEOF()) { + if (m_zin.avail_in == 0) { + m_zin.next_in = m_in; + size_t r = m_h->read(m_in, BLOCK_SIZE); + if (r <= 0) + return readTotal; + } + Task::yield(NULL); + int r = inflate(&m_zin, Z_SYNC_FLUSH); + size_t didRead = count - m_zin.avail_out; + readTotal += didRead; + count -= didRead; + if (r == Z_STREAM_END) { + m_eof = true; + break; + } + EAssert(r == Z_OK, "inflate() didn't return Z_OK but %i", r); + } + return readTotal; +} + +ssize_t Balau::ZStream::write(const void * buf, size_t count) throw (GeneralException) { + if (m_closed || m_eof) + return 0; + + AAssert(m_h->canWrite(), "Can't call ZStream::write on a non-writable handle."); + + size_t wroteTotal = 0; + m_zout.next_in = (Bytef *) const_cast<void *>(buf); + m_zout.avail_in = count; + void * obuf = alloca(BLOCK_SIZE); + while ((count != 0) && !m_h->isClosed() && !m_h->isEOF()) { + m_zout.next_out = (Bytef *) obuf; + m_zout.avail_out = BLOCK_SIZE; + int r = deflate(&m_zout, Z_NO_FLUSH); + EAssert(r == Z_OK, "deflate() didn't return Z_OK but %i", r); + Task::yield(NULL); + size_t compressed = BLOCK_SIZE - m_zout.avail_out; + if (compressed) { + size_t w = m_h->write(obuf, compressed); + if (w <= 0) + return wroteTotal; + } + wroteTotal += compressed; + count -= compressed; + } + return wroteTotal; +} + +void Balau::ZStream::doFlush(bool finish) { + void * buf = alloca(BLOCK_SIZE); + m_zout.next_in = NULL; + m_zout.avail_in = 0; + int r; + do { + m_zout.next_out = (Bytef *) buf; + m_zout.avail_out = BLOCK_SIZE; + r = deflate(&m_zout, finish ? Z_FINISH : Z_SYNC_FLUSH); + size_t compressed = BLOCK_SIZE - m_zout.avail_out; + if (compressed) { + size_t w = m_h->write(buf, compressed); + if (w <= 0) + return; + } + } while (r == Z_OK && finish); +} |