Merge github.com:grpc/grpc into now-i-get-to-show-it-to-EVERYBODY

Conflicts:
	src/cpp/client/channel_arguments.cc
pull/2370/head
Craig Tiller 10 years ago
commit b4fdd32100
  1. 6
      BUILD
  2. 1613
      Makefile
  3. 4
      build.json
  4. 3
      gRPC.podspec
  5. 5
      include/grpc++/channel_arguments.h
  6. 10
      include/grpc++/client_context.h
  7. 14
      include/grpc++/server_context.h
  8. 31
      include/grpc/compression.h
  9. 29
      src/core/channel/channel_args.c
  10. 13
      src/core/channel/channel_args.h
  11. 325
      src/core/channel/compress_filter.c
  12. 65
      src/core/channel/compress_filter.h
  13. 16
      src/core/client_config/README.md
  14. 46
      src/core/compression/algorithm.c
  15. 62
      src/core/support/string.c
  16. 11
      src/core/support/string.h
  17. 96
      src/core/surface/call.c
  18. 13
      src/core/surface/channel.c
  19. 3
      src/core/surface/channel.h
  20. 2
      src/core/surface/channel_create.c
  21. 2
      src/core/surface/secure_channel_create.c
  22. 6
      src/core/surface/server.c
  23. 6
      src/core/surface/server.h
  24. 5
      src/core/surface/server_create.c
  25. 11
      src/core/transport/chttp2/frame_data.c
  26. 1
      src/core/transport/chttp2/frame_data.h
  27. 6
      src/core/transport/chttp2/stream_encoder.c
  28. 6
      src/core/transport/stream_op.c
  29. 5
      src/core/transport/stream_op.h
  30. 5
      src/cpp/client/channel_arguments.cc
  31. 15
      src/cpp/client/client_context.cc
  32. 4
      src/cpp/proto/proto_utils.cc
  33. 19
      src/cpp/server/server_context.cc
  34. 9
      src/ruby/ext/grpc/rb_call.c
  35. 3
      src/ruby/ext/grpc/rb_channel.c
  36. 3
      src/ruby/ext/grpc/rb_channel_args.c
  37. 2
      src/ruby/ext/grpc/rb_completion_queue.c
  38. 2
      src/ruby/ext/grpc/rb_credentials.c
  39. 9
      src/ruby/ext/grpc/rb_grpc.c
  40. 3
      src/ruby/ext/grpc/rb_server.c
  41. 2
      src/ruby/ext/grpc/rb_server_credentials.c
  42. 4
      test/core/compression/message_compress_test.c
  43. 13
      test/core/end2end/cq_verifier.c
  44. 135
      test/core/end2end/fixtures/chttp2_fullstack_compression.c
  45. 2
      test/core/end2end/fixtures/chttp2_socket_pair.c
  46. 2
      test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
  47. 2
      test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c
  48. 2
      test/core/end2end/gen_build_json.py
  49. 315
      test/core/end2end/tests/request_with_compressed_payload.c
  50. 48
      test/core/security/fetch_oauth2.c
  51. 93
      test/core/security/oauth2_utils.c
  52. 51
      test/core/security/oauth2_utils.h
  53. 116
      test/core/support/string_test.c
  54. 3
      test/cpp/end2end/end2end_test.cc
  55. 1
      test/cpp/end2end/generic_end2end_test.cc
  56. 9
      test/cpp/interop/client.cc
  57. 39
      test/cpp/interop/client_helper.cc
  58. 2
      test/cpp/interop/client_helper.h
  59. 23
      test/cpp/interop/interop_client.cc
  60. 3
      test/cpp/interop/interop_client.h
  61. 2
      tools/doxygen/Doxyfile.core.internal
  62. 72
      tools/jenkins/run_distribution.sh
  63. 2307
      tools/run_tests/sources_and_headers.json
  64. 1292
      tools/run_tests/tests.json
  65. 516
      vsprojects/Grpc.mak
  66. 3
      vsprojects/grpc/grpc.vcxproj
  67. 6
      vsprojects/grpc/grpc.vcxproj.filters
  68. 3
      vsprojects/grpc_test_util/grpc_test_util.vcxproj
  69. 3
      vsprojects/grpc_unsecure/grpc_unsecure.vcxproj
  70. 6
      vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters

@ -154,6 +154,7 @@ cc_library(
"src/core/channel/channel_args.h",
"src/core/channel/channel_stack.h",
"src/core/channel/client_channel.h",
"src/core/channel/compress_filter.h",
"src/core/channel/connected_channel.h",
"src/core/channel/context.h",
"src/core/channel/http_client_filter.h",
@ -273,6 +274,7 @@ cc_library(
"src/core/channel/channel_args.c",
"src/core/channel/channel_stack.c",
"src/core/channel/client_channel.c",
"src/core/channel/compress_filter.c",
"src/core/channel/connected_channel.c",
"src/core/channel/http_client_filter.c",
"src/core/channel/http_server_filter.c",
@ -408,6 +410,7 @@ cc_library(
"src/core/channel/channel_args.h",
"src/core/channel/channel_stack.h",
"src/core/channel/client_channel.h",
"src/core/channel/compress_filter.h",
"src/core/channel/connected_channel.h",
"src/core/channel/context.h",
"src/core/channel/http_client_filter.h",
@ -504,6 +507,7 @@ cc_library(
"src/core/channel/channel_args.c",
"src/core/channel/channel_stack.c",
"src/core/channel/client_channel.c",
"src/core/channel/compress_filter.c",
"src/core/channel/connected_channel.c",
"src/core/channel/http_client_filter.c",
"src/core/channel/http_server_filter.c",
@ -981,6 +985,7 @@ objc_library(
"src/core/channel/channel_args.c",
"src/core/channel/channel_stack.c",
"src/core/channel/client_channel.c",
"src/core/channel/compress_filter.c",
"src/core/channel/connected_channel.c",
"src/core/channel/http_client_filter.c",
"src/core/channel/http_server_filter.c",
@ -1118,6 +1123,7 @@ objc_library(
"src/core/channel/channel_args.h",
"src/core/channel/channel_stack.h",
"src/core/channel/client_channel.h",
"src/core/channel/compress_filter.h",
"src/core/channel/connected_channel.h",
"src/core/channel/context.h",
"src/core/channel/http_client_filter.h",

1613
Makefile

File diff suppressed because one or more lines are too long

@ -115,6 +115,7 @@
"src/core/channel/channel_args.h",
"src/core/channel/channel_stack.h",
"src/core/channel/client_channel.h",
"src/core/channel/compress_filter.h",
"src/core/channel/connected_channel.h",
"src/core/channel/context.h",
"src/core/channel/http_client_filter.h",
@ -211,6 +212,7 @@
"src/core/channel/channel_args.c",
"src/core/channel/channel_stack.c",
"src/core/channel/client_channel.c",
"src/core/channel/compress_filter.c",
"src/core/channel/connected_channel.c",
"src/core/channel/http_client_filter.c",
"src/core/channel/http_server_filter.c",
@ -322,6 +324,7 @@
"headers": [
"test/core/end2end/cq_verifier.h",
"test/core/iomgr/endpoint_tests.h",
"test/core/security/oauth2_utils.h",
"test/core/util/grpc_profiler.h",
"test/core/util/parse_hexstring.h",
"test/core/util/port.h",
@ -330,6 +333,7 @@
"src": [
"test/core/end2end/cq_verifier.c",
"test/core/iomgr/endpoint_tests.c",
"test/core/security/oauth2_utils.c",
"test/core/util/grpc_profiler.c",
"test/core/util/parse_hexstring.c",
"test/core/util/port_posix.c",

@ -156,6 +156,7 @@ Pod::Spec.new do |s|
'src/core/channel/channel_args.h',
'src/core/channel/channel_stack.h',
'src/core/channel/client_channel.h',
'src/core/channel/compress_filter.h',
'src/core/channel/connected_channel.h',
'src/core/channel/context.h',
'src/core/channel/http_client_filter.h',
@ -282,6 +283,7 @@ Pod::Spec.new do |s|
'src/core/channel/channel_args.c',
'src/core/channel/channel_stack.c',
'src/core/channel/client_channel.c',
'src/core/channel/compress_filter.c',
'src/core/channel/connected_channel.c',
'src/core/channel/http_client_filter.c',
'src/core/channel/http_server_filter.c',
@ -418,6 +420,7 @@ Pod::Spec.new do |s|
'src/core/channel/channel_args.h',
'src/core/channel/channel_stack.h',
'src/core/channel/client_channel.h',
'src/core/channel/compress_filter.h',
'src/core/channel/connected_channel.h',
'src/core/channel/context.h',
'src/core/channel/http_client_filter.h',

@ -67,8 +67,9 @@ class ChannelArguments {
void SetSslTargetNameOverride(const grpc::string& name);
// TODO(yangg) add flow control options
// Set the compression level for the channel.
void SetCompressionLevel(grpc_compression_level level);
// Set the compression algorithm for the channel.
void _Experimental_SetCompressionAlgorithm(
grpc_compression_algorithm algorithm);
// Generic channel argument setters. Only for advanced use cases.
void SetInt(const grpc::string& key, int value);

@ -38,6 +38,7 @@
#include <memory>
#include <string>
#include <grpc/compression.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc++/auth_context.h>
@ -109,6 +110,13 @@ class ClientContext {
creds_ = creds;
}
grpc_compression_algorithm _experimental_get_compression_algorithm() const {
return compression_algorithm_;
}
void _experimental_set_compression_algorithm(
grpc_compression_algorithm algorithm);
std::shared_ptr<const AuthContext> auth_context() const;
// Get and set census context
@ -167,6 +175,8 @@ class ClientContext {
std::multimap<grpc::string, grpc::string> send_initial_metadata_;
std::multimap<grpc::string, grpc::string> recv_initial_metadata_;
std::multimap<grpc::string, grpc::string> trailing_metadata_;
grpc_compression_algorithm compression_algorithm_;
};
} // namespace grpc

@ -37,6 +37,7 @@
#include <map>
#include <memory>
#include <grpc/compression.h>
#include <grpc/support/time.h>
#include <grpc++/auth_context.h>
#include <grpc++/config.h>
@ -103,6 +104,16 @@ class ServerContext {
return client_metadata_;
}
grpc_compression_level get_compression_level() const {
return compression_level_;
}
void set_compression_level(grpc_compression_level level);
grpc_compression_algorithm get_compression_algorithm() const {
return compression_algorithm_;
}
void set_compression_algorithm(grpc_compression_algorithm algorithm);
std::shared_ptr<const AuthContext> auth_context() const;
private:
@ -154,6 +165,9 @@ class ServerContext {
std::multimap<grpc::string, grpc::string> client_metadata_;
std::multimap<grpc::string, grpc::string> initial_metadata_;
std::multimap<grpc::string, grpc::string> trailing_metadata_;
grpc_compression_level compression_level_;
grpc_compression_algorithm compression_algorithm_;
};
} // namespace grpc

@ -34,8 +34,12 @@
#ifndef GRPC_COMPRESSION_H
#define GRPC_COMPRESSION_H
#ifdef __cplusplus
extern "C" {
#endif
/** To be used in channel arguments */
#define GRPC_COMPRESSION_LEVEL_ARG "grpc.compression_level"
#define GRPC_COMPRESSION_ALGORITHM_ARG "grpc.compression_algorithm"
/* The various compression algorithms supported by GRPC */
typedef enum {
@ -50,13 +54,34 @@ typedef enum {
GRPC_COMPRESS_LEVEL_NONE = 0,
GRPC_COMPRESS_LEVEL_LOW,
GRPC_COMPRESS_LEVEL_MED,
GRPC_COMPRESS_LEVEL_HIGH
GRPC_COMPRESS_LEVEL_HIGH,
GRPC_COMPRESS_LEVEL_COUNT
} grpc_compression_level;
const char *grpc_compression_algorithm_name(
/** Parses \a name as a grpc_compression_algorithm instance, updating \a
* algorithm. Returns 1 upon success, 0 otherwise. */
int grpc_compression_algorithm_parse(const char *name,
grpc_compression_algorithm *algorithm);
/** Updates \a name with the encoding name corresponding to a valid \a
* algorithm. Returns 1 upon success, 0 otherwise. */
int grpc_compression_algorithm_name(grpc_compression_algorithm algorithm,
char **name);
/** Returns the compression level corresponding to \a algorithm.
*
* It abort()s for unknown algorithms. */
grpc_compression_level grpc_compression_level_for_algorithm(
grpc_compression_algorithm algorithm);
/** Returns the compression algorithm corresponding to \a level.
*
* It abort()s for unknown levels . */
grpc_compression_algorithm grpc_compression_algorithm_for_level(
grpc_compression_level level);
#ifdef __cplusplus
}
#endif
#endif /* GRPC_COMPRESSION_H */

@ -114,7 +114,7 @@ void grpc_channel_args_destroy(grpc_channel_args *a) {
}
int grpc_channel_args_is_census_enabled(const grpc_channel_args *a) {
unsigned i;
size_t i;
if (a == NULL) return 0;
for (i = 0; i < a->num_args; i++) {
if (0 == strcmp(a->args[i].key, GRPC_ARG_ENABLE_CENSUS)) {
@ -124,26 +124,25 @@ int grpc_channel_args_is_census_enabled(const grpc_channel_args *a) {
return 0;
}
grpc_compression_level grpc_channel_args_get_compression_level(
grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
const grpc_channel_args *a) {
size_t i;
if (a) {
for (i = 0; a && i < a->num_args; ++i) {
if (a->args[i].type == GRPC_ARG_INTEGER &&
!strcmp(GRPC_COMPRESSION_LEVEL_ARG, a->args[i].key)) {
return a->args[i].value.integer;
break;
}
if (a == NULL) return 0;
for (i = 0; i < a->num_args; ++i) {
if (a->args[i].type == GRPC_ARG_INTEGER &&
!strcmp(GRPC_COMPRESSION_ALGORITHM_ARG, a->args[i].key)) {
return a->args[i].value.integer;
break;
}
}
return GRPC_COMPRESS_LEVEL_NONE;
return GRPC_COMPRESS_NONE;
}
void grpc_channel_args_set_compression_level(grpc_channel_args **a,
grpc_compression_level level) {
grpc_channel_args *grpc_channel_args_set_compression_algorithm(
grpc_channel_args *a, grpc_compression_algorithm algorithm) {
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
tmp.key = GRPC_COMPRESSION_LEVEL_ARG;
tmp.value.integer = level;
*a = grpc_channel_args_copy_and_add(*a, &tmp, 1);
tmp.key = GRPC_COMPRESSION_ALGORITHM_ARG;
tmp.value.integer = algorithm;
return grpc_channel_args_copy_and_add(a, &tmp, 1);
}

@ -57,13 +57,14 @@ void grpc_channel_args_destroy(grpc_channel_args *a);
* is specified in channel args, otherwise returns 0. */
int grpc_channel_args_is_census_enabled(const grpc_channel_args *a);
/** Returns the compression level set in \a a. */
grpc_compression_level grpc_channel_args_get_compression_level(
/** Returns the compression algorithm set in \a a. */
grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
const grpc_channel_args *a);
/** Sets the compression level in \a a to \a level. Setting it to
* GRPC_COMPRESS_LEVEL_NONE disables compression for the channel. */
void grpc_channel_args_set_compression_level(grpc_channel_args **a,
grpc_compression_level level);
/** Returns a channel arg instance with compression enabled. If \a a is
* non-NULL, its args are copied. N.B. GRPC_COMPRESS_NONE disables compression
* for the channel. */
grpc_channel_args *grpc_channel_args_set_compression_algorithm(
grpc_channel_args *a, grpc_compression_algorithm algorithm);
#endif /* GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H */

@ -0,0 +1,325 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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 <assert.h>
#include <string.h>
#include <grpc/compression.h>
#include <grpc/support/log.h>
#include <grpc/support/slice_buffer.h>
#include "src/core/channel/compress_filter.h"
#include "src/core/channel/channel_args.h"
#include "src/core/compression/message_compress.h"
typedef struct call_data {
gpr_slice_buffer slices; /**< Buffers up input slices to be compressed */
grpc_linked_mdelem compression_algorithm_storage;
int remaining_slice_bytes; /**< Input data to be read, as per BEGIN_MESSAGE */
int written_initial_metadata; /**< Already processed initial md? */
/** Compression algorithm we'll try to use. It may be given by incoming
* metadata, or by the channel's default compression settings. */
grpc_compression_algorithm compression_algorithm;
/** If true, contents of \a compression_algorithm are authoritative */
int has_compression_algorithm;
} call_data;
typedef struct channel_data {
/** Metadata key for the incoming (requested) compression algorithm */
grpc_mdstr *mdstr_request_compression_algorithm_key;
/** Metadata key for the outgoing (used) compression algorithm */
grpc_mdstr *mdstr_outgoing_compression_algorithm_key;
/** Precomputed metadata elements for all available compression algorithms */
grpc_mdelem *mdelem_compression_algorithms[GRPC_COMPRESS_ALGORITHMS_COUNT];
/** The default, channel-level, compression algorithm */
grpc_compression_algorithm default_compression_algorithm;
} channel_data;
/** Compress \a slices in place using \a algorithm. Returns 1 if compression did
* actually happen, 0 otherwise (for example if the compressed output size was
* larger than the raw input).
*
* Returns 1 if the data was actually compress and 0 otherwise. */
static int compress_send_sb(grpc_compression_algorithm algorithm,
gpr_slice_buffer *slices) {
int did_compress;
gpr_slice_buffer tmp;
gpr_slice_buffer_init(&tmp);
did_compress = grpc_msg_compress(algorithm, slices, &tmp);
if (did_compress) {
gpr_slice_buffer_swap(slices, &tmp);
}
gpr_slice_buffer_destroy(&tmp);
return did_compress;
}
/** For each \a md element from the incoming metadata, filter out the entry for
* "grpc-encoding", using its value to populate the call data's
* compression_algorithm field. */
static grpc_mdelem* compression_md_filter(void *user_data, grpc_mdelem *md) {
grpc_call_element *elem = user_data;
call_data *calld = elem->call_data;
channel_data *channeld = elem->channel_data;
if (md->key == channeld->mdstr_request_compression_algorithm_key) {
const char *md_c_str = grpc_mdstr_as_c_string(md->value);
if (!grpc_compression_algorithm_parse(md_c_str,
&calld->compression_algorithm)) {
gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'. Ignoring.",
md_c_str);
calld->compression_algorithm = GRPC_COMPRESS_NONE;
}
calld->has_compression_algorithm = 1;
return NULL;
}
return md;
}
static int skip_compression(channel_data *channeld, call_data *calld) {
if (calld->has_compression_algorithm) {
if (calld->compression_algorithm == GRPC_COMPRESS_NONE) {
return 1;
}
return 0; /* we have an actual call-specific algorithm */
}
/* no per-call compression override */
return channeld->default_compression_algorithm == GRPC_COMPRESS_NONE;
}
/** Assembles a new grpc_stream_op_buffer with the compressed slices, modifying
* the associated GRPC_OP_BEGIN_MESSAGE accordingly (new compressed length,
* flags indicating compression is in effect) and replaces \a send_ops with it.
* */
static void finish_compressed_sopb(grpc_stream_op_buffer *send_ops,
grpc_call_element *elem) {
size_t i;
call_data *calld = elem->call_data;
int new_slices_added = 0; /* GPR_FALSE */
grpc_metadata_batch metadata;
grpc_stream_op_buffer new_send_ops;
grpc_sopb_init(&new_send_ops);
for (i = 0; i < send_ops->nops; i++) {
grpc_stream_op *sop = &send_ops->ops[i];
switch (sop->type) {
case GRPC_OP_BEGIN_MESSAGE:
grpc_sopb_add_begin_message(
&new_send_ops, calld->slices.length,
sop->data.begin_message.flags | GRPC_WRITE_INTERNAL_COMPRESS);
break;
case GRPC_OP_SLICE:
/* Once we reach the slices section of the original buffer, simply add
* all the new (compressed) slices. We obviously want to do this only
* once, hence the "new_slices_added" guard. */
if (!new_slices_added) {
size_t j;
for (j = 0; j < calld->slices.count; ++j) {
grpc_sopb_add_slice(&new_send_ops,
gpr_slice_ref(calld->slices.slices[j]));
}
new_slices_added = 1; /* GPR_TRUE */
}
break;
case GRPC_OP_METADATA:
/* move the metadata to the new buffer. */
grpc_metadata_batch_move(&metadata, &sop->data.metadata);
grpc_sopb_add_metadata(&new_send_ops, metadata);
break;
case GRPC_NO_OP:
break;
}
}
grpc_sopb_swap(send_ops, &new_send_ops);
grpc_sopb_destroy(&new_send_ops);
}
/** Filter's "main" function, called for any incoming grpc_transport_stream_op
* instance that holds a non-zero number of send operations, accesible to this
* function in \a send_ops. */
static void process_send_ops(grpc_call_element *elem,
grpc_stream_op_buffer *send_ops) {
call_data *calld = elem->call_data;
channel_data *channeld = elem->channel_data;
size_t i;
int did_compress = 0;
for (i = 0; i < send_ops->nops; ++i) {
grpc_stream_op *sop = &send_ops->ops[i];
switch (sop->type) {
case GRPC_OP_BEGIN_MESSAGE:
/* buffer up slices until we've processed all the expected ones (as
* given by GRPC_OP_BEGIN_MESSAGE) */
calld->remaining_slice_bytes = sop->data.begin_message.length;
if (sop->data.begin_message.flags & GRPC_WRITE_NO_COMPRESS) {
calld->has_compression_algorithm = 1; /* GPR_TRUE */
calld->compression_algorithm = GRPC_COMPRESS_NONE;
}
break;
case GRPC_OP_METADATA:
if (!calld->written_initial_metadata) {
/* Parse incoming request for compression. If any, it'll be available
* at calld->compression_algorithm */
grpc_metadata_batch_filter(&(sop->data.metadata),
compression_md_filter, elem);
if (!calld->has_compression_algorithm) {
/* If no algorithm was found in the metadata and we aren't
* exceptionally skipping compression, fall back to the channel
* default */
calld->compression_algorithm =
channeld->default_compression_algorithm;
calld->has_compression_algorithm = 1; /* GPR_TRUE */
}
grpc_metadata_batch_add_head(
&(sop->data.metadata), &calld->compression_algorithm_storage,
grpc_mdelem_ref(channeld->mdelem_compression_algorithms
[calld->compression_algorithm]));
calld->written_initial_metadata = 1; /* GPR_TRUE */
}
break;
case GRPC_OP_SLICE:
if (skip_compression(channeld, calld)) continue;
GPR_ASSERT(calld->remaining_slice_bytes > 0);
/* Increase input ref count, gpr_slice_buffer_add takes ownership. */
gpr_slice_buffer_add(&calld->slices, gpr_slice_ref(sop->data.slice));
calld->remaining_slice_bytes -= GPR_SLICE_LENGTH(sop->data.slice);
if (calld->remaining_slice_bytes == 0) {
did_compress =
compress_send_sb(calld->compression_algorithm, &calld->slices);
}
break;
case GRPC_NO_OP:
break;
}
}
/* Modify the send_ops stream_op_buffer depending on whether compression was
* carried out */
if (did_compress) {
finish_compressed_sopb(send_ops, elem);
}
}
/* Called either:
- in response to an API call (or similar) from above, to send something
- a network event (or similar) from below, to receive something
op contains type and call direction information, in addition to the data
that is being sent or received. */
static void compress_start_transport_stream_op(grpc_call_element *elem,
grpc_transport_stream_op *op) {
if (op->send_ops && op->send_ops->nops > 0) {
process_send_ops(elem, op->send_ops);
}
/* pass control down the stack */
grpc_call_next_op(elem, op);
}
/* Constructor for call_data */
static void init_call_elem(grpc_call_element *elem,
const void *server_transport_data,
grpc_transport_stream_op *initial_op) {
/* grab pointers to our data from the call element */
call_data *calld = elem->call_data;
/* initialize members */
gpr_slice_buffer_init(&calld->slices);
calld->has_compression_algorithm = 0;
calld->written_initial_metadata = 0; /* GPR_FALSE */
if (initial_op) {
if (initial_op->send_ops && initial_op->send_ops->nops > 0) {
process_send_ops(elem, initial_op->send_ops);
}
}
}
/* Destructor for call_data */
static void destroy_call_elem(grpc_call_element *elem) {
/* grab pointers to our data from the call element */
call_data *calld = elem->call_data;
gpr_slice_buffer_destroy(&calld->slices);
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
const grpc_channel_args *args, grpc_mdctx *mdctx,
int is_first, int is_last) {
channel_data *channeld = elem->channel_data;
grpc_compression_algorithm algo_idx;
channeld->default_compression_algorithm =
grpc_channel_args_get_compression_algorithm(args);
channeld->mdstr_request_compression_algorithm_key =
grpc_mdstr_from_string(mdctx, GRPC_COMPRESS_REQUEST_ALGORITHM_KEY);
channeld->mdstr_outgoing_compression_algorithm_key =
grpc_mdstr_from_string(mdctx, "grpc-encoding");
for (algo_idx = 0; algo_idx < GRPC_COMPRESS_ALGORITHMS_COUNT; ++algo_idx) {
char *algorith_name;
GPR_ASSERT(grpc_compression_algorithm_name(algo_idx, &algorith_name) != 0);
channeld->mdelem_compression_algorithms[algo_idx] =
grpc_mdelem_from_metadata_strings(
mdctx,
grpc_mdstr_ref(channeld->mdstr_outgoing_compression_algorithm_key),
grpc_mdstr_from_string(mdctx, algorith_name));
}
GPR_ASSERT(!is_last);
}
/* Destructor for channel data */
static void destroy_channel_elem(grpc_channel_element *elem) {
channel_data *channeld = elem->channel_data;
grpc_compression_algorithm algo_idx;
grpc_mdstr_unref(channeld->mdstr_request_compression_algorithm_key);
grpc_mdstr_unref(channeld->mdstr_outgoing_compression_algorithm_key);
for (algo_idx = 0; algo_idx < GRPC_COMPRESS_ALGORITHMS_COUNT;
++algo_idx) {
grpc_mdelem_unref(channeld->mdelem_compression_algorithms[algo_idx]);
}
}
const grpc_channel_filter grpc_compress_filter = {
compress_start_transport_stream_op,
grpc_channel_next_op,
sizeof(call_data),
init_call_elem,
destroy_call_elem,
sizeof(channel_data),
init_channel_elem,
destroy_channel_elem,
"compress"};

@ -0,0 +1,65 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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.
*
*/
#ifndef GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H
#define GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H
#include "src/core/channel/channel_stack.h"
#define GRPC_COMPRESS_REQUEST_ALGORITHM_KEY "internal:grpc-encoding-request"
/** Compression filter for outgoing data.
*
* See <grpc/compression.h> for the available compression settings.
*
* Compression settings may come from:
* - Channel configuration, as established at channel creation time.
* - The metadata accompanying the outgoing data to be compressed. This is
* taken as a request only. We may choose not to honor it. The metadata key
* is given by \a GRPC_COMPRESS_REQUEST_ALGORITHM_KEY.
*
* Compression can be disabled for concrete messages (for instance in order to
* prevent CRIME/BEAST type attacks) by having the GRPC_WRITE_NO_COMPRESS set in
* the BEGIN_MESSAGE flags.
*
* The attempted compression mechanism is added to the resulting initial
* metadata under the'grpc-encoding' key.
*
* If compression is actually performed, BEGIN_MESSAGE's flag is modified to
* incorporate GRPC_WRITE_INTERNAL_COMPRESS. Otherwise, and regardless of the
* aforementioned 'grpc-encoding' metadata value, data will pass through
* uncompressed. */
extern const grpc_channel_filter grpc_compress_filter;
#endif /* GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H */

@ -24,7 +24,7 @@ Load Balancing
Load balancing configuration is provided by a grpc_lb_policy object, stored as
part of grpc_client_config.
A load balancing policies primary job is to pick a target server given only the
The primary job of the load balancing policies is to pick a target server given only the
initial metadata for a request. It does this by providing a grpc_subchannel
object to the owning channel.
@ -39,15 +39,16 @@ decisions (for example, by avoiding disconnected backends).
Configured sub-channels are fully setup to participate in the grpc data plane.
Their behavior is specified by a set of grpc channel filters defined at their
construction. To customize this behavior, resolvers build grpc_subchannel_factory
objects, which use the decorator pattern to customize construction arguments for
concrete grpc_subchannel instances.
construction. To customize this behavior, resolvers build
grpc_subchannel_factory objects, which use the decorator pattern to customize
construction arguments for concrete grpc_subchannel instances.
Naming for GRPC
===============
Names in GRPC are represented by a URI.
Names in GRPC are represented by a URI (as defined in
[RFC 3986](https://tools.ietf.org/html/rfc3986)).
The following schemes are currently supported:
@ -56,5 +57,6 @@ dns:///host:port - dns schemes are currently supported so long as authority is
release)
unix:path - the unix scheme is used to create and connect to unix domain
sockets - the authority must be empty, and the path represents
the absolute or relative path to the desired socket
sockets - the authority must be empty, and the path
represents the absolute or relative path to the desired
socket

@ -32,21 +32,39 @@
*/
#include <stdlib.h>
#include <string.h>
#include <grpc/compression.h>
const char *grpc_compression_algorithm_name(
grpc_compression_algorithm algorithm) {
int grpc_compression_algorithm_parse(const char* name,
grpc_compression_algorithm *algorithm) {
if (strcmp(name, "none") == 0) {
*algorithm = GRPC_COMPRESS_NONE;
} else if (strcmp(name, "gzip") == 0) {
*algorithm = GRPC_COMPRESS_GZIP;
} else if (strcmp(name, "deflate") == 0) {
*algorithm = GRPC_COMPRESS_DEFLATE;
} else {
return 0;
}
return 1;
}
int grpc_compression_algorithm_name(grpc_compression_algorithm algorithm,
char **name) {
switch (algorithm) {
case GRPC_COMPRESS_NONE:
return "none";
*name = "none";
break;
case GRPC_COMPRESS_DEFLATE:
return "deflate";
*name = "deflate";
break;
case GRPC_COMPRESS_GZIP:
return "gzip";
case GRPC_COMPRESS_ALGORITHMS_COUNT:
return "error";
*name = "gzip";
break;
default:
return 0;
}
return "error";
return 1;
}
/* TODO(dgq): Add the ability to specify parameters to the individual
@ -65,3 +83,15 @@ grpc_compression_algorithm grpc_compression_algorithm_for_level(
abort();
}
}
grpc_compression_level grpc_compression_level_for_algorithm(
grpc_compression_algorithm algorithm) {
grpc_compression_level clevel;
for (clevel = GRPC_COMPRESS_LEVEL_NONE; clevel < GRPC_COMPRESS_LEVEL_COUNT;
++clevel) {
if (grpc_compression_algorithm_for_level(clevel) == algorithm) {
return clevel;
}
}
abort();
}

@ -38,6 +38,7 @@
#include <string.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include <grpc/support/useful.h>
@ -174,6 +175,12 @@ int gpr_ltoa(long value, char *string) {
}
char *gpr_strjoin(const char **strs, size_t nstrs, size_t *final_length) {
return gpr_strjoin_sep(strs, nstrs, "", final_length);
}
char *gpr_strjoin_sep(const char **strs, size_t nstrs, const char *sep,
size_t *final_length) {
const size_t sep_len = strlen(sep);
size_t out_length = 0;
size_t i;
char *out;
@ -181,10 +188,17 @@ char *gpr_strjoin(const char **strs, size_t nstrs, size_t *final_length) {
out_length += strlen(strs[i]);
}
out_length += 1; /* null terminator */
if (nstrs > 0) {
out_length += sep_len * (nstrs - 1); /* separators */
}
out = gpr_malloc(out_length);
out_length = 0;
for (i = 0; i < nstrs; i++) {
size_t slen = strlen(strs[i]);
const size_t slen = strlen(strs[i]);
if (i != 0) {
memcpy(out + out_length, sep, sep_len);
out_length += sep_len;
}
memcpy(out + out_length, strs[i], slen);
out_length += slen;
}
@ -195,6 +209,52 @@ char *gpr_strjoin(const char **strs, size_t nstrs, size_t *final_length) {
return out;
}
/** Finds the initial (\a begin) and final (\a end) offsets of the next
* substring from \a str + \a read_offset until the next \a sep or the end of \a
* str.
*
* Returns 1 and updates \a begin and \a end. Returns 0 otherwise. */
static int slice_find_separator_offset(const gpr_slice str,
const char *sep,
const size_t read_offset,
size_t *begin,
size_t *end) {
size_t i;
const gpr_uint8 *str_ptr = GPR_SLICE_START_PTR(str) + read_offset;
const size_t str_len = GPR_SLICE_LENGTH(str) - read_offset;
const size_t sep_len = strlen(sep);
if (str_len < sep_len) {
return 0;
}
for (i = 0; i <= str_len - sep_len; i++) {
if (memcmp(str_ptr + i, sep, sep_len) == 0) {
*begin = read_offset;
*end = read_offset + i;
return 1;
}
}
return 0;
}
void gpr_slice_split(gpr_slice str, const char *sep, gpr_slice_buffer *dst) {
const size_t sep_len = strlen(sep);
size_t begin, end;
GPR_ASSERT(sep_len > 0);
if (slice_find_separator_offset(str, sep, 0, &begin, &end) != 0) {
do {
gpr_slice_buffer_add_indexed(dst, gpr_slice_sub(str, begin, end));
} while (slice_find_separator_offset(str, sep, end + sep_len, &begin,
&end) != 0);
gpr_slice_buffer_add_indexed(
dst, gpr_slice_sub(str, end + sep_len, GPR_SLICE_LENGTH(str)));
} else { /* no sep found, add whole input */
gpr_slice_buffer_add_indexed(dst, gpr_slice_ref(str));
}
}
void gpr_strvec_init(gpr_strvec *sv) {
memset(sv, 0, sizeof(*sv));
}

@ -37,6 +37,7 @@
#include <stddef.h>
#include <grpc/support/port_platform.h>
#include <grpc/support/slice_buffer.h>
#include <grpc/support/slice.h>
#ifdef __cplusplus
@ -77,6 +78,16 @@ void gpr_reverse_bytes(char *str, int len);
if it is non-null. */
char *gpr_strjoin(const char **strs, size_t nstrs, size_t *total_length);
/* Join a set of strings using a separator, returning the resulting string.
Total combined length (excluding null terminator) is returned in total_length
if it is non-null. */
char *gpr_strjoin_sep(const char **strs, size_t nstrs, const char *sep,
size_t *total_length);
/** Split \a str by the separator \a sep. Results are stored in \a dst, which
* should be a properly initialized instance. */
void gpr_slice_split(gpr_slice str, const char *sep, gpr_slice_buffer *dst);
/* A vector of strings... for building up a final string one piece at a time */
typedef struct {
char **strs;

@ -30,24 +30,25 @@
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <grpc/compression.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include "src/core/census/grpc_context.h"
#include "src/core/surface/call.h"
#include "src/core/channel/channel_stack.h"
#include "src/core/iomgr/alarm.h"
#include "src/core/profiling/timers.h"
#include "src/core/support/string.h"
#include "src/core/surface/byte_buffer_queue.h"
#include "src/core/surface/call.h"
#include "src/core/surface/channel.h"
#include "src/core/surface/completion_queue.h"
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
/** The maximum number of completions possible.
Based upon the maximum number of individually queueable ops in the batch
@ -235,8 +236,8 @@ struct grpc_call {
/* Received call statuses from various sources */
received_status status[STATUS_SOURCE_COUNT];
/* Compression level for the call */
grpc_compression_level compression_level;
/* Compression algorithm for the call */
grpc_compression_algorithm compression_algorithm;
/* Contexts for various subsystems (security, tracing, ...). */
grpc_call_context_element context[GRPC_CONTEXT_COUNT];
@ -469,9 +470,14 @@ static void set_status_code(grpc_call *call, status_source source,
}
}
static void set_decode_compression_level(grpc_call *call,
grpc_compression_level clevel) {
call->compression_level = clevel;
static void set_compression_algorithm(grpc_call *call,
grpc_compression_algorithm algo) {
call->compression_algorithm = algo;
}
grpc_compression_algorithm grpc_call_get_compression_algorithm(
const grpc_call *call) {
return call->compression_algorithm;
}
static void set_status_details(grpc_call *call, status_source source,
@ -762,8 +768,18 @@ static void call_on_done_send(void *pc, int success) {
static void finish_message(grpc_call *call) {
if (call->error_status_set == 0) {
/* TODO(ctiller): this could be a lot faster if coded directly */
grpc_byte_buffer *byte_buffer = grpc_raw_byte_buffer_create(
call->incoming_message.slices, call->incoming_message.count);
grpc_byte_buffer *byte_buffer;
/* some aliases for readability */
gpr_slice *slices = call->incoming_message.slices;
const size_t nslices = call->incoming_message.count;
if ((call->incoming_message_flags & GRPC_WRITE_INTERNAL_COMPRESS) &&
(call->compression_algorithm > GRPC_COMPRESS_NONE)) {
byte_buffer = grpc_raw_compressed_byte_buffer_create(
slices, nslices, call->compression_algorithm);
} else {
byte_buffer = grpc_raw_byte_buffer_create(slices, nslices);
}
grpc_bbq_push(&call->incoming_queue, byte_buffer);
}
gpr_slice_buffer_reset_and_unref(&call->incoming_message);
@ -782,6 +798,25 @@ static int begin_message(grpc_call *call, grpc_begin_message msg) {
gpr_free(message);
return 0;
}
/* sanity check: if message flags indicate a compressed message, the
* compression level should already be present in the call, as parsed off its
* corresponding metadata. */
if ((msg.flags & GRPC_WRITE_INTERNAL_COMPRESS) &&
(call->compression_algorithm == GRPC_COMPRESS_NONE)) {
char *message = NULL;
char *alg_name;
if (!grpc_compression_algorithm_name(call->compression_algorithm,
&alg_name)) {
/* This shouldn't happen, other than due to data corruption */
alg_name = "<unknown>";
}
gpr_asprintf(&message,
"Invalid compression algorithm (%s) for compressed message.",
alg_name);
cancel_with_status(call, GRPC_STATUS_INTERNAL, message);
gpr_free(message);
return 0;
}
/* stash away parameters, and prepare for incoming slices */
if (msg.length > grpc_channel_get_max_message_length(call->channel)) {
char *message = NULL;
@ -1276,25 +1311,20 @@ static gpr_uint32 decode_status(grpc_mdelem *md) {
static void destroy_compression(void *ignored) {}
static gpr_uint32 decode_compression(grpc_mdelem *md) {
grpc_compression_level clevel;
void *user_data = grpc_mdelem_get_user_data(md, destroy_status);
grpc_compression_algorithm algorithm;
void *user_data = grpc_mdelem_get_user_data(md, destroy_compression);
if (user_data) {
clevel = ((grpc_compression_level)(gpr_intptr)user_data) - COMPRESS_OFFSET;
algorithm = ((grpc_compression_level)(gpr_intptr)user_data) - COMPRESS_OFFSET;
} else {
gpr_uint32 parsed_clevel_bytes;
if (gpr_parse_bytes_to_uint32(grpc_mdstr_as_c_string(md->value),
GPR_SLICE_LENGTH(md->value->slice),
&parsed_clevel_bytes)) {
/* the following cast is safe, as a gpr_uint32 should be able to hold all
* possible values of the grpc_compression_level enum */
clevel = (grpc_compression_level)parsed_clevel_bytes;
} else {
clevel = GRPC_COMPRESS_LEVEL_NONE; /* could not parse, no compression */
const char *md_c_str = grpc_mdstr_as_c_string(md->value);
if (!grpc_compression_algorithm_parse(md_c_str, &algorithm)) {
gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'", md_c_str);
assert(0);
}
grpc_mdelem_set_user_data(md, destroy_compression,
(void *)(gpr_intptr)(clevel + COMPRESS_OFFSET));
(void *)(gpr_intptr)(algorithm + COMPRESS_OFFSET));
}
return clevel;
return algorithm;
}
static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) {
@ -1313,8 +1343,8 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) {
} else if (key == grpc_channel_get_message_string(call->channel)) {
set_status_details(call, STATUS_FROM_WIRE, GRPC_MDSTR_REF(md->value));
} else if (key ==
grpc_channel_get_compresssion_level_string(call->channel)) {
set_decode_compression_level(call, decode_compression(md));
grpc_channel_get_compression_algorithm_string(call->channel)) {
set_compression_algorithm(call, decode_compression(md));
} else {
dest = &call->buffered_metadata[is_trailing];
if (dest->count == dest->capacity) {
@ -1429,7 +1459,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
req = &reqs[out++];
req->op = GRPC_IOREQ_SEND_MESSAGE;
req->data.send_message = op->data.send_message;
req->flags = ops->flags;
req->flags = op->flags;
break;
case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
/* Flag validation: currently allow no flags */

@ -63,7 +63,7 @@ struct grpc_channel {
grpc_mdctx *metadata_context;
/** mdstr for the grpc-status key */
grpc_mdstr *grpc_status_string;
grpc_mdstr *grpc_compression_level_string;
grpc_mdstr *grpc_compression_algorithm_string;
grpc_mdstr *grpc_message_string;
grpc_mdstr *path_string;
grpc_mdstr *authority_string;
@ -98,8 +98,8 @@ grpc_channel *grpc_channel_create_from_filters(
gpr_ref_init(&channel->refs, 1);
channel->metadata_context = mdctx;
channel->grpc_status_string = grpc_mdstr_from_string(mdctx, "grpc-status");
channel->grpc_compression_level_string =
grpc_mdstr_from_string(mdctx, "grpc-compression-level");
channel->grpc_compression_algorithm_string =
grpc_mdstr_from_string(mdctx, "grpc-encoding");
channel->grpc_message_string = grpc_mdstr_from_string(mdctx, "grpc-message");
for (i = 0; i < NUM_CACHED_STATUS_ELEMS; i++) {
char buf[GPR_LTOA_MIN_BUFSIZE];
@ -209,7 +209,7 @@ static void destroy_channel(void *p, int ok) {
GRPC_MDELEM_UNREF(channel->grpc_status_elem[i]);
}
GRPC_MDSTR_UNREF(channel->grpc_status_string);
GRPC_MDSTR_UNREF(channel->grpc_compression_level_string);
GRPC_MDSTR_UNREF(channel->grpc_compression_algorithm_string);
GRPC_MDSTR_UNREF(channel->grpc_message_string);
GRPC_MDSTR_UNREF(channel->path_string);
GRPC_MDSTR_UNREF(channel->authority_string);
@ -262,8 +262,9 @@ grpc_mdstr *grpc_channel_get_status_string(grpc_channel *channel) {
return channel->grpc_status_string;
}
grpc_mdstr *grpc_channel_get_compresssion_level_string(grpc_channel *channel) {
return channel->grpc_compression_level_string;
grpc_mdstr *grpc_channel_get_compression_algorithm_string(
grpc_channel *channel) {
return channel->grpc_compression_algorithm_string;
}
grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) {

@ -54,7 +54,8 @@ grpc_mdctx *grpc_channel_get_metadata_context(grpc_channel *channel);
grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel,
int status_code);
grpc_mdstr *grpc_channel_get_status_string(grpc_channel *channel);
grpc_mdstr *grpc_channel_get_compresssion_level_string(grpc_channel *channel);
grpc_mdstr *grpc_channel_get_compression_algorithm_string(
grpc_channel *channel);
grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel);
gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel);

@ -40,6 +40,7 @@
#include "src/core/channel/channel_args.h"
#include "src/core/channel/client_channel.h"
#include "src/core/channel/compress_filter.h"
#include "src/core/channel/http_client_filter.h"
#include "src/core/client_config/resolver_registry.h"
#include "src/core/iomgr/tcp_client.h"
@ -163,6 +164,7 @@ grpc_channel *grpc_channel_create(const char *target,
if (grpc_channel_args_is_census_enabled(args)) {
filters[n++] = &grpc_client_census_filter;
} */
filters[n++] = &grpc_compress_filter;
filters[n++] = &grpc_client_channel_filter;
GPR_ASSERT(n <= MAX_FILTERS);

@ -40,6 +40,7 @@
#include "src/core/channel/channel_args.h"
#include "src/core/channel/client_channel.h"
#include "src/core/channel/compress_filter.h"
#include "src/core/channel/http_client_filter.h"
#include "src/core/client_config/resolver_registry.h"
#include "src/core/iomgr/tcp_client.h"
@ -213,6 +214,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
if (grpc_channel_args_is_census_enabled(args)) {
filters[n++] = &grpc_client_census_filter;
} */
filters[n++] = &grpc_compress_filter;
filters[n++] = &grpc_client_channel_filter;
GPR_ASSERT(n <= MAX_FILTERS);

@ -739,9 +739,9 @@ void grpc_server_register_completion_queue(grpc_server *server,
server->cqs[n] = cq;
}
grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
size_t filter_count,
const grpc_channel_args *args) {
grpc_server *grpc_server_create_from_filters(
const grpc_channel_filter **filters, size_t filter_count,
const grpc_channel_args *args) {
size_t i;
/* TODO(census): restore this once we finalize census filter etc.
int census_enabled = grpc_channel_args_is_census_enabled(args); */

@ -39,9 +39,9 @@
#include "src/core/transport/transport.h"
/* Create a server */
grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
size_t filter_count,
const grpc_channel_args *args);
grpc_server *grpc_server_create_from_filters(
const grpc_channel_filter **filters, size_t filter_count,
const grpc_channel_args *args);
/* Add a listener to the server: when the server starts, it will call start,
and when it shuts down, it will call destroy */

@ -34,7 +34,10 @@
#include <grpc/grpc.h>
#include "src/core/surface/completion_queue.h"
#include "src/core/surface/server.h"
#include "src/core/channel/compress_filter.h"
grpc_server *grpc_server_create(const grpc_channel_args *args) {
return grpc_server_create_from_filters(NULL, 0, args);
const grpc_channel_filter *filters[] = {&grpc_compress_filter};
return grpc_server_create_from_filters(filters, GPR_ARRAY_SIZE(filters),
args);
}

@ -76,6 +76,7 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
gpr_uint8 *const end = GPR_SLICE_END_PTR(slice);
gpr_uint8 *cur = beg;
grpc_chttp2_data_parser *p = parser;
gpr_uint32 message_flags = 0;
if (is_last && p->is_last_frame) {
stream_parsing->received_close = 1;
@ -94,8 +95,8 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
/* noop */
break;
case 1:
gpr_log(GPR_ERROR, "Compressed GRPC frames not yet supported");
return GRPC_CHTTP2_STREAM_ERROR;
p->is_frame_compressed = 1; /* GPR_TRUE */
break;
default:
gpr_log(GPR_ERROR, "Bad GRPC frame type 0x%02x", p->frame_type);
return GRPC_CHTTP2_STREAM_ERROR;
@ -130,7 +131,11 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
p->frame_size |= ((gpr_uint32)*cur);
p->state = GRPC_CHTTP2_DATA_FRAME;
++cur;
grpc_sopb_add_begin_message(&p->incoming_sopb, p->frame_size, 0);
if (p->is_frame_compressed) {
message_flags |= GRPC_WRITE_INTERNAL_COMPRESS;
}
grpc_sopb_add_begin_message(&p->incoming_sopb, p->frame_size,
message_flags);
/* fallthrough */
case GRPC_CHTTP2_DATA_FRAME:
if (cur == end) {

@ -56,6 +56,7 @@ typedef struct {
gpr_uint8 frame_type;
gpr_uint32 frame_size;
int is_frame_compressed;
grpc_stream_op_buffer incoming_sopb;
} grpc_chttp2_data_parser;

@ -477,6 +477,7 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count,
gpr_uint32 flow_controlled_bytes_taken = 0;
gpr_uint32 curop = 0;
gpr_uint8 *p;
int compressed_flag_set = 0;
while (curop < *inops_count) {
GPR_ASSERT(flow_controlled_bytes_taken <= max_flow_controlled_bytes);
@ -496,9 +497,12 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count,
case GRPC_OP_BEGIN_MESSAGE:
/* begin op: for now we just convert the op to a slice and fall
through - this lets us reuse the slice framing code below */
compressed_flag_set =
(op->data.begin_message.flags & GRPC_WRITE_INTERNAL_COMPRESS) != 0;
slice = gpr_slice_malloc(5);
p = GPR_SLICE_START_PTR(slice);
p[0] = 0;
p[0] = compressed_flag_set;
p[1] = op->data.begin_message.length >> 24;
p[2] = op->data.begin_message.length >> 16;
p[3] = op->data.begin_message.length >> 8;

@ -286,6 +286,12 @@ void grpc_metadata_batch_merge(grpc_metadata_batch *target,
}
}
void grpc_metadata_batch_move(grpc_metadata_batch *dst,
grpc_metadata_batch *src) {
*dst = *src;
memset(src, 0, sizeof(grpc_metadata_batch));
}
void grpc_metadata_batch_filter(grpc_metadata_batch *batch,
grpc_mdelem *(*filter)(void *user_data,
grpc_mdelem *elem),

@ -102,6 +102,11 @@ void grpc_metadata_batch_destroy(grpc_metadata_batch *batch);
void grpc_metadata_batch_merge(grpc_metadata_batch *target,
grpc_metadata_batch *add);
/** Moves the metadata information from \a src to \a dst. Upon return, \a src is
* zeroed. */
void grpc_metadata_batch_move(grpc_metadata_batch *dst,
grpc_metadata_batch *src);
/** Add \a storage to the beginning of \a batch. storage->md is
assumed to be valid.
\a storage is owned by the caller and must survive for the

@ -75,8 +75,9 @@ void ChannelArguments::Swap(ChannelArguments& other) {
strings_.swap(other.strings_);
}
void ChannelArguments::SetCompressionLevel(grpc_compression_level level) {
SetInt(GRPC_COMPRESSION_LEVEL_ARG, level);
void ChannelArguments::_Experimental_SetCompressionAlgorithm(
grpc_compression_algorithm algorithm) {
SetInt(GRPC_COMPRESSION_ALGORITHM_ARG, algorithm);
}
void ChannelArguments::SetInt(const grpc::string& key, int value) {

@ -34,8 +34,11 @@
#include <grpc++/client_context.h>
#include <grpc/grpc.h>
#include <grpc/support/string_util.h>
#include <grpc++/credentials.h>
#include <grpc++/time.h>
#include "src/core/channel/compress_filter.h"
#include "src/cpp/common/create_auth_context.h"
namespace grpc {
@ -76,6 +79,18 @@ void ClientContext::set_call(grpc_call* call,
}
}
void ClientContext::_experimental_set_compression_algorithm(
grpc_compression_algorithm algorithm) {
char* algorithm_name = NULL;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
gpr_log(GPR_ERROR, "Name for compression algorithm '%d' unknown.",
algorithm);
abort();
}
GPR_ASSERT(algorithm_name != NULL);
AddMetadata(GRPC_COMPRESS_REQUEST_ALGORITHM_KEY, algorithm_name);
}
std::shared_ptr<const AuthContext> ClientContext::auth_context() const {
if (auth_context_.get() == nullptr) {
auth_context_ = CreateAuthContext(call_);

@ -103,7 +103,9 @@ class GrpcBufferReader GRPC_FINAL
: byte_count_(0), backup_count_(0) {
grpc_byte_buffer_reader_init(&reader_, buffer);
}
~GrpcBufferReader() GRPC_OVERRIDE {}
~GrpcBufferReader() GRPC_OVERRIDE {
grpc_byte_buffer_reader_destroy(&reader_);
}
bool Next(const void** data, int* size) GRPC_OVERRIDE {
if (backup_count_ > 0) {

@ -39,6 +39,7 @@
#include <grpc++/impl/sync.h>
#include <grpc++/time.h>
#include "src/core/channel/compress_filter.h"
#include "src/cpp/common/create_auth_context.h"
namespace grpc {
@ -148,6 +149,24 @@ bool ServerContext::IsCancelled() const {
return completion_op_ && completion_op_->CheckCancelled(cq_);
}
void ServerContext::set_compression_level(grpc_compression_level level) {
const grpc_compression_algorithm algorithm_for_level =
grpc_compression_algorithm_for_level(level);
set_compression_algorithm(algorithm_for_level);
}
void ServerContext::set_compression_algorithm(
grpc_compression_algorithm algorithm) {
char* algorithm_name = NULL;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
gpr_log(GPR_ERROR, "Name for compression algorithm '%d' unknown.",
algorithm);
abort();
}
GPR_ASSERT(algorithm_name != NULL);
AddInitialMetadata(GRPC_COMPRESS_REQUEST_ALGORITHM_KEY, algorithm_name);
}
void ServerContext::set_call(grpc_call* call) {
call_ = call;
auth_context_ = CreateAuthContext(call);

@ -131,7 +131,8 @@ static size_t md_ary_datasize(const void *p) {
static const rb_data_type_t grpc_rb_md_ary_data_type = {
"grpc_metadata_array",
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, md_ary_datasize},
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, md_ary_datasize,
{NULL, NULL}},
NULL,
NULL,
0};
@ -139,7 +140,8 @@ static const rb_data_type_t grpc_rb_md_ary_data_type = {
/* Describes grpc_call struct for RTypedData */
static const rb_data_type_t grpc_call_data_type = {
"grpc_call",
{GRPC_RB_GC_NOT_MARKED, grpc_rb_call_destroy, GRPC_RB_MEMSIZE_UNAVAILABLE},
{GRPC_RB_GC_NOT_MARKED, grpc_rb_call_destroy, GRPC_RB_MEMSIZE_UNAVAILABLE,
{NULL, NULL}},
NULL,
NULL,
/* it is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because
@ -275,6 +277,8 @@ static int grpc_rb_md_ary_capacity_hash_cb(VALUE key, VALUE val,
VALUE md_ary_obj) {
grpc_metadata_array *md_ary = NULL;
(void)key;
/* Construct a metadata object from key and value and add it */
TypedData_Get_Struct(md_ary_obj, grpc_metadata_array,
&grpc_rb_md_ary_data_type, md_ary);
@ -348,6 +352,7 @@ VALUE grpc_rb_md_ary_to_h(grpc_metadata_array *md_ary) {
*/
static int grpc_rb_call_check_op_keys_hash_cb(VALUE key, VALUE val,
VALUE ops_ary) {
(void)val;
/* Update the capacity; the value is an array, add capacity for each value in
* the array */
if (TYPE(key) != T_FIXNUM) {

@ -107,7 +107,8 @@ static void grpc_rb_channel_mark(void *p) {
static rb_data_type_t grpc_channel_data_type = {
"grpc_channel",
{grpc_rb_channel_mark, grpc_rb_channel_free, GRPC_RB_MEMSIZE_UNAVAILABLE},
{grpc_rb_channel_mark, grpc_rb_channel_free, GRPC_RB_MEMSIZE_UNAVAILABLE,
{NULL, NULL}},
NULL, NULL,
RUBY_TYPED_FREE_IMMEDIATELY
};

@ -41,7 +41,8 @@
static rb_data_type_t grpc_rb_channel_args_data_type = {
"grpc_channel_args",
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, GRPC_RB_MEMSIZE_UNAVAILABLE},
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, GRPC_RB_MEMSIZE_UNAVAILABLE,
{NULL, NULL}},
NULL, NULL,
RUBY_TYPED_FREE_IMMEDIATELY
};

@ -119,7 +119,7 @@ static void grpc_rb_completion_queue_destroy(void *p) {
static rb_data_type_t grpc_rb_completion_queue_data_type = {
"grpc_completion_queue",
{GRPC_RB_GC_NOT_MARKED, grpc_rb_completion_queue_destroy,
GRPC_RB_MEMSIZE_UNAVAILABLE},
GRPC_RB_MEMSIZE_UNAVAILABLE, {NULL, NULL}},
NULL, NULL,
/* cannot immediately free because grpc_rb_completion_queue_shutdown_drain
* calls rb_thread_call_without_gvl. */

@ -89,7 +89,7 @@ static void grpc_rb_credentials_mark(void *p) {
static rb_data_type_t grpc_rb_credentials_data_type = {
"grpc_credentials",
{grpc_rb_credentials_mark, grpc_rb_credentials_free,
GRPC_RB_MEMSIZE_UNAVAILABLE},
GRPC_RB_MEMSIZE_UNAVAILABLE, {NULL, NULL}},
NULL,
NULL,
RUBY_TYPED_FREE_IMMEDIATELY};

@ -51,7 +51,8 @@ static VALUE grpc_rb_cTimeVal = Qnil;
static rb_data_type_t grpc_rb_timespec_data_type = {
"gpr_timespec",
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, GRPC_RB_MEMSIZE_UNAVAILABLE},
{GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, GRPC_RB_MEMSIZE_UNAVAILABLE,
{NULL, NULL}},
NULL,
NULL,
RUBY_TYPED_FREE_IMMEDIATELY};
@ -75,6 +76,7 @@ VALUE grpc_rb_cannot_init(VALUE self) {
/* Init/Clone func that fails by raising an exception. */
VALUE grpc_rb_cannot_init_copy(VALUE copy, VALUE self) {
(void)self;
rb_raise(rb_eTypeError,
"initialization of %s only allowed from the gRPC native layer",
rb_obj_classname(copy));
@ -258,7 +260,10 @@ static void Init_grpc_time_consts() {
id_tv_nsec = rb_intern("tv_nsec");
}
static void grpc_rb_shutdown(ruby_vm_t *vm) { grpc_shutdown(); }
static void grpc_rb_shutdown(ruby_vm_t *vm) {
(void)vm;
grpc_shutdown();
}
/* Initialize the GRPC module structs */

@ -94,7 +94,8 @@ static void grpc_rb_server_mark(void *p) {
static const rb_data_type_t grpc_rb_server_data_type = {
"grpc_server",
{grpc_rb_server_mark, grpc_rb_server_free, GRPC_RB_MEMSIZE_UNAVAILABLE},
{grpc_rb_server_mark, grpc_rb_server_free, GRPC_RB_MEMSIZE_UNAVAILABLE,
{NULL, NULL}},
NULL,
NULL,
/* It is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because the free function would block

@ -89,7 +89,7 @@ static void grpc_rb_server_credentials_mark(void *p) {
static const rb_data_type_t grpc_rb_server_credentials_data_type = {
"grpc_server_credentials",
{grpc_rb_server_credentials_mark, grpc_rb_server_credentials_free,
GRPC_RB_MEMSIZE_UNAVAILABLE},
GRPC_RB_MEMSIZE_UNAVAILABLE, {NULL, NULL}},
NULL, NULL,
RUBY_TYPED_FREE_IMMEDIATELY
};

@ -61,13 +61,15 @@ static void assert_passthrough(gpr_slice value,
gpr_slice_buffer output;
gpr_slice final;
int was_compressed;
char *algorithm_name;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algorithm_name) != 0);
gpr_log(GPR_INFO,
"assert_passthrough: value_length=%d value_hash=0x%08x "
"algorithm='%s' uncompressed_split='%s' compressed_split='%s'",
GPR_SLICE_LENGTH(value), gpr_murmur_hash3(GPR_SLICE_START_PTR(value),
GPR_SLICE_LENGTH(value), 0),
grpc_compression_algorithm_name(algorithm),
algorithm_name,
grpc_slice_split_mode_name(uncompressed_split_mode),
grpc_slice_split_mode_name(compressed_split_mode));

@ -40,6 +40,7 @@
#include "src/core/surface/event_string.h"
#include "src/core/support/string.h"
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
@ -144,7 +145,17 @@ static int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b) {
}
int byte_buffer_eq_string(grpc_byte_buffer *bb, const char *str) {
return byte_buffer_eq_slice(bb, gpr_slice_from_copied_string(str));
grpc_byte_buffer_reader reader;
grpc_byte_buffer* rbb;
int res;
grpc_byte_buffer_reader_init(&reader, bb);
rbb = grpc_raw_byte_buffer_from_reader(&reader);
res = byte_buffer_eq_slice(rbb, gpr_slice_from_copied_string(str));
grpc_byte_buffer_reader_destroy(&reader);
grpc_byte_buffer_destroy(rbb);
return res;
}
static void verify_matches(expectation *e, grpc_event *ev) {

@ -0,0 +1,135 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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 "test/core/end2end/end2end_tests.h"
#include <string.h>
#include "src/core/channel/channel_args.h"
#include "src/core/channel/client_channel.h"
#include "src/core/channel/connected_channel.h"
#include "src/core/channel/http_server_filter.h"
#include "src/core/surface/channel.h"
#include "src/core/surface/server.h"
#include "src/core/transport/chttp2_transport.h"
#include <grpc/support/alloc.h>
#include <grpc/support/host_port.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include <grpc/support/thd.h>
#include <grpc/support/useful.h>
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
typedef struct fullstack_compression_fixture_data {
char *localaddr;
grpc_channel_args* client_args_compression;
grpc_channel_args* server_args_compression;
} fullstack_compression_fixture_data;
static grpc_end2end_test_fixture chttp2_create_fixture_fullstack_compression(
grpc_channel_args *client_args, grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
int port = grpc_pick_unused_port_or_die();
fullstack_compression_fixture_data *ffd =
gpr_malloc(sizeof(fullstack_compression_fixture_data));
memset(ffd, 0, sizeof(fullstack_compression_fixture_data));
gpr_join_host_port(&ffd->localaddr, "localhost", port);
memset(&f, 0, sizeof(f));
f.fixture_data = ffd;
f.cq = grpc_completion_queue_create();
return f;
}
void chttp2_init_client_fullstack_compression(grpc_end2end_test_fixture *f,
grpc_channel_args *client_args) {
fullstack_compression_fixture_data *ffd = f->fixture_data;
if (ffd->client_args_compression != NULL) {
grpc_channel_args_destroy(ffd->client_args_compression);
}
ffd->client_args_compression = grpc_channel_args_set_compression_algorithm(
client_args, GRPC_COMPRESS_GZIP);
f->client = grpc_channel_create(ffd->localaddr, ffd->client_args_compression);
}
void chttp2_init_server_fullstack_compression(grpc_end2end_test_fixture *f,
grpc_channel_args *server_args) {
fullstack_compression_fixture_data *ffd = f->fixture_data;
if (ffd->server_args_compression != NULL) {
grpc_channel_args_destroy(ffd->server_args_compression);
}
ffd->server_args_compression = grpc_channel_args_set_compression_algorithm(
server_args, GRPC_COMPRESS_GZIP);
if (f->server) {
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(ffd->server_args_compression);
grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
void chttp2_tear_down_fullstack_compression(grpc_end2end_test_fixture *f) {
fullstack_compression_fixture_data *ffd = f->fixture_data;
grpc_channel_args_destroy(ffd->client_args_compression);
grpc_channel_args_destroy(ffd->server_args_compression);
gpr_free(ffd->localaddr);
gpr_free(ffd);
}
/* All test configurations */
static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack_compression", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION,
chttp2_create_fixture_fullstack_compression,
chttp2_init_client_fullstack_compression,
chttp2_init_server_fullstack_compression,
chttp2_tear_down_fullstack_compression},
};
int main(int argc, char **argv) {
size_t i;
grpc_test_init(argc, argv);
grpc_init();
for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
grpc_end2end_tests(configs[i]);
}
grpc_shutdown();
return 0;
}

@ -36,6 +36,7 @@
#include <string.h>
#include "src/core/channel/client_channel.h"
#include "src/core/channel/compress_filter.h"
#include "src/core/channel/connected_channel.h"
#include "src/core/channel/http_client_filter.h"
#include "src/core/channel/http_server_filter.h"
@ -75,6 +76,7 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
sp_client_setup *cs = ts;
const grpc_channel_filter *filters[] = {&grpc_http_client_filter,
&grpc_compress_filter,
&grpc_connected_channel_filter};
size_t nfilters = sizeof(filters) / sizeof(*filters);
grpc_channel *channel = grpc_channel_create_from_filters(

@ -39,6 +39,7 @@
#include "src/core/channel/connected_channel.h"
#include "src/core/channel/http_client_filter.h"
#include "src/core/channel/http_server_filter.h"
#include "src/core/channel/compress_filter.h"
#include "src/core/iomgr/endpoint_pair.h"
#include "src/core/iomgr/iomgr.h"
#include "src/core/surface/channel.h"
@ -75,6 +76,7 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
sp_client_setup *cs = ts;
const grpc_channel_filter *filters[] = {&grpc_http_client_filter,
&grpc_compress_filter,
&grpc_connected_channel_filter};
size_t nfilters = sizeof(filters) / sizeof(*filters);
grpc_channel *channel = grpc_channel_create_from_filters(

@ -39,6 +39,7 @@
#include "src/core/channel/connected_channel.h"
#include "src/core/channel/http_client_filter.h"
#include "src/core/channel/http_server_filter.h"
#include "src/core/channel/compress_filter.h"
#include "src/core/iomgr/endpoint_pair.h"
#include "src/core/iomgr/iomgr.h"
#include "src/core/support/env.h"
@ -76,6 +77,7 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
sp_client_setup *cs = ts;
const grpc_channel_filter *filters[] = {&grpc_http_client_filter,
&grpc_compress_filter,
&grpc_connected_channel_filter};
size_t nfilters = sizeof(filters) / sizeof(*filters);
grpc_channel *channel = grpc_channel_create_from_filters(

@ -44,6 +44,7 @@ default_secure_fixture_options = FixtureOptions(True, ['windows', 'posix'])
END2END_FIXTURES = {
'chttp2_fake_security': default_secure_fixture_options,
'chttp2_fullstack': default_unsecure_fixture_options,
'chttp2_fullstack_compression': default_unsecure_fixture_options,
'chttp2_fullstack_with_poll': FixtureOptions(False, ['posix']),
'chttp2_fullstack_uds_posix': FixtureOptions(False, ['posix']),
'chttp2_simple_ssl_fullstack': default_secure_fixture_options,
@ -84,6 +85,7 @@ END2END_TESTS = {
'request_response_with_payload_and_call_creds': TestOptions(flaky=False, secure=True),
'request_with_large_metadata': default_test_options,
'request_with_payload': default_test_options,
'request_with_compressed_payload': default_test_options,
'request_with_flags': default_test_options,
'server_finishes_request': default_test_options,
'simple_delayed_request': default_test_options,

@ -0,0 +1,315 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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 "test/core/end2end/end2end_tests.h"
#include <stdio.h>
#include <string.h>
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
#include "test/core/end2end/cq_verifier.h"
#include "src/core/channel/channel_args.h"
#include "src/core/channel/compress_filter.h"
enum { TIMEOUT = 200000 };
static void *tag(gpr_intptr t) { return (void *)t; }
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
const char *test_name,
grpc_channel_args *client_args,
grpc_channel_args *server_args) {
grpc_end2end_test_fixture f;
gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
f = config.create_fixture(client_args, server_args);
config.init_client(&f, client_args);
config.init_server(&f, server_args);
return f;
}
static gpr_timespec n_seconds_time(int n) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
}
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
} while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000),
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5))
.type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server);
f->server = NULL;
}
static void shutdown_client(grpc_end2end_test_fixture *f) {
if (!f->client) return;
grpc_channel_destroy(f->client);
f->client = NULL;
}
static void end_test(grpc_end2end_test_fixture *f) {
shutdown_server(f);
shutdown_client(f);
grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq);
}
static void request_with_payload_template(
grpc_end2end_test_config config, const char *test_name,
gpr_uint32 send_flags_bitmask,
grpc_compression_algorithm requested_compression_algorithm,
grpc_compression_algorithm expected_compression_algorithm,
grpc_metadata *client_metadata) {
grpc_call *c;
grpc_call *s;
gpr_slice request_payload_slice;
grpc_byte_buffer *request_payload;
gpr_timespec deadline = five_seconds_time();
grpc_channel_args *client_args;
grpc_channel_args *server_args;
grpc_end2end_test_fixture f;
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_byte_buffer *request_payload_recv = NULL;
grpc_call_details call_details;
grpc_status_code status;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
cq_verifier *cqv;
char str[1024];
memset(str, 'x', 1023); str[1023] = '\0';
request_payload_slice = gpr_slice_from_copied_string(str);
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
client_args = grpc_channel_args_set_compression_algorithm(
NULL, requested_compression_algorithm);
server_args = grpc_channel_args_set_compression_algorithm(
NULL, requested_compression_algorithm);
f = begin_test(config, test_name, client_args, server_args);
cqv = cq_verifier_create(f.cq);
c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
if (client_metadata != NULL) {
op->data.send_initial_metadata.count = 1;
op->data.send_initial_metadata.metadata = client_metadata;
} else {
op->data.send_initial_metadata.count = 0;
}
op->flags = 0;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message = request_payload;
op->flags = send_flags_bitmask;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op->flags = 0;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op->flags = 0;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op->flags = 0;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message = &request_payload_recv;
op->flags = 0;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(cqv, tag(102), 1);
cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op++;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_OK;
op->data.send_status_from_server.status_details = "xyz";
op->flags = 0;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103)));
cq_expect_completion(cqv, tag(103), 1);
cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr"));
GPR_ASSERT(was_cancelled == 0);
GPR_ASSERT(request_payload_recv->type == GRPC_BB_RAW);
GPR_ASSERT(request_payload_recv->data.raw.compression ==
expected_compression_algorithm);
GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, str));
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(cqv);
gpr_slice_unref(request_payload_slice);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(request_payload_recv);
grpc_channel_args_destroy(client_args);
grpc_channel_args_destroy(server_args);
end_test(&f);
config.tear_down_data(&f);
}
static void test_invoke_request_with_exceptionally_uncompressed_payload(
grpc_end2end_test_config config) {
request_with_payload_template(
config, "test_invoke_request_with_exceptionally_uncompressed_payload",
GRPC_WRITE_NO_COMPRESS, GRPC_COMPRESS_GZIP, GRPC_COMPRESS_NONE,
NULL);
}
static void test_invoke_request_with_uncompressed_payload(
grpc_end2end_test_config config) {
request_with_payload_template(
config, "test_invoke_request_with_uncompressed_payload", 0,
GRPC_COMPRESS_NONE, GRPC_COMPRESS_NONE, NULL);
}
static void test_invoke_request_with_compressed_payload(
grpc_end2end_test_config config) {
request_with_payload_template(
config, "test_invoke_request_with_compressed_payload", 0,
GRPC_COMPRESS_GZIP, GRPC_COMPRESS_GZIP, NULL);
}
static void test_invoke_request_with_compressed_payload_md_override(
grpc_end2end_test_config config) {
grpc_metadata gzip_compression_override;
grpc_metadata none_compression_override;
gzip_compression_override.key = GRPC_COMPRESS_REQUEST_ALGORITHM_KEY;
gzip_compression_override.value = "gzip";
gzip_compression_override.value_length = 4;
memset(&gzip_compression_override.internal_data, 0,
sizeof(gzip_compression_override.internal_data));
none_compression_override.key = GRPC_COMPRESS_REQUEST_ALGORITHM_KEY;
none_compression_override.value = "none";
none_compression_override.value_length = 4;
memset(&none_compression_override.internal_data, 0,
sizeof(none_compression_override.internal_data));
/* Channel default NONE, call override to GZIP */
request_with_payload_template(
config, "test_invoke_request_with_compressed_payload_md_override_1", 0,
GRPC_COMPRESS_NONE, GRPC_COMPRESS_GZIP, &gzip_compression_override);
/* Channel default DEFLATE, call override to GZIP */
request_with_payload_template(
config, "test_invoke_request_with_compressed_payload_md_override_2", 0,
GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_GZIP, &gzip_compression_override);
/* Channel default DEFLATE, call override to NONE */
request_with_payload_template(
config, "test_invoke_request_with_compressed_payload_md_override_3", 0,
GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_NONE, &none_compression_override);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_invoke_request_with_exceptionally_uncompressed_payload(config);
test_invoke_request_with_uncompressed_payload(config);
test_invoke_request_with_compressed_payload(config);
test_invoke_request_with_compressed_payload_md_override(config);
}

@ -44,35 +44,7 @@
#include "src/core/security/credentials.h"
#include "src/core/support/file.h"
typedef struct {
grpc_pollset pollset;
int is_done;
} synchronizer;
static void on_oauth2_response(void *user_data,
grpc_credentials_md *md_elems,
size_t num_md, grpc_credentials_status status) {
synchronizer *sync = user_data;
char *token;
gpr_slice token_slice;
if (status == GRPC_CREDENTIALS_ERROR) {
gpr_log(GPR_ERROR, "Fetching token failed.");
} else {
GPR_ASSERT(num_md == 1);
token_slice = md_elems[0].value;
token = gpr_malloc(GPR_SLICE_LENGTH(token_slice) + 1);
memcpy(token, GPR_SLICE_START_PTR(token_slice),
GPR_SLICE_LENGTH(token_slice));
token[GPR_SLICE_LENGTH(token_slice)] = '\0';
printf("Got token: %s.\n", token);
gpr_free(token);
}
gpr_mu_lock(GRPC_POLLSET_MU(&sync->pollset));
sync->is_done = 1;
grpc_pollset_kick(&sync->pollset);
gpr_mu_unlock(GRPC_POLLSET_MU(&sync->pollset));
}
#include "test/core/security/oauth2_utils.h"
static grpc_credentials *create_service_account_creds(
const char *json_key_file_path, const char *scope) {
@ -101,10 +73,10 @@ static grpc_credentials *create_refresh_token_creds(
}
int main(int argc, char **argv) {
synchronizer sync;
grpc_credentials *creds = NULL;
char *json_key_file_path = NULL;
char *json_refresh_token_file_path = NULL;
char *token = NULL;
int use_gce = 0;
char *scope = NULL;
gpr_cmdline *cl = gpr_cmdline_create("fetch_oauth2");
@ -175,17 +147,11 @@ int main(int argc, char **argv) {
}
GPR_ASSERT(creds != NULL);
grpc_pollset_init(&sync.pollset);
sync.is_done = 0;
grpc_credentials_get_request_metadata(creds, &sync.pollset, "", on_oauth2_response, &sync);
gpr_mu_lock(GRPC_POLLSET_MU(&sync.pollset));
while (!sync.is_done)
grpc_pollset_work(&sync.pollset, gpr_inf_future(GPR_CLOCK_REALTIME));
gpr_mu_unlock(GRPC_POLLSET_MU(&sync.pollset));
grpc_pollset_destroy(&sync.pollset);
token = grpc_test_fetch_oauth2_token_with_credentials(creds);
if (token != NULL) {
printf("Got token: %s.\n", token);
gpr_free(token);
}
grpc_credentials_release(creds);
gpr_cmdline_destroy(cl);
grpc_shutdown();

@ -0,0 +1,93 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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 "test/core/security/oauth2_utils.h"
#include <string.h>
#include <grpc/grpc.h>
#include <grpc/grpc_security.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/slice.h>
#include <grpc/support/sync.h>
#include "src/core/security/credentials.h"
typedef struct {
grpc_pollset pollset;
int is_done;
char *token;
} oauth2_request;
static void on_oauth2_response(void *user_data, grpc_credentials_md *md_elems,
size_t num_md, grpc_credentials_status status) {
oauth2_request *request = user_data;
char *token = NULL;
gpr_slice token_slice;
if (status == GRPC_CREDENTIALS_ERROR) {
gpr_log(GPR_ERROR, "Fetching token failed.");
} else {
GPR_ASSERT(num_md == 1);
token_slice = md_elems[0].value;
token = gpr_malloc(GPR_SLICE_LENGTH(token_slice) + 1);
memcpy(token, GPR_SLICE_START_PTR(token_slice),
GPR_SLICE_LENGTH(token_slice));
token[GPR_SLICE_LENGTH(token_slice)] = '\0';
}
gpr_mu_lock(GRPC_POLLSET_MU(&request->pollset));
request->is_done = 1;
request->token = token;
grpc_pollset_kick(&request->pollset);
gpr_mu_unlock(GRPC_POLLSET_MU(&request->pollset));
}
static void do_nothing(void *unused) {}
char *grpc_test_fetch_oauth2_token_with_credentials(grpc_credentials *creds) {
oauth2_request request;
grpc_pollset_init(&request.pollset);
request.is_done = 0;
grpc_credentials_get_request_metadata(creds, &request.pollset, "",
on_oauth2_response, &request);
gpr_mu_lock(GRPC_POLLSET_MU(&request.pollset));
while (!request.is_done)
grpc_pollset_work(&request.pollset, gpr_inf_future(GPR_CLOCK_REALTIME));
gpr_mu_unlock(GRPC_POLLSET_MU(&request.pollset));
grpc_pollset_shutdown(&request.pollset, do_nothing, NULL);
grpc_pollset_destroy(&request.pollset);
return request.token;
}

@ -0,0 +1,51 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * 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.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* 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.
*
*/
#ifndef GRPC_TEST_CORE_SECURITY_OAUTH2_UTILS_H
#define GRPC_TEST_CORE_SECURITY_OAUTH2_UTILS_H
#include "src/core/security/credentials.h"
#ifdef __cplusplus
extern "C" {
#endif
/* Fetch oauth2 access token with a credentials object. Does not take ownership.
Returns NULL on a failure. The caller should call gpr_free on the token. */
char *grpc_test_fetch_oauth2_token_with_credentials(grpc_credentials *creds);
#ifdef __cplusplus
}
#endif
#endif /* GRPC_TEST_CORE_SECURITY_OAUTH2_UTILS_H */

@ -173,6 +173,119 @@ static void test_asprintf(void) {
}
}
static void test_strjoin(void) {
const char *parts[4] = {"one", "two", "three", "four"};
size_t joined_len;
char *joined;
LOG_TEST_NAME("test_strjoin");
joined = gpr_strjoin(parts, 4, &joined_len);
GPR_ASSERT(0 == strcmp("onetwothreefour", joined));
gpr_free(joined);
joined = gpr_strjoin(parts, 0, &joined_len);
GPR_ASSERT(0 == strcmp("", joined));
gpr_free(joined);
joined = gpr_strjoin(parts, 1, &joined_len);
GPR_ASSERT(0 == strcmp("one", joined));
gpr_free(joined);
}
static void test_strjoin_sep(void) {
const char *parts[4] = {"one", "two", "three", "four"};
size_t joined_len;
char *joined;
LOG_TEST_NAME("test_strjoin_sep");
joined = gpr_strjoin_sep(parts, 4, ", ", &joined_len);
GPR_ASSERT(0 == strcmp("one, two, three, four", joined));
gpr_free(joined);
/* empty separator */
joined = gpr_strjoin_sep(parts, 4, "", &joined_len);
GPR_ASSERT(0 == strcmp("onetwothreefour", joined));
gpr_free(joined);
/* degenerated case specifying zero input parts */
joined = gpr_strjoin_sep(parts, 0, ", ", &joined_len);
GPR_ASSERT(0 == strcmp("", joined));
gpr_free(joined);
/* single part should have no separator */
joined = gpr_strjoin_sep(parts, 1, ", ", &joined_len);
GPR_ASSERT(0 == strcmp("one", joined));
gpr_free(joined);
}
static void test_strsplit(void) {
gpr_slice_buffer* parts;
gpr_slice str;
LOG_TEST_NAME("test_strsplit");
parts = gpr_malloc(sizeof(gpr_slice_buffer));
gpr_slice_buffer_init(parts);
str = gpr_slice_from_copied_string("one, two, three, four");
gpr_slice_split(str, ", ", parts);
GPR_ASSERT(4 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], "one"));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[1], "two"));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[2], "three"));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[3], "four"));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
/* separator not present in string */
str = gpr_slice_from_copied_string("one two three four");
gpr_slice_split(str, ", ", parts);
GPR_ASSERT(1 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], "one two three four"));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
/* separator at the end */
str = gpr_slice_from_copied_string("foo,");
gpr_slice_split(str, ",", parts);
GPR_ASSERT(2 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], "foo"));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[1], ""));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
/* separator at the beginning */
str = gpr_slice_from_copied_string(",foo");
gpr_slice_split(str, ",", parts);
GPR_ASSERT(2 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], ""));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[1], "foo"));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
/* standalone separator */
str = gpr_slice_from_copied_string(",");
gpr_slice_split(str, ",", parts);
GPR_ASSERT(2 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], ""));
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[1], ""));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
/* empty input */
str = gpr_slice_from_copied_string("");
gpr_slice_split(str, ", ", parts);
GPR_ASSERT(1 == parts->count);
GPR_ASSERT(0 == gpr_slice_str_cmp(parts->slices[0], ""));
gpr_slice_buffer_reset_and_unref(parts);
gpr_slice_unref(str);
gpr_slice_buffer_destroy(parts);
gpr_free(parts);
}
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
test_strdup();
@ -180,5 +293,8 @@ int main(int argc, char **argv) {
test_dump_slice();
test_parse_uint32();
test_asprintf();
test_strjoin();
test_strjoin_sep();
test_strsplit();
return 0;
}

@ -270,10 +270,11 @@ static void SendRpc(grpc::cpp::test::util::TestService::Stub* stub,
int num_rpcs) {
EchoRequest request;
EchoResponse response;
request.set_message("Hello");
request.set_message("Hello hello hello hello");
for (int i = 0; i < num_rpcs; ++i) {
ClientContext context;
context._experimental_set_compression_algorithm(GRPC_COMPRESS_GZIP);
Status s = stub->Echo(&context, request, &response);
EXPECT_EQ(response.message(), request.message());
EXPECT_TRUE(s.ok());

@ -227,6 +227,7 @@ TEST_F(GenericEnd2endTest, SimpleBidiStreaming) {
GenericServerContext srv_ctx;
GenericServerAsyncReaderWriter srv_stream(&srv_ctx);
cli_ctx._experimental_set_compression_algorithm(GRPC_COMPRESS_GZIP);
send_request.set_message("Hello");
std::unique_ptr<GenericClientAsyncReaderWriter> cli_stream =
generic_stub_->Call(&cli_ctx, kMethodName, &cli_cq_, tag(1));

@ -68,6 +68,7 @@ DEFINE_string(test_case, "large_unary",
"service_account_creds : large_unary with service_account auth; "
"compute_engine_creds: large_unary with compute engine auth; "
"jwt_token_creds: large_unary with JWT token auth; "
"oauth2_auth_token: raw oauth2 access token auth; "
"all : all of above.");
DEFINE_string(default_service_account, "",
"Email of GCE default service account");
@ -113,6 +114,9 @@ int main(int argc, char** argv) {
} else if (FLAGS_test_case == "jwt_token_creds") {
grpc::string json_key = GetServiceAccountJsonKey();
client.DoJwtTokenCreds(json_key);
} else if (FLAGS_test_case == "oauth2_auth_token") {
grpc::string json_key = GetServiceAccountJsonKey();
client.DoOauth2AuthToken(json_key, FLAGS_oauth_scope);
} else if (FLAGS_test_case == "all") {
client.DoEmpty();
client.DoLargeUnary();
@ -128,6 +132,7 @@ int main(int argc, char** argv) {
grpc::string json_key = GetServiceAccountJsonKey();
client.DoServiceAccountCreds(json_key, FLAGS_oauth_scope);
client.DoJwtTokenCreds(json_key);
client.DoOauth2AuthToken(json_key, FLAGS_oauth_scope);
}
// compute_engine_creds only runs in GCE.
} else {
@ -136,8 +141,8 @@ int main(int argc, char** argv) {
"Unsupported test case %s. Valid options are all|empty_unary|"
"large_unary|client_streaming|server_streaming|half_duplex|ping_pong|"
"cancel_after_begin|cancel_after_first_response|"
"timeout_on_sleeping_server|"
"service_account_creds|compute_engine_creds|jwt_token_creds",
"timeout_on_sleeping_server|service_account_creds|compute_engine_creds|"
"jwt_token_creds|oauth2_auth_token",
FLAGS_test_case.c_str());
ret = 1;
}

@ -40,6 +40,7 @@
#include <unistd.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <gflags/gflags.h>
#include <grpc++/channel_arguments.h>
@ -47,6 +48,8 @@
#include <grpc++/create_channel.h>
#include <grpc++/credentials.h>
#include <grpc++/stream.h>
#include "src/cpp/client/secure_credentials.h"
#include "test/core/security/oauth2_utils.h"
#include "test/cpp/util/create_test_channel.h"
DECLARE_bool(enable_ssl);
@ -62,6 +65,16 @@ DECLARE_string(oauth_scope);
namespace grpc {
namespace testing {
namespace {
std::shared_ptr<Credentials> CreateServiceAccountCredentials() {
GPR_ASSERT(FLAGS_enable_ssl);
grpc::string json_key = GetServiceAccountJsonKey();
std::chrono::seconds token_lifetime = std::chrono::hours(1);
return ServiceAccountCredentials(json_key, FLAGS_oauth_scope,
token_lifetime.count());
}
} // namespace
grpc::string GetServiceAccountJsonKey() {
static grpc::string json_key;
if (json_key.empty()) {
@ -73,6 +86,20 @@ grpc::string GetServiceAccountJsonKey() {
return json_key;
}
grpc::string GetOauth2AccessToken() {
std::shared_ptr<Credentials> creds = CreateServiceAccountCredentials();
SecureCredentials* secure_creds =
dynamic_cast<SecureCredentials*>(creds.get());
GPR_ASSERT(secure_creds != nullptr);
grpc_credentials* c_creds = secure_creds->GetRawCreds();
char* token = grpc_test_fetch_oauth2_token_with_credentials(c_creds);
GPR_ASSERT(token != nullptr);
gpr_log(GPR_INFO, "Get raw oauth2 access token: %s", token);
grpc::string access_token(token + sizeof("Bearer ") - 1);
gpr_free(token);
return access_token;
}
std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
const grpc::string& test_case) {
GPR_ASSERT(FLAGS_server_port);
@ -82,12 +109,7 @@ std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
FLAGS_server_port);
if (test_case == "service_account_creds") {
std::shared_ptr<Credentials> creds;
GPR_ASSERT(FLAGS_enable_ssl);
grpc::string json_key = GetServiceAccountJsonKey();
std::chrono::seconds token_lifetime = std::chrono::hours(1);
creds = ServiceAccountCredentials(json_key, FLAGS_oauth_scope,
token_lifetime.count());
std::shared_ptr<Credentials> creds = CreateServiceAccountCredentials();
return CreateTestChannel(host_port, FLAGS_server_host_override,
FLAGS_enable_ssl, FLAGS_use_prod_roots, creds);
} else if (test_case == "compute_engine_creds") {
@ -104,6 +126,11 @@ std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
creds = JWTCredentials(json_key, token_lifetime.count());
return CreateTestChannel(host_port, FLAGS_server_host_override,
FLAGS_enable_ssl, FLAGS_use_prod_roots, creds);
} else if (test_case == "oauth2_auth_token") {
grpc::string raw_token = GetOauth2AccessToken();
std::shared_ptr<Credentials> creds = AccessTokenCredentials(raw_token);
return CreateTestChannel(host_port, FLAGS_server_host_override,
FLAGS_enable_ssl, FLAGS_use_prod_roots, creds);
} else {
return CreateTestChannel(host_port, FLAGS_server_host_override,
FLAGS_enable_ssl, FLAGS_use_prod_roots);

@ -44,6 +44,8 @@ namespace testing {
grpc::string GetServiceAccountJsonKey();
grpc::string GetOauth2AccessToken();
std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
const grpc::string& test_case);

@ -143,6 +143,29 @@ void InteropClient::DoServiceAccountCreds(const grpc::string& username,
gpr_log(GPR_INFO, "Large unary with service account creds done.");
}
void InteropClient::DoOauth2AuthToken(const grpc::string& username,
const grpc::string& oauth_scope) {
gpr_log(GPR_INFO,
"Sending a unary rpc with raw oauth2 access token credentials ...");
SimpleRequest request;
SimpleResponse response;
request.set_fill_username(true);
request.set_fill_oauth_scope(true);
std::unique_ptr<TestService::Stub> stub(TestService::NewStub(channel_));
ClientContext context;
Status s = stub->UnaryCall(&context, request, &response);
AssertOkOrPrintErrorStatus(s);
GPR_ASSERT(!response.username().empty());
GPR_ASSERT(!response.oauth_scope().empty());
GPR_ASSERT(username.find(response.username()) != grpc::string::npos);
const char* oauth_scope_str = response.oauth_scope().c_str();
GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
gpr_log(GPR_INFO, "Unary with oauth2 access token credentials done.");
}
void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
gpr_log(GPR_INFO, "Sending a large unary rpc with JWT token credentials ...");
SimpleRequest request;

@ -68,6 +68,9 @@ class InteropClient {
// username is a string containing the user email
void DoServiceAccountCreds(const grpc::string& username,
const grpc::string& oauth_scope);
// username is a string containing the user email
void DoOauth2AuthToken(const grpc::string& username,
const grpc::string& oauth_scope);
private:
void PerformLargeUnary(SimpleRequest* request, SimpleResponse* response);

@ -789,6 +789,7 @@ src/core/channel/census_filter.h \
src/core/channel/channel_args.h \
src/core/channel/channel_stack.h \
src/core/channel/client_channel.h \
src/core/channel/compress_filter.h \
src/core/channel/connected_channel.h \
src/core/channel/context.h \
src/core/channel/http_client_filter.h \
@ -908,6 +909,7 @@ src/core/census/grpc_context.c \
src/core/channel/channel_args.c \
src/core/channel/channel_stack.c \
src/core/channel/client_channel.c \
src/core/channel/compress_filter.c \
src/core/channel/connected_channel.c \
src/core/channel/http_client_filter.c \
src/core/channel/http_server_filter.c \

@ -32,24 +32,66 @@
# linuxbrew installation of a selected language
set -ex
sha1=$(sha1sum tools/jenkins/grpc_linuxbrew/Dockerfile | cut -f1 -d\ )
DOCKER_IMAGE_NAME=grpc_linuxbrew_$sha1
if [ "$platform" == "linux" ]; then
docker build -t $DOCKER_IMAGE_NAME tools/jenkins/grpc_linuxbrew
if [ "$dist_channel" == "homebrew" ]; then
supported="python nodejs ruby php"
sha1=$(sha1sum tools/jenkins/grpc_linuxbrew/Dockerfile | cut -f1 -d\ )
DOCKER_IMAGE_NAME=grpc_linuxbrew_$sha1
docker build -t $DOCKER_IMAGE_NAME tools/jenkins/grpc_linuxbrew
supported="python nodejs ruby php"
if [ "$language" == "core" ]; then
command="curl -fsSL https://goo.gl/getgrpc | bash -"
elif [[ "$supported" =~ "$language" ]]; then
command="curl -fsSL https://goo.gl/getgrpc | bash -s $language"
else
echo "unsupported language $language"
exit 1
fi
docker run $DOCKER_IMAGE_NAME bash -l \
-c "nvm use 0.12; \
npm set unsafe-perm true; \
rvm use ruby-2.1; \
$command"
else
echo "Unsupported $platform dist_channel $dist_channel"
exit 1
fi
elif [ "$platform" == "macos" ]; then
if [ "$dist_channel" == "homebrew" ]; then
which brew # TODO: for debug, can be removed later
brew list -l
rm -rf /tmp/homebrew-test
mkdir -p /tmp/homebrew-test
git clone https://github.com/Homebrew/homebrew.git /tmp/homebrew-test
cd /tmp/homebrew-test
# TODO: Uncomment these when the general structure of the script is verified
# PATH=/tmp/homebrew-test/bin:$PATH brew tap homebrew/dupes
# PATH=/tmp/homebrew-test/bin:$PATH brew install zlib
# PATH=/tmp/homebrew-test/bin:$PATH brew install openssl
# PATH=/tmp/homebrew-test/bin:$PATH brew tap grpc/grpc
# PATH=/tmp/homebrew-test/bin:$PATH brew install --without-python google-protobuf
# PATH=/tmp/homebrew-test/bin:$PATH brew install grpc
PATH=/tmp/homebrew-test/bin:$PATH brew list -l
brew list -l
cd ~/
rm -rf /tmp/homebrew-test
echo $PATH # TODO: for debug, can be removed later
brew list -l # TODO: for debug, can be removed later
else
echo "Unsupported $platform dist_channel $dist_channel"
exit 1
fi
if [ "$language" == "core" ]; then
command="curl -fsSL https://goo.gl/getgrpc | bash -"
elif [[ "$supported" =~ "$language" ]]; then
command="curl -fsSL https://goo.gl/getgrpc | bash -s $language"
else
echo "unsupported language $language"
echo "unsupported platform $platform"
exit 1
fi
docker run $DOCKER_IMAGE_NAME bash -l \
-c "nvm use 0.12; \
npm set unsafe-perm true; \
rvm use ruby-2.1; \
$command"

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because one or more lines are too long

@ -178,6 +178,7 @@
<ClInclude Include="..\..\src\core\channel\channel_args.h" />
<ClInclude Include="..\..\src\core\channel\channel_stack.h" />
<ClInclude Include="..\..\src\core\channel\client_channel.h" />
<ClInclude Include="..\..\src\core\channel\compress_filter.h" />
<ClInclude Include="..\..\src\core\channel\connected_channel.h" />
<ClInclude Include="..\..\src\core\channel\context.h" />
<ClInclude Include="..\..\src\core\channel\http_client_filter.h" />
@ -327,6 +328,8 @@
</ClCompile>
<ClCompile Include="..\..\src\core\channel\client_channel.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\compress_filter.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\connected_channel.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\http_client_filter.c">

@ -85,6 +85,9 @@
<ClCompile Include="..\..\src\core\channel\client_channel.c">
<Filter>src\core\channel</Filter>
</ClCompile>
<ClCompile Include="..\..\src\core\channel\compress_filter.c">
<Filter>src\core\channel</Filter>
</ClCompile>
<ClCompile Include="..\..\src\core\channel\connected_channel.c">
<Filter>src\core\channel</Filter>
</ClCompile>
@ -494,6 +497,9 @@
<ClInclude Include="..\..\src\core\channel\client_channel.h">
<Filter>src\core\channel</Filter>
</ClInclude>
<ClInclude Include="..\..\src\core\channel\compress_filter.h">
<Filter>src\core\channel</Filter>
</ClInclude>
<ClInclude Include="..\..\src\core\channel\connected_channel.h">
<Filter>src\core\channel</Filter>
</ClInclude>

@ -149,6 +149,7 @@
<ClInclude Include="..\..\test\core\end2end\data\ssl_test_data.h" />
<ClInclude Include="..\..\test\core\end2end\cq_verifier.h" />
<ClInclude Include="..\..\test\core\iomgr\endpoint_tests.h" />
<ClInclude Include="..\..\test\core\security\oauth2_utils.h" />
<ClInclude Include="..\..\test\core\util\grpc_profiler.h" />
<ClInclude Include="..\..\test\core\util\parse_hexstring.h" />
<ClInclude Include="..\..\test\core\util\port.h" />
@ -165,6 +166,8 @@
</ClCompile>
<ClCompile Include="..\..\test\core\iomgr\endpoint_tests.c">
</ClCompile>
<ClCompile Include="..\..\test\core\security\oauth2_utils.c">
</ClCompile>
<ClCompile Include="..\..\test\core\util\grpc_profiler.c">
</ClCompile>
<ClCompile Include="..\..\test\core\util\parse_hexstring.c">

@ -159,6 +159,7 @@
<ClInclude Include="..\..\src\core\channel\channel_args.h" />
<ClInclude Include="..\..\src\core\channel\channel_stack.h" />
<ClInclude Include="..\..\src\core\channel\client_channel.h" />
<ClInclude Include="..\..\src\core\channel\compress_filter.h" />
<ClInclude Include="..\..\src\core\channel\connected_channel.h" />
<ClInclude Include="..\..\src\core\channel\context.h" />
<ClInclude Include="..\..\src\core\channel\http_client_filter.h" />
@ -262,6 +263,8 @@
</ClCompile>
<ClCompile Include="..\..\src\core\channel\client_channel.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\compress_filter.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\connected_channel.c">
</ClCompile>
<ClCompile Include="..\..\src\core\channel\http_client_filter.c">

@ -16,6 +16,9 @@
<ClCompile Include="..\..\src\core\channel\client_channel.c">
<Filter>src\core\channel</Filter>
</ClCompile>
<ClCompile Include="..\..\src\core\channel\compress_filter.c">
<Filter>src\core\channel</Filter>
</ClCompile>
<ClCompile Include="..\..\src\core\channel\connected_channel.c">
<Filter>src\core\channel</Filter>
</ClCompile>
@ -371,6 +374,9 @@
<ClInclude Include="..\..\src\core\channel\client_channel.h">
<Filter>src\core\channel</Filter>
</ClInclude>
<ClInclude Include="..\..\src\core\channel\compress_filter.h">
<Filter>src\core\channel</Filter>
</ClInclude>
<ClInclude Include="..\..\src\core\channel\connected_channel.h">
<Filter>src\core\channel</Filter>
</ClInclude>

Loading…
Cancel
Save