src/tscpp/api/GzipInflateTransformation.cc

/**
  Licensed to the Apache Software Foundation (ASF) under one
  or more contributor license agreements.  See the NOTICE file
  distributed with this work for additional information
  regarding copyright ownership.  The ASF licenses this file
  to you under the Apache License, Version 2.0 (the
  "License"); you may not use this file except in compliance
  with the License.  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.
 */

/**
 * @file GzipInflateTransformation.cc
 */

#include <cstring>
#include <vector>
#include <zlib.h>
#include <cinttypes>
#include <string_view>
#include "tscpp/api/TransformationPlugin.h"
#include "tscpp/api/GzipInflateTransformation.h"
#include "logging_internal.h"

using namespace atscppapi::transformations;
using std::vector;

namespace
{
const int    WINDOW_BITS          = 31; // Always use 31 for gzip.
unsigned int INFLATE_SCALE_FACTOR = 6;
} // namespace

/**
 * @private
 */
struct atscppapi::transformations::GzipInflateTransformationState : noncopyable {
  z_stream                   z_stream_;
  bool                       z_stream_initialized_;
  int64_t                    bytes_produced_;
  TransformationPlugin::Type transformation_type_;

  GzipInflateTransformationState(TransformationPlugin::Type type)
    : z_stream_initialized_(false), bytes_produced_(0), transformation_type_(type)
  {
    memset(&z_stream_, 0, sizeof(z_stream_));

    int err = inflateInit2(&z_stream_, WINDOW_BITS);
    if (Z_OK != err) {
      LOG_ERROR("inflateInit2 failed with error code '%d'.", err);
    } else {
      z_stream_initialized_ = true;
    }
  };

  ~GzipInflateTransformationState()
  {
    if (z_stream_initialized_) {
      int err = inflateEnd(&z_stream_);
      if (Z_OK != err && Z_STREAM_END != err) {
        LOG_ERROR("Unable to inflateEnd(), returned error code '%d'", err);
      }
    }
  };
};

GzipInflateTransformation::GzipInflateTransformation(Transaction &transaction, TransformationPlugin::Type type)
  : TransformationPlugin(transaction, type)
{
  state_ = std::make_unique<GzipInflateTransformationState>(type);
}

GzipInflateTransformation::~GzipInflateTransformation() {}

void
GzipInflateTransformation::consume(std::string_view data)
{
  if (data.size() == 0) {
    return;
  }

  if (!state_->z_stream_initialized_) {
    LOG_ERROR("Unable to inflate output because the z_stream was not initialized.");
    return;
  }

  int          err                = Z_OK;
  int          iteration          = 0;
  int          inflate_block_size = INFLATE_SCALE_FACTOR * data.size();
  vector<char> buffer(inflate_block_size);

  // Setup the compressed input
  state_->z_stream_.next_in  = reinterpret_cast<unsigned char *>(const_cast<char *>(data.data()));
  state_->z_stream_.avail_in = data.length();

  // Loop while we have more data to inflate
  while (state_->z_stream_.avail_in > 0 && err != Z_STREAM_END) {
    LOG_DEBUG("Iteration %d: Gzip has %d bytes to inflate", ++iteration, state_->z_stream_.avail_in);

    // Setup where the decompressed output will go
    // next_out needs to be set to nullptr before we return since it points to a local buffer
    // coverity[WRAPPER_ESCAPE: FALSE]
    state_->z_stream_.next_out  = reinterpret_cast<unsigned char *>(&buffer[0]);
    state_->z_stream_.avail_out = inflate_block_size;

    // Uncompress the data
    err = inflate(&state_->z_stream_, Z_SYNC_FLUSH);

    if (err != Z_OK && err != Z_STREAM_END) {
      LOG_ERROR("Iteration %d: Inflate failed with error '%d'", iteration, err);
      state_->z_stream_.next_out = nullptr;
      return;
    }

    LOG_DEBUG("Iteration %d: Gzip inflated a total of %d bytes, producingOutput...", iteration,
              (inflate_block_size - state_->z_stream_.avail_out));
    produce(std::string_view(&buffer[0], (inflate_block_size - state_->z_stream_.avail_out)));
    state_->bytes_produced_ += (inflate_block_size - state_->z_stream_.avail_out);
  }
  state_->z_stream_.next_out = nullptr;
}

void
GzipInflateTransformation::handleInputComplete()
{
  int64_t bytes_written = setOutputComplete();
  if (state_->bytes_produced_ != bytes_written) {
    LOG_ERROR("Gzip bytes produced sanity check failed, inflated bytes = %" PRId64 " != written bytes = %" PRId64,
              state_->bytes_produced_, bytes_written);
  }
}