#include <grpc/support/port_platform.h>
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/compression/message_compress.h"
#include <string.h>
#include <zconf.h>
#include <memory>
#include <zconf.h>
#include <zlib.h>
#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/lib/slice/slice.h"
namespace grpc_core {
class GzipCompressionOptionsImpl : public CompressionOptions {
public:
GzipCompressionOptionsImpl(): gzip_compression_level_(Z_DEFAULT_COMPRESSION) {}
explicit GzipCompressionOptionsImpl(const ChannelArgs& args)
: gzip_compression_level_(
Clamp(
args.GetInt(GRPC_ARG_GZIP_COMPRESSION_LEVEL).value_or(Z_DEFAULT_COMPRESSION),
Z_NO_COMPRESSION, Z_BEST_COMPRESSION)) {}
int gzip_compression_level() const { return gzip_compression_level_; }
private:
int gzip_compression_level_;
};
std::unique_ptr<CompressionOptions> MakeCompressionOptions(
const ChannelArgs& args) {
return absl::make_unique<GzipCompressionOptionsImpl>(args);
}
std::unique_ptr<CompressionOptions> MakeCompressionOptions() {
return absl::make_unique<GzipCompressionOptionsImpl>();
}
}
#define OUTPUT_BLOCK_SIZE 1024
static int zlib_body(z_stream* zs, grpc_slice_buffer* input,
grpc_slice_buffer* output,
int (*flate)(z_stream* zs, int flush)) {
int r = Z_STREAM_END; int flush;
size_t i;
grpc_slice outbuf = GRPC_SLICE_MALLOC(OUTPUT_BLOCK_SIZE);
const uInt uint_max = ~uInt{0};
GPR_ASSERT(GRPC_SLICE_LENGTH(outbuf) <= uint_max);
zs->avail_out = static_cast<uInt> GRPC_SLICE_LENGTH(outbuf);
zs->next_out = GRPC_SLICE_START_PTR(outbuf);
flush = Z_NO_FLUSH;
for (i = 0; i < input->count; i++) {
if (i == input->count - 1) flush = Z_FINISH;
GPR_ASSERT(GRPC_SLICE_LENGTH(input->slices[i]) <= uint_max);
zs->avail_in = static_cast<uInt> GRPC_SLICE_LENGTH(input->slices[i]);
zs->next_in = GRPC_SLICE_START_PTR(input->slices[i]);
do {
if (zs->avail_out == 0) {
grpc_slice_buffer_add_indexed(output, outbuf);
outbuf = GRPC_SLICE_MALLOC(OUTPUT_BLOCK_SIZE);
GPR_ASSERT(GRPC_SLICE_LENGTH(outbuf) <= uint_max);
zs->avail_out = static_cast<uInt> GRPC_SLICE_LENGTH(outbuf);
zs->next_out = GRPC_SLICE_START_PTR(outbuf);
}
r = flate(zs, flush);
if (r < 0 && r != Z_BUF_ERROR ) {
gpr_log(GPR_INFO, "zlib error (%d)", r);
goto error;
}
} while (zs->avail_out == 0);
if (zs->avail_in) {
gpr_log(GPR_INFO, "zlib: not all input consumed");
goto error;
}
}
if (r != Z_STREAM_END) {
gpr_log(GPR_INFO, "zlib: Data error");
goto error;
}
GPR_ASSERT(outbuf.refcount);
outbuf.data.refcounted.length -= zs->avail_out;
grpc_slice_buffer_add_indexed(output, outbuf);
return 1;
error:
grpc_core::CSliceUnref(outbuf);
return 0;
}
static void* zalloc_gpr(void* , unsigned int items,
unsigned int size) {
return gpr_malloc(items * size);
}
static void zfree_gpr(void* , void* address) { gpr_free(address); }
static int zlib_compress(grpc_slice_buffer* input, grpc_slice_buffer* output,
int gzip,
const grpc_core::GzipCompressionOptionsImpl* options) {
z_stream zs;
int r;
size_t i;
size_t count_before = output->count;
size_t length_before = output->length;
memset(&zs, 0, sizeof(zs));
zs.zalloc = zalloc_gpr;
zs.zfree = zfree_gpr;
r = deflateInit2(&zs, options->gzip_compression_level(), Z_DEFLATED,
15 | (gzip ? 16 : 0), 8, Z_DEFAULT_STRATEGY);
GPR_ASSERT(r == Z_OK);
r = zlib_body(&zs, input, output, deflate) && output->length < input->length;
if (!r) {
for (i = count_before; i < output->count; i++) {
grpc_core::CSliceUnref(output->slices[i]);
}
output->count = count_before;
output->length = length_before;
}
deflateEnd(&zs);
return r;
}
static int zlib_decompress(grpc_slice_buffer* input, grpc_slice_buffer* output,
int gzip) {
z_stream zs;
int r;
size_t i;
size_t count_before = output->count;
size_t length_before = output->length;
memset(&zs, 0, sizeof(zs));
zs.zalloc = zalloc_gpr;
zs.zfree = zfree_gpr;
r = inflateInit2(&zs, 15 | (gzip ? 16 : 0));
GPR_ASSERT(r == Z_OK);
r = zlib_body(&zs, input, output, inflate);
if (!r) {
for (i = count_before; i < output->count; i++) {
grpc_core::CSliceUnref(output->slices[i]);
}
output->count = count_before;
output->length = length_before;
}
inflateEnd(&zs);
return r;
}
static int copy(grpc_slice_buffer* input, grpc_slice_buffer* output) {
size_t i;
for (i = 0; i < input->count; i++) {
grpc_slice_buffer_add(output, grpc_core::CSliceRef(input->slices[i]));
}
return 1;
}
static int compress_inner(
grpc_compression_algorithm algorithm, grpc_slice_buffer* input,
grpc_slice_buffer* output,
const grpc_core::GzipCompressionOptionsImpl* options) {
switch (algorithm) {
case GRPC_COMPRESS_NONE:
return 0;
case GRPC_COMPRESS_DEFLATE:
return zlib_compress(input, output, 0, options);
case GRPC_COMPRESS_GZIP:
return zlib_compress(input, output, 1, options);
case GRPC_COMPRESS_ALGORITHMS_COUNT:
break;
}
gpr_log(GPR_ERROR, "invalid compression algorithm %d", algorithm);
return 0;
}
int grpc_msg_compress(grpc_compression_algorithm algorithm,
grpc_slice_buffer* input, grpc_slice_buffer* output,
const grpc_core::CompressionOptions* options) {
auto compression_options =
static_cast<const grpc_core::GzipCompressionOptionsImpl*>(options);
if (!compress_inner(algorithm, input, output, compression_options)) {
copy(input, output);
return 0;
}
return 1;
}
int grpc_msg_decompress(grpc_compression_algorithm algorithm,
grpc_slice_buffer* input, grpc_slice_buffer* output) {
switch (algorithm) {
case GRPC_COMPRESS_NONE:
return copy(input, output);
case GRPC_COMPRESS_DEFLATE:
return zlib_decompress(input, output, 0);
case GRPC_COMPRESS_GZIP:
return zlib_decompress(input, output, 1);
case GRPC_COMPRESS_ALGORITHMS_COUNT:
break;
}
gpr_log(GPR_ERROR, "invalid compression algorithm %d", algorithm);
return 0;
}