Add interface abstraction and identity module

pull/12399/head
Muxi Yan 7 years ago
parent 31a4ebeded
commit 130e07061f
  1. 2
      build.yaml
  2. 6
      src/core/ext/transport/chttp2/transport/chttp2_transport.c
  3. 2
      src/core/ext/transport/chttp2/transport/writing.c
  4. 173
      src/core/lib/compression/stream_compression.c
  5. 34
      src/core/lib/compression/stream_compression.h
  6. 218
      src/core/lib/compression/stream_compression_gzip.c
  7. 26
      src/core/lib/compression/stream_compression_gzip.h
  8. 77
      src/core/lib/compression/stream_compression_identity.c
  9. 26
      src/core/lib/compression/stream_compression_identity.h

@ -197,6 +197,7 @@ filegroups:
- src/core/lib/compression/stream_compression.c - src/core/lib/compression/stream_compression.c
- src/core/lib/debug/stats.c - src/core/lib/debug/stats.c
- src/core/lib/debug/stats_data.c - src/core/lib/debug/stats_data.c
- src/core/lib/compression/stream_compression_gzip.c
- src/core/lib/http/format_request.c - src/core/lib/http/format_request.c
- src/core/lib/http/httpcli.c - src/core/lib/http/httpcli.c
- src/core/lib/http/parser.c - src/core/lib/http/parser.c
@ -349,6 +350,7 @@ filegroups:
- src/core/lib/compression/stream_compression.h - src/core/lib/compression/stream_compression.h
- src/core/lib/debug/stats.h - src/core/lib/debug/stats.h
- src/core/lib/debug/stats_data.h - src/core/lib/debug/stats_data.h
- src/core/lib/compression/stream_compression_gzip.h
- src/core/lib/http/format_request.h - src/core/lib/http/format_request.h
- src/core/lib/http/httpcli.h - src/core/lib/http/httpcli.h
- src/core/lib/http/parser.h - src/core/lib/http/parser.h

@ -1731,7 +1731,7 @@ void grpc_chttp2_maybe_complete_recv_message(grpc_exec_ctx *exec_ctx,
if (!s->stream_decompression_ctx) { if (!s->stream_decompression_ctx) {
s->stream_decompression_ctx = s->stream_decompression_ctx =
grpc_stream_compression_context_create( grpc_stream_compression_context_create(
GRPC_STREAM_COMPRESSION_DECOMPRESS); GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS);
} }
if (!grpc_stream_decompress(s->stream_decompression_ctx, if (!grpc_stream_decompress(s->stream_decompression_ctx,
&s->unprocessed_incoming_frames_buffer, &s->unprocessed_incoming_frames_buffer,
@ -1804,7 +1804,7 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_exec_ctx *exec_ctx,
bool end_of_context; bool end_of_context;
if (!s->stream_decompression_ctx) { if (!s->stream_decompression_ctx) {
s->stream_decompression_ctx = grpc_stream_compression_context_create( s->stream_decompression_ctx = grpc_stream_compression_context_create(
GRPC_STREAM_COMPRESSION_DECOMPRESS); GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS);
} }
if (!grpc_stream_decompress(s->stream_decompression_ctx, if (!grpc_stream_decompress(s->stream_decompression_ctx,
&s->frame_storage, &s->frame_storage,
@ -2694,7 +2694,7 @@ static grpc_error *incoming_byte_stream_pull(grpc_exec_ctx *exec_ctx,
bool end_of_context; bool end_of_context;
if (!s->stream_decompression_ctx) { if (!s->stream_decompression_ctx) {
s->stream_decompression_ctx = grpc_stream_compression_context_create( s->stream_decompression_ctx = grpc_stream_compression_context_create(
GRPC_STREAM_COMPRESSION_DECOMPRESS); GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS);
} }
if (!grpc_stream_decompress(s->stream_decompression_ctx, if (!grpc_stream_decompress(s->stream_decompression_ctx,
&s->unprocessed_incoming_frames_buffer, &s->unprocessed_incoming_frames_buffer,

@ -329,7 +329,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
if (s->stream_compression_ctx == NULL) { if (s->stream_compression_ctx == NULL) {
s->stream_compression_ctx = s->stream_compression_ctx =
grpc_stream_compression_context_create( grpc_stream_compression_context_create(
GRPC_STREAM_COMPRESSION_COMPRESS); GRPC_STREAM_COMPRESSION_GZIP_COMPRESS);
} }
s->uncompressed_data_size = s->flow_controlled_buffer.length; s->uncompressed_data_size = s->flow_controlled_buffer.length;
GPR_ASSERT(grpc_stream_compress( GPR_ASSERT(grpc_stream_compress(

@ -16,176 +16,57 @@
* *
*/ */
#include <grpc/support/alloc.h>
#include <grpc/support/log.h> #include <grpc/support/log.h>
#include "src/core/lib/compression/stream_compression.h" #include "src/core/lib/compression/stream_compression.h"
#include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/compression/stream_compression_gzip.h"
#include "src/core/lib/slice/slice_internal.h"
#define OUTPUT_BLOCK_SIZE (1024) extern const grpc_stream_compression_vtable grpc_stream_compression_identity_vtable;
static bool gzip_flate(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size, int flush,
bool *end_of_context) {
GPR_ASSERT(flush == 0 || flush == Z_SYNC_FLUSH || flush == Z_FINISH);
/* Full flush is not allowed when inflating. */
GPR_ASSERT(!(ctx->flate == inflate && (flush == Z_FINISH)));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
int r;
bool eoc = false;
size_t original_max_output_size = max_output_size;
while (max_output_size > 0 && (in->length > 0 || flush) && !eoc) {
size_t slice_size = max_output_size < OUTPUT_BLOCK_SIZE ? max_output_size
: OUTPUT_BLOCK_SIZE;
grpc_slice slice_out = GRPC_SLICE_MALLOC(slice_size);
ctx->zs.avail_out = (uInt)slice_size;
ctx->zs.next_out = GRPC_SLICE_START_PTR(slice_out);
while (ctx->zs.avail_out > 0 && in->length > 0 && !eoc) {
grpc_slice slice = grpc_slice_buffer_take_first(in);
ctx->zs.avail_in = (uInt)GRPC_SLICE_LENGTH(slice);
ctx->zs.next_in = GRPC_SLICE_START_PTR(slice);
r = ctx->flate(&ctx->zs, Z_NO_FLUSH);
if (r < 0 && r != Z_BUF_ERROR) {
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
} else if (r == Z_STREAM_END && ctx->flate == inflate) {
eoc = true;
}
if (ctx->zs.avail_in > 0) {
grpc_slice_buffer_undo_take_first(
in,
grpc_slice_sub(slice, GRPC_SLICE_LENGTH(slice) - ctx->zs.avail_in,
GRPC_SLICE_LENGTH(slice)));
}
grpc_slice_unref_internal(&exec_ctx, slice);
}
if (flush != 0 && ctx->zs.avail_out > 0 && !eoc) {
GPR_ASSERT(in->length == 0);
r = ctx->flate(&ctx->zs, flush);
if (flush == Z_SYNC_FLUSH) {
switch (r) {
case Z_OK:
/* Maybe flush is not complete; just made some partial progress. */
if (ctx->zs.avail_out > 0) {
flush = 0;
}
break;
case Z_BUF_ERROR:
case Z_STREAM_END:
flush = 0;
break;
default:
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
}
} else if (flush == Z_FINISH) {
switch (r) {
case Z_OK:
case Z_BUF_ERROR:
/* Wait for the next loop to assign additional output space. */
GPR_ASSERT(ctx->zs.avail_out == 0);
break;
case Z_STREAM_END:
flush = 0;
break;
default:
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
}
}
}
if (ctx->zs.avail_out == 0) {
grpc_slice_buffer_add(out, slice_out);
} else if (ctx->zs.avail_out < slice_size) {
slice_out.data.refcounted.length -= ctx->zs.avail_out;
grpc_slice_buffer_add(out, slice_out);
} else {
grpc_slice_unref_internal(&exec_ctx, slice_out);
}
max_output_size -= (slice_size - ctx->zs.avail_out);
}
grpc_exec_ctx_finish(&exec_ctx);
if (end_of_context) {
*end_of_context = eoc;
}
if (output_size) {
*output_size = original_max_output_size - max_output_size;
}
return true;
}
bool grpc_stream_compress(grpc_stream_compression_context *ctx, bool grpc_stream_compress(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out, grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size, size_t *output_size, size_t max_output_size,
grpc_stream_compression_flush flush) { grpc_stream_compression_flush flush) {
GPR_ASSERT(ctx->flate == deflate); return ctx->vtable->compress(ctx, in, out, output_size, max_output_size, flush);
int gzip_flush;
switch (flush) {
case GRPC_STREAM_COMPRESSION_FLUSH_NONE:
gzip_flush = 0;
break;
case GRPC_STREAM_COMPRESSION_FLUSH_SYNC:
gzip_flush = Z_SYNC_FLUSH;
break;
case GRPC_STREAM_COMPRESSION_FLUSH_FINISH:
gzip_flush = Z_FINISH;
break;
default:
gzip_flush = 0;
}
return gzip_flate(ctx, in, out, output_size, max_output_size, gzip_flush,
NULL);
} }
bool grpc_stream_decompress(grpc_stream_compression_context *ctx, bool grpc_stream_decompress(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out, grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size, size_t *output_size, size_t max_output_size,
bool *end_of_context) { bool *end_of_context) {
GPR_ASSERT(ctx->flate == inflate); return ctx->vtable->decompress(ctx, in, out, output_size, max_output_size, end_of_context);
return gzip_flate(ctx, in, out, output_size, max_output_size, Z_SYNC_FLUSH,
end_of_context);
} }
grpc_stream_compression_context *grpc_stream_compression_context_create( grpc_stream_compression_context *grpc_stream_compression_context_create(
grpc_stream_compression_method method) { grpc_stream_compression_method method) {
grpc_stream_compression_context *ctx = switch (method) {
gpr_zalloc(sizeof(grpc_stream_compression_context)); case GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS:
int r; case GRPC_STREAM_COMPRESSION_IDENTITY_DECOMPRESS:
if (ctx == NULL) { return grpc_stream_compression_identity_vtable.context_create(method);
return NULL; case GRPC_STREAM_COMPRESSION_GZIP_COMPRESS:
} case GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS:
if (method == GRPC_STREAM_COMPRESSION_DECOMPRESS) { return grpc_stream_compression_gzip_vtable.context_create(method);
r = inflateInit2(&ctx->zs, 0x1F); default:
ctx->flate = inflate; gpr_log(GPR_ERROR, "Unknown stream compression method: %d", method);
} else { return NULL;
r = deflateInit2(&ctx->zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 0x1F, 8,
Z_DEFAULT_STRATEGY);
ctx->flate = deflate;
}
if (r != Z_OK) {
gpr_free(ctx);
return NULL;
} }
return ctx;
} }
void grpc_stream_compression_context_destroy( void grpc_stream_compression_context_destroy(
grpc_stream_compression_context *ctx) { grpc_stream_compression_context *ctx) {
if (ctx->flate == inflate) { ctx->vtable->context_destroy(ctx);
inflateEnd(&ctx->zs); }
int grpc_stream_compression_method_parse(
grpc_slice value, bool is_compress, grpc_stream_compression_method *method) {
if (grpc_slice_eq(value, GRPC_MDSTR_IDENTITY)) {
*method = is_compress ? GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS : GRPC_STREAM_COMPRESSION_IDENTITY_DECOMPRESS;
return 1;
} else if (grpc_slice_eq(value, GRPC_MDSTR_GZIP)) {
*method = is_compress ? GRPC_STREAM_COMPRESSION_GZIP_COMPRESS : GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS;
return 1;
} else { } else {
deflateEnd(&ctx->zs); return 0;
} }
gpr_free(ctx);
} }

@ -24,15 +24,20 @@
#include <grpc/slice_buffer.h> #include <grpc/slice_buffer.h>
#include <zlib.h> #include <zlib.h>
#include "src/core/lib/transport/static_metadata.h"
typedef struct grpc_stream_compression_vtable grpc_stream_compression_vtable;
/* Stream compression/decompression context */ /* Stream compression/decompression context */
typedef struct grpc_stream_compression_context { typedef struct grpc_stream_compression_context {
z_stream zs; const grpc_stream_compression_vtable *vtable;
int (*flate)(z_stream *zs, int flush);
} grpc_stream_compression_context; } grpc_stream_compression_context;
typedef enum grpc_stream_compression_method { typedef enum grpc_stream_compression_method {
GRPC_STREAM_COMPRESSION_COMPRESS = 0, GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS = 0,
GRPC_STREAM_COMPRESSION_DECOMPRESS, GRPC_STREAM_COMPRESSION_IDENTITY_DECOMPRESS,
GRPC_STREAM_COMPRESSION_GZIP_COMPRESS,
GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS,
GRPC_STREAM_COMPRESSION_METHOD_COUNT GRPC_STREAM_COMPRESSION_METHOD_COUNT
} grpc_stream_compression_method; } grpc_stream_compression_method;
@ -43,6 +48,21 @@ typedef enum grpc_stream_compression_flush {
GRPC_STREAM_COMPRESSION_FLUSH_COUNT GRPC_STREAM_COMPRESSION_FLUSH_COUNT
} grpc_stream_compression_flush; } grpc_stream_compression_flush;
struct grpc_stream_compression_vtable {
bool (*compress)(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
grpc_stream_compression_flush flush);
bool (*decompress)(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
bool *end_of_context);
grpc_stream_compression_context *(*context_create)(
grpc_stream_compression_method method);
void (*context_destroy)(
grpc_stream_compression_context *ctx);
};
/** /**
* Compress bytes provided in \a in with a given context, with an optional flush * Compress bytes provided in \a in with a given context, with an optional flush
* at the end of compression. Emits at most \a max_output_size compressed bytes * at the end of compression. Emits at most \a max_output_size compressed bytes
@ -87,4 +107,10 @@ grpc_stream_compression_context *grpc_stream_compression_context_create(
void grpc_stream_compression_context_destroy( void grpc_stream_compression_context_destroy(
grpc_stream_compression_context *ctx); grpc_stream_compression_context *ctx);
/**
* Parse stream compression method based on algorithm name
*/
int grpc_stream_compression_method_parse(
grpc_slice value, bool is_compress, grpc_stream_compression_method *method);
#endif #endif

@ -0,0 +1,218 @@
/*
*
* Copyright 2017 gRPC authors.
*
* Licensed 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.
*
*/
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/lib/compression/stream_compression.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/slice/slice_internal.h"
#define OUTPUT_BLOCK_SIZE (1024)
typedef struct grpc_stream_compression_context_gzip {
grpc_stream_compression_context base;
z_stream zs;
int (*flate)(z_stream *zs, int flush);
} grpc_stream_compression_context_gzip;
static bool gzip_flate(grpc_stream_compression_context_gzip *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size, int flush,
bool *end_of_context) {
GPR_ASSERT(flush == 0 || flush == Z_SYNC_FLUSH || flush == Z_FINISH);
/* Full flush is not allowed when inflating. */
GPR_ASSERT(!(ctx->flate == inflate && (flush == Z_FINISH)));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
int r;
bool eoc = false;
size_t original_max_output_size = max_output_size;
while (max_output_size > 0 && (in->length > 0 || flush) && !eoc) {
size_t slice_size = max_output_size < OUTPUT_BLOCK_SIZE ? max_output_size
: OUTPUT_BLOCK_SIZE;
grpc_slice slice_out = GRPC_SLICE_MALLOC(slice_size);
ctx->zs.avail_out = (uInt)slice_size;
ctx->zs.next_out = GRPC_SLICE_START_PTR(slice_out);
while (ctx->zs.avail_out > 0 && in->length > 0 && !eoc) {
grpc_slice slice = grpc_slice_buffer_take_first(in);
ctx->zs.avail_in = (uInt)GRPC_SLICE_LENGTH(slice);
ctx->zs.next_in = GRPC_SLICE_START_PTR(slice);
r = ctx->flate(&ctx->zs, Z_NO_FLUSH);
if (r < 0 && r != Z_BUF_ERROR) {
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
} else if (r == Z_STREAM_END && ctx->flate == inflate) {
eoc = true;
}
if (ctx->zs.avail_in > 0) {
grpc_slice_buffer_undo_take_first(
in,
grpc_slice_sub(slice, GRPC_SLICE_LENGTH(slice) - ctx->zs.avail_in,
GRPC_SLICE_LENGTH(slice)));
}
grpc_slice_unref_internal(&exec_ctx, slice);
}
if (flush != 0 && ctx->zs.avail_out > 0 && !eoc) {
GPR_ASSERT(in->length == 0);
r = ctx->flate(&ctx->zs, flush);
if (flush == Z_SYNC_FLUSH) {
switch (r) {
case Z_OK:
/* Maybe flush is not complete; just made some partial progress. */
if (ctx->zs.avail_out > 0) {
flush = 0;
}
break;
case Z_BUF_ERROR:
case Z_STREAM_END:
flush = 0;
break;
default:
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
}
} else if (flush == Z_FINISH) {
switch (r) {
case Z_OK:
case Z_BUF_ERROR:
/* Wait for the next loop to assign additional output space. */
GPR_ASSERT(ctx->zs.avail_out == 0);
break;
case Z_STREAM_END:
flush = 0;
break;
default:
gpr_log(GPR_ERROR, "zlib error (%d)", r);
grpc_slice_unref_internal(&exec_ctx, slice_out);
grpc_exec_ctx_finish(&exec_ctx);
return false;
}
}
}
if (ctx->zs.avail_out == 0) {
grpc_slice_buffer_add(out, slice_out);
} else if (ctx->zs.avail_out < slice_size) {
slice_out.data.refcounted.length -= ctx->zs.avail_out;
grpc_slice_buffer_add(out, slice_out);
} else {
grpc_slice_unref_internal(&exec_ctx, slice_out);
}
max_output_size -= (slice_size - ctx->zs.avail_out);
}
grpc_exec_ctx_finish(&exec_ctx);
if (end_of_context) {
*end_of_context = eoc;
}
if (output_size) {
*output_size = original_max_output_size - max_output_size;
}
return true;
}
static bool grpc_stream_compress_gzip(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
grpc_stream_compression_flush flush) {
if (ctx == NULL) {
return false;
}
grpc_stream_compression_context_gzip *gzip_ctx = (grpc_stream_compression_context_gzip *)ctx;
GPR_ASSERT(gzip_ctx->flate == deflate);
int gzip_flush;
switch (flush) {
case GRPC_STREAM_COMPRESSION_FLUSH_NONE:
gzip_flush = 0;
break;
case GRPC_STREAM_COMPRESSION_FLUSH_SYNC:
gzip_flush = Z_SYNC_FLUSH;
break;
case GRPC_STREAM_COMPRESSION_FLUSH_FINISH:
gzip_flush = Z_FINISH;
break;
default:
gzip_flush = 0;
}
return gzip_flate(gzip_ctx, in, out, output_size, max_output_size, gzip_flush,
NULL);
}
static bool grpc_stream_decompress_gzip(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
bool *end_of_context) {
if (ctx == NULL) {
return false;
}
grpc_stream_compression_context_gzip *gzip_ctx = (grpc_stream_compression_context_gzip *)ctx;
GPR_ASSERT(gzip_ctx->flate == inflate);
return gzip_flate(gzip_ctx, in, out, output_size, max_output_size, Z_SYNC_FLUSH,
end_of_context);
}
static grpc_stream_compression_context *grpc_stream_compression_context_create_gzip(
grpc_stream_compression_method method) {
GPR_ASSERT(method == GRPC_STREAM_COMPRESSION_GZIP_COMPRESS || method == GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS);
grpc_stream_compression_context_gzip *gzip_ctx =
gpr_zalloc(sizeof(grpc_stream_compression_context_gzip));
int r;
if (gzip_ctx == NULL) {
return NULL;
}
if (method == GRPC_STREAM_COMPRESSION_GZIP_DECOMPRESS) {
r = inflateInit2(&gzip_ctx->zs, 0x1F);
gzip_ctx->flate = inflate;
} else {
r = deflateInit2(&gzip_ctx->zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, 0x1F, 8,
Z_DEFAULT_STRATEGY);
gzip_ctx->flate = deflate;
}
if (r != Z_OK) {
gpr_free(gzip_ctx);
return NULL;
}
return (grpc_stream_compression_context *)gzip_ctx;
}
static void grpc_stream_compression_context_destroy_gzip(
grpc_stream_compression_context *ctx) {
if (ctx == NULL) {
return;
}
grpc_stream_compression_context_gzip *gzip_ctx = (grpc_stream_compression_context_gzip*)ctx;
if (gzip_ctx->flate == inflate) {
inflateEnd(&gzip_ctx->zs);
} else {
deflateEnd(&gzip_ctx->zs);
}
gpr_free(ctx);
}
const grpc_stream_compression_vtable grpc_stream_compression_gzip_vtable = {
.compress = grpc_stream_compress_gzip,
.decompress = grpc_stream_decompress_gzip,
.context_create = grpc_stream_compression_context_create_gzip,
.context_destroy = grpc_stream_compression_context_destroy_gzip
};

@ -0,0 +1,26 @@
/*
*
* Copyright 2017 gRPC authors.
*
* Licensed 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.
*
*/
#ifndef GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_GZIP_H
#define GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_GZIP_H
#include "src/core/lib/compression/stream_compression.h"
extern const grpc_stream_compression_vtable grpc_stream_compression_gzip_vtable;
#endif

@ -0,0 +1,77 @@
/*
*
* Copyright 2017 gRPC authors.
*
* Licensed 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.
*
*/
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/lib/compression/stream_compression.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/slice/slice_internal.h"
#define OUTPUT_BLOCK_SIZE (1024)
static void grpc_stream_compression_pass_through(grpc_slice_buffer *in, grpc_slice_buffer *out, size_t *output_size, size_t max_output_size) {
if (max_output_size >= in->length) {
*output_size = in->length;
grpc_slice_buffer_move_into(in, out);
} else {
*output_size = max_output_size;
grpc_slice_buffer_move_first(in, max_output_size, out);
}
}
static bool grpc_stream_compress_identity(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
grpc_stream_compression_flush flush) {
if (ctx == NULL) {
return false;
}
grpc_stream_compression_pass_through(in, out, output_size, max_output_size);
return true;
}
static bool grpc_stream_decompress_identity(grpc_stream_compression_context *ctx,
grpc_slice_buffer *in, grpc_slice_buffer *out,
size_t *output_size, size_t max_output_size,
bool *end_of_context) {
if (ctx == NULL) {
return false;
}
grpc_stream_compression_pass_through(in, out, output_size, max_output_size);
return true;
}
static grpc_stream_compression_context *grpc_stream_compression_context_create_identity(
grpc_stream_compression_method method) {
GPR_ASSERT(method == GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS || method == GRPC_STREAM_COMPRESSION_IDENTITY_DECOMPRESS);
/* No context needed in this case. Use fake context instead. */
return (grpc_stream_compression_context *)1;
}
static void grpc_stream_compression_context_destroy_identity(
grpc_stream_compression_context *ctx) {
return;
}
const grpc_stream_compression_vtable grpc_stream_compression_identity_vtable = {
.compress = grpc_stream_compress_identity,
.decompress = grpc_stream_decompress_identity,
.context_create = grpc_stream_compression_context_create_identity,
.context_destroy = grpc_stream_compression_context_destroy_identity
};

@ -0,0 +1,26 @@
/*
*
* Copyright 2017 gRPC authors.
*
* Licensed 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.
*
*/
#ifndef GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_IDENTITY_H
#define GRPC_CORE_LIB_COMPRESSION_STREAM_COMPRESSION_IDENTITY_H
#include "src/core/lib/compression/stream_compression.h"
extern const grpc_stream_compression_vtable grpc_stream_compression_identity_vtable;
#endif
Loading…
Cancel
Save