diff options
Diffstat (limited to 'src/core/lib/surface/channel.c')
-rw-r--r-- | src/core/lib/surface/channel.c | 69 |
1 files changed, 59 insertions, 10 deletions
diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index b6b760b5d8..8936e7164f 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -36,16 +36,17 @@ #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/lib/channel/channel_args.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/surface/call.h" #include "src/core/lib/surface/channel_init.h" -#include "src/core/lib/surface/init.h" #include "src/core/lib/transport/static_metadata.h" /** Cache grpc-status: X mdelems for X = 0..NUM_CACHED_STATUS_ELEMS. @@ -64,10 +65,12 @@ typedef struct registered_call { struct grpc_channel { int is_client; uint32_t max_message_length; + grpc_compression_options compression_options; grpc_mdelem *default_authority; gpr_mu registered_call_mu; registered_call *registered_calls; + char *target; }; @@ -80,7 +83,8 @@ struct grpc_channel { /* the protobuf library will (by default) start warning at 100megs */ #define DEFAULT_MAX_MESSAGE_LENGTH (100 * 1024 * 1024) -static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, bool success); +static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error); grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, const grpc_channel_args *input_args, @@ -111,6 +115,7 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, channel->registered_calls = NULL; channel->max_message_length = DEFAULT_MAX_MESSAGE_LENGTH; + grpc_compression_options_init(&channel->compression_options); if (args) { for (size_t i = 0; i < args->num_args; i++) { if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_MESSAGE_LENGTH)) { @@ -151,6 +156,27 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, ":authority", args->args[i].value.string); } } + } else if (0 == strcmp(args->args[i].key, + GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL)) { + channel->compression_options.default_level.is_set = true; + GPR_ASSERT(args->args[i].value.integer >= 0 && + args->args[i].value.integer < GRPC_COMPRESS_LEVEL_COUNT); + channel->compression_options.default_level.level = + (grpc_compression_level)args->args[i].value.integer; + } else if (0 == strcmp(args->args[i].key, + GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM)) { + channel->compression_options.default_algorithm.is_set = true; + GPR_ASSERT(args->args[i].value.integer >= 0 && + args->args[i].value.integer < + GRPC_COMPRESS_ALGORITHMS_COUNT); + channel->compression_options.default_algorithm.algorithm = + (grpc_compression_algorithm)args->args[i].value.integer; + } else if (0 == + strcmp(args->args[i].key, + GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET)) { + channel->compression_options.enabled_algorithms_bitset = + (uint32_t)args->args[i].value.integer | + 0x1; /* always support no compression */ } } grpc_channel_args_destroy(args); @@ -166,12 +192,14 @@ char *grpc_channel_get_target(grpc_channel *channel) { static grpc_call *grpc_channel_create_call_internal( grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, - grpc_completion_queue *cq, grpc_mdelem *path_mdelem, - grpc_mdelem *authority_mdelem, gpr_timespec deadline) { + grpc_completion_queue *cq, grpc_pollset_set *pollset_set_alternative, + grpc_mdelem *path_mdelem, grpc_mdelem *authority_mdelem, + gpr_timespec deadline) { grpc_mdelem *send_metadata[2]; size_t num_metadata = 0; GPR_ASSERT(channel->is_client); + GPR_ASSERT(!(cq != NULL && pollset_set_alternative != NULL)); send_metadata[num_metadata++] = path_mdelem; if (authority_mdelem != NULL) { @@ -180,8 +208,9 @@ static grpc_call *grpc_channel_create_call_internal( send_metadata[num_metadata++] = GRPC_MDELEM_REF(channel->default_authority); } - return grpc_call_create(channel, parent_call, propagation_mask, cq, NULL, - send_metadata, num_metadata, deadline); + return grpc_call_create(channel, parent_call, propagation_mask, cq, + pollset_set_alternative, NULL, send_metadata, + num_metadata, deadline); } grpc_call *grpc_channel_create_call(grpc_channel *channel, @@ -201,7 +230,22 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); return grpc_channel_create_call_internal( - channel, parent_call, propagation_mask, cq, + channel, parent_call, propagation_mask, cq, NULL, + grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + grpc_mdstr_from_string(method)), + host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY, + grpc_mdstr_from_string(host)) + : NULL, + deadline); +} + +grpc_call *grpc_channel_create_pollset_set_call( + grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, + grpc_pollset_set *pollset_set, const char *method, const char *host, + gpr_timespec deadline, void *reserved) { + GPR_ASSERT(!reserved); + return grpc_channel_create_call_internal( + channel, parent_call, propagation_mask, NULL, pollset_set, grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, grpc_mdstr_from_string(method)), host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY, @@ -245,7 +289,7 @@ grpc_call *grpc_channel_create_registered_call( (int)deadline.tv_nsec, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); return grpc_channel_create_call_internal( - channel, parent_call, propagation_mask, completion_queue, + channel, parent_call, propagation_mask, completion_queue, NULL, GRPC_MDELEM_REF(rc->path), rc->authority ? GRPC_MDELEM_REF(rc->authority) : NULL, deadline); } @@ -267,7 +311,7 @@ void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, } static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, - bool iomgr_success) { + grpc_error *error) { grpc_channel *channel = arg; grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CHANNEL(channel)); while (channel->registered_calls) { @@ -293,7 +337,7 @@ void grpc_channel_destroy(grpc_channel *channel) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; GRPC_API_TRACE("grpc_channel_destroy(channel=%p)", 1, (channel)); memset(&op, 0, sizeof(op)); - op.disconnect = 1; + op.disconnect_with_error = GRPC_ERROR_CREATE("Channel Destroyed"); elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CHANNEL(channel), 0); elem->filter->start_transport_op(&exec_ctx, elem, &op); @@ -306,6 +350,11 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) { return CHANNEL_STACK_FROM_CHANNEL(channel); } +grpc_compression_options grpc_channel_compression_options( + const grpc_channel *channel) { + return channel->compression_options; +} + grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) { char tmp[GPR_LTOA_MIN_BUFSIZE]; switch (i) { |