text: /*
   * librdkafka - The Apache Kafka C/C++ library
   *
   * Copyright (c) 2018 Magnus Edenhill
   * All rights reserved.
  * 
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions are met: 
   * 
   * 1. Redistributions of source code must retain the above copyright notice,
   *    this list of conditions and the following disclaimer. 
   * 2. Redistributions in binary form must reproduce the above copyright notice,
   *    this list of conditions and the following disclaimer in the documentation
   *    and/or other materials provided with the distribution. 
   * 
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
   * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
   * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
   * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
   * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
   * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
   * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
   * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   * POSSIBILITY OF SUCH DAMAGE.
   */
  
  #include "rdkafka_int.h"
  #include "rdkafka_zstd.h"
  
  #if WITH_ZSTD_STATIC
  /* Enable advanced/unstable API for initCStream_srcSize */
  #define ZSTD_STATIC_LINKING_ONLY
  #endif
  
  #include <zstd.h>
  #include <zstd_errors.h>
  
  rd_kafka_resp_err_t
  rd_kafka_zstd_decompress (rd_kafka_broker_t *rkb,
                           char *inbuf, size_t inlen,
                           void **outbuf, size_t *outlenp) {
          size_t out_bufsize = ZSTD_getFrameContentSize(inbuf, inlen);
  
          switch (out_bufsize) {
          case ZSTD_CONTENTSIZE_UNKNOWN:
                  /* Decompressed size cannot be determined, make a guess */
                  out_bufsize = inlen * 2;
                  break;
          case ZSTD_CONTENTSIZE_ERROR:
                  /* Error calculating frame content size */
                  rd_rkb_dbg(rkb, MSG, "ZSTD",
                             "Unable to begin ZSTD decompression "
                             "(out buffer is %"PRIusz" bytes): %s",
                             out_bufsize, "Error in determining frame size");
                  return RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
          default:
                  break;
          }
  
          /* Increase output buffer until it can fit the entire result,
           * capped by message.max.bytes */
          while (out_bufsize <= (size_t)rkb->rkb_rk->rk_conf.recv_max_msg_size) {
                  size_t ret;
                  char *decompressed;
  
                  decompressed = rd_malloc(out_bufsize);
                  if (!decompressed) {
                          rd_rkb_dbg(rkb, MSG, "ZSTD",
                                     "Unable to allocate output buffer "
                                     "(%"PRIusz" bytes for %"PRIusz
                                     " compressed bytes): %s",
                                     out_bufsize, inlen, rd_strerror(errno));
                          return RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE;
                  }
  
  
                  ret = ZSTD_decompress(decompressed, out_bufsize,
                                        inbuf, inlen);
                  if (!ZSTD_isError(ret)) {
                          *outlenp = ret;
                          *outbuf = decompressed;
                          return RD_KAFKA_RESP_ERR_NO_ERROR;
                  }
  
                  rd_free(decompressed);
  
                  /* Check if the destination size is too small */
                  if (ZSTD_getErrorCode(ret) == ZSTD_error_dstSize_tooSmall) {
  
                          /* Grow quadratically */
                          out_bufsize += RD_MAX(out_bufsize * 2, 4000);
  
                          rd_atomic64_add(&rkb->rkb_c.zbuf_grow, 1);
  
                  } else {
                          /* Fail on any other error */
                          rd_rkb_dbg(rkb, MSG, "ZSTD",
                                     "Unable to begin ZSTD decompression "
                                     "(out buffer is %"PRIusz" bytes): %s",
                                     out_bufsize, ZSTD_getErrorName(ret));
                          return RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                  }
          }
  
          rd_rkb_dbg(rkb, MSG, "ZSTD",
                     "Unable to decompress ZSTD "
                     "(input buffer %"PRIusz", output buffer %"PRIusz"): "
                     "output would exceed receive.message.max.bytes (%d)",
                     inlen, out_bufsize, rkb->rkb_rk->rk_conf.max_msg_size);
  
          return RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
  }
  
  
  rd_kafka_resp_err_t
  rd_kafka_zstd_compress (rd_kafka_broker_t *rkb, int comp_level,
                         rd_slice_t *slice, void **outbuf, size_t *outlenp) {
          ZSTD_CStream *cctx;
          size_t r;
          rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
          size_t len = rd_slice_remains(slice);
          ZSTD_outBuffer out;
          ZSTD_inBuffer in;
  
          *outbuf = NULL;
          out.pos = 0;
          out.size = ZSTD_compressBound(len);
          out.dst = rd_malloc(out.size);
          if (!out.dst) {
                  rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                             "Unable to allocate output buffer "
                             "(%"PRIusz" bytes): %s",
                             out.size, rd_strerror(errno));
                  return RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE;
          }
  
  
          cctx = ZSTD_createCStream();
          if (!cctx) {
                  rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                             "Unable to create ZSTD compression context");
                  err = RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE;
                  goto done;
          }
  
  #if defined(WITH_ZSTD_STATIC) && ZSTD_VERSION_NUMBER >= (1*100*100+2*100+1) /* v1.2.1 */
          r = ZSTD_initCStream_srcSize(cctx, comp_level, len);
  #else
          /* libzstd not linked statically (or zstd version < 1.2.1):
           * decompression in consumer may be more costly due to
           * decompressed size not included in header by librdkafka producer */
          r = ZSTD_initCStream(cctx, comp_level);
  #endif
          if (ZSTD_isError(r)) {
                  rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                             "Unable to begin ZSTD compression "
                             "(out buffer is %"PRIusz" bytes): %s",
                             out.size, ZSTD_getErrorName(r));
                  err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                  goto done;
          }
  
          while ((in.size = rd_slice_reader(slice, &in.src))) {
                  in.pos = 0;
                  r = ZSTD_compressStream(cctx, &out, &in);
                  if (unlikely(ZSTD_isError(r))) {
                          rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                                     "ZSTD compression failed "
                                     "(at of %"PRIusz" bytes, with "
                                     "%"PRIusz" bytes remaining in out buffer): "
                                     "%s",
                                     in.size, out.size - out.pos,
                                     ZSTD_getErrorName(r));
                          err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                          goto done;
                  }
  
                  /* No space left in output buffer,
                   * but input isn't fully consumed */
                  if (in.pos < in.size) {
                          err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                          goto done;
                  }
          }
  
          if (rd_slice_remains(slice) != 0) {
                  rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                             "Failed to finalize ZSTD compression "
                             "of %"PRIusz" bytes: %s",
                             len, "Unexpected trailing data");
                  err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                  goto done;
          }
  
          r = ZSTD_endStream(cctx, &out);
          if (unlikely(ZSTD_isError(r) || r > 0)) {
                  rd_rkb_dbg(rkb, MSG, "ZSTDCOMPR",
                             "Failed to finalize ZSTD compression "
                             "of %"PRIusz" bytes: %s",
                             len, ZSTD_getErrorName(r));
                  err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
                  goto done;
          }
  
          *outbuf  = out.dst;
          *outlenp = out.pos;
  
   done:
          if (cctx)
                  ZSTD_freeCStream(cctx);
  
          if (err)
                  rd_free(out.dst);
  
          return err;
  
  }
  