/*** Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved. File: ZSTDCompressor.hpp Date: 2022-2-15 Author: Reece ***/ #pragma once #include "zstd.h" namespace Aurora::Compression { struct ZSTDDeflate : BaseStream { CompressInfo meta; ZSTDDeflate(const CompressInfo &meta) : meta(meta), BaseStream(meta.uInternalStreamSize) {} ~ZSTDDeflate() { if (auto cctx = AuExchange(this->cctx_, {})) { ZSTD_freeCCtx(cctx); } } bool Init(const AuSPtr &pReader) override { AuUInt uRet; if (!this->IsValid()) { SysPushErrorMem(); return false; } this->pReader_ = pReader; this->cctx_ = ZSTD_createCCtx(); if (!this->cctx_) { SysPushErrorGen("Couldn't create compressor"); } uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_compressionLevel, meta.uCompressionLevel); if (ZSTD_isError(uRet)) { SysPushErrorGen("Invalid compression level"); return false; } uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_checksumFlag, 1); if (ZSTD_isError(uRet)) { SysPushErrorGen("Invalid option"); return false; } uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_nbWorkers, AuMax(meta.uThreads, AuUInt8(1u))); if (ZSTD_isError(uRet)) { SysPushErrorGen(); return false; } this->pIterator_ = this->din_; this->uAvailableIn_ = 0; SetArray(this->din_); return true; } AuStreamReadWrittenPair_t Ingest_s(AuUInt32 input) override { AuUInt32 uLength = AuUInt32(ZSTD_DStreamInSize()); AuUInt32 uOutFrameLength = AuUInt32(ZSTD_DStreamOutSize()); AuUInt32 done {}, read {}; if (!this->pReader_) { return {}; } while (read < input) { read += IngestForInPointer(this->pReader_, this->pIterator_, this->uAvailableIn_, input - read); if (!this->uAvailableIn_) { return {read, done}; } this->input_ = ZSTD_inBuffer {this->pIterator_, this->uAvailableIn_, 0}; while (this->input_.pos < this->input_.size) { ZSTD_outBuffer output = {this->dout_, uOutFrameLength, 0}; auto uRet = ZSTD_compressStream(this->cctx_, &output, &this->input_); if (ZSTD_isError(uRet)) { SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet)); this->uAvailableIn_ -= AuUInt32(output.pos); this->pReader_.reset(); return AuMakePair(read, 0); } if (!output.pos) { continue; } done += AuUInt32(output.pos); if (!Write(reinterpret_cast(this->dout_), AuUInt32(output.pos))) { this->pReader_.reset(); return AuMakePair(read, 0); } } this->uAvailableIn_ -= AuUInt32(this->input_.pos); } return {read, done}; } bool Flush() override { return RunFlush(ZSTD_e_continue); } bool Finish() override { return RunFlush(ZSTD_e_end); } bool RunFlush(ZSTD_EndDirective type) { if (!this->pReader_) { return false; } AuUInt32 uLength = AuUInt32(ZSTD_DStreamInSize()); AuUInt32 uOutFrameLength = AuUInt32(ZSTD_DStreamOutSize()); if (!this->uAvailableIn_) { ZSTD_outBuffer output = { this->dout_, uOutFrameLength, 0 }; ZSTD_inBuffer input = { NULL, 0, 0 }; auto uRet = ZSTD_compressStream2(this->cctx_, &output, &input, type); if (ZSTD_isError(uRet)) { SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet)); this->uAvailableIn_ -= AuUInt32(output.pos); return {}; } if (!Write(reinterpret_cast(output.dst), AuUInt32(output.pos))) { this->pReader_.reset(); return false; } return true; } this->input_ = ZSTD_inBuffer {this->pIterator_, this->uAvailableIn_, 0}; while (this->input_.pos < this->input_.size) { ZSTD_outBuffer output = {this->dout_, uOutFrameLength, 0}; auto uRet = ZSTD_compressStream(this->cctx_, &output, &this->input_); if (ZSTD_isError(uRet)) { SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet)); this->uAvailableIn_ -= AuUInt32(output.pos); return {}; } if (!Write(reinterpret_cast(this->dout_), AuUInt32(output.pos))) { this->pReader_.reset(); return false; } } this->uAvailableIn_ -= AuUInt32(this->input_.pos); return true; } private: AuSPtr pReader_; ZSTD_CCtx *cctx_ {}; char din_[ZSTD_BLOCKSIZE_MAX]; char dout_[ZSTD_BLOCKSIZE_MAX + 3 /*ZSTD_BLOCKHEADERSIZE*/]; char *pIterator_ {}; AuUInt32 uAvailableIn_ {}; ZSTD_inBuffer input_; }; }