diff options
Diffstat (limited to 'src/core/surface/secure_channel_create.c')
| -rw-r--r-- | src/core/surface/secure_channel_create.c | 254 |
1 files changed, 118 insertions, 136 deletions
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index cfa869ec71..5f77e4c06e 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -31,178 +31,148 @@ * */ -#include "src/core/iomgr/sockaddr.h" - #include <grpc/grpc.h> #include <stdlib.h> #include <string.h> -#include "src/core/channel/census_filter.h" +#include <grpc/support/alloc.h> + #include "src/core/channel/channel_args.h" #include "src/core/channel/client_channel.h" -#include "src/core/channel/client_setup.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/iomgr/resolve_address.h" +#include "src/core/client_config/resolver_registry.h" #include "src/core/iomgr/tcp_client.h" #include "src/core/security/auth_filters.h" #include "src/core/security/credentials.h" #include "src/core/security/secure_transport_setup.h" -#include "src/core/support/string.h" #include "src/core/surface/channel.h" -#include "src/core/surface/client.h" #include "src/core/transport/chttp2_transport.h" -#include <grpc/grpc_security.h> -#include <grpc/support/alloc.h> -#include <grpc/support/log.h> -#include <grpc/support/string_util.h> -#include <grpc/support/sync.h> -#include <grpc/support/useful.h> #include "src/core/tsi/transport_security_interface.h" -typedef struct setup setup; - -/* A single setup request (started via initiate) */ typedef struct { - grpc_client_setup_request *cs_request; - setup *setup; - /* Resolved addresses, or null if resolution not yet completed. */ - grpc_resolved_addresses *resolved; - /* which address in resolved should we pick for the next connection attempt */ - size_t resolved_index; -} request; + grpc_connector base; + gpr_refcount refs; -struct setup { grpc_channel_security_connector *security_connector; - const char *target; - grpc_transport_setup_callback setup_callback; - void *setup_user_data; -}; -static int maybe_try_next_resolved(request *r); + grpc_iomgr_closure *notify; + grpc_connect_in_args args; + grpc_connect_out_args *result; +} connector; -static void done(request *r, int was_successful) { - grpc_client_setup_request_finish(r->cs_request, was_successful); - if (r->resolved) { - grpc_resolved_addresses_destroy(r->resolved); +static void connector_ref(grpc_connector *con) { + connector *c = (connector *)con; + gpr_ref(&c->refs); +} + +static void connector_unref(grpc_connector *con) { + connector *c = (connector *)con; + if (gpr_unref(&c->refs)) { + gpr_free(c); } - gpr_free(r); } -static void on_secure_transport_setup_done(void *rp, +static void on_secure_transport_setup_done(void *arg, grpc_security_status status, grpc_endpoint *secure_endpoint) { - request *r = rp; + connector *c = arg; + grpc_iomgr_closure *notify; if (status != GRPC_SECURITY_OK) { gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status); - done(r, 0); - } else if (grpc_client_setup_cb_begin(r->cs_request, - "on_secure_transport_setup_done")) { - grpc_create_chttp2_transport( - r->setup->setup_callback, r->setup->setup_user_data, - grpc_client_setup_get_channel_args(r->cs_request), secure_endpoint, - NULL, 0, grpc_client_setup_get_mdctx(r->cs_request), 1); - grpc_client_setup_cb_end(r->cs_request, "on_secure_transport_setup_done"); - done(r, 1); + memset(c->result, 0, sizeof(*c->result)); } else { - done(r, 0); + c->result->transport = grpc_create_chttp2_transport( + c->args.channel_args, secure_endpoint, c->args.metadata_context, 1); + grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0); + c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2); + c->result->filters[0] = &grpc_client_auth_filter; + c->result->filters[1] = &grpc_http_client_filter; + c->result->num_filters = 2; } + notify = c->notify; + c->notify = NULL; + grpc_iomgr_add_callback(notify); } -/* connection callback: tcp is either valid, or null on error */ -static void on_connect(void *rp, grpc_endpoint *tcp) { - request *r = rp; - - if (!grpc_client_setup_request_should_continue(r->cs_request, - "on_connect.secure")) { - if (tcp) { - grpc_endpoint_shutdown(tcp); - grpc_endpoint_destroy(tcp); - } - done(r, 0); - return; - } - - if (!tcp) { - if (!maybe_try_next_resolved(r)) { - done(r, 0); - return; - } else { - return; - } +static void connected(void *arg, grpc_endpoint *tcp) { + connector *c = arg; + grpc_iomgr_closure *notify; + if (tcp != NULL) { + grpc_setup_secure_transport(&c->security_connector->base, tcp, + on_secure_transport_setup_done, c); } else { - grpc_setup_secure_transport(&r->setup->security_connector->base, tcp, - on_secure_transport_setup_done, r); + memset(c->result, 0, sizeof(*c->result)); + notify = c->notify; + c->notify = NULL; + grpc_iomgr_add_callback(notify); } } -/* attempt to connect to the next available resolved address */ -static int maybe_try_next_resolved(request *r) { - grpc_resolved_address *addr; - if (!r->resolved) return 0; - if (r->resolved_index == r->resolved->naddrs) return 0; - addr = &r->resolved->addrs[r->resolved_index++]; - grpc_tcp_client_connect( - on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request), - (struct sockaddr *)&addr->addr, addr->len, - grpc_client_setup_request_deadline(r->cs_request)); - return 1; +static void connector_connect(grpc_connector *con, + const grpc_connect_in_args *args, + grpc_connect_out_args *result, + grpc_iomgr_closure *notify) { + connector *c = (connector *)con; + GPR_ASSERT(c->notify == NULL); + GPR_ASSERT(notify->cb); + c->notify = notify; + c->args = *args; + c->result = result; + grpc_tcp_client_connect(connected, c, args->interested_parties, args->addr, + args->addr_len, args->deadline); } -/* callback for when our target address has been resolved */ -static void on_resolved(void *rp, grpc_resolved_addresses *resolved) { - request *r = rp; +static const grpc_connector_vtable connector_vtable = { + connector_ref, connector_unref, connector_connect}; - /* if we're not still the active request, abort */ - if (!grpc_client_setup_request_should_continue(r->cs_request, - "on_resolved.secure")) { - if (resolved) { - grpc_resolved_addresses_destroy(resolved); - } - done(r, 0); - return; - } +typedef struct { + grpc_subchannel_factory base; + gpr_refcount refs; + grpc_mdctx *mdctx; + grpc_channel_args *merge_args; + grpc_channel_security_connector *security_connector; +} subchannel_factory; - if (!resolved) { - done(r, 0); - return; - } else { - r->resolved = resolved; - r->resolved_index = 0; - if (!maybe_try_next_resolved(r)) { - done(r, 0); - } - } +static void subchannel_factory_ref(grpc_subchannel_factory *scf) { + subchannel_factory *f = (subchannel_factory *)scf; + gpr_ref(&f->refs); } -static void initiate_setup(void *sp, grpc_client_setup_request *cs_request) { - request *r = gpr_malloc(sizeof(request)); - r->setup = sp; - r->cs_request = cs_request; - r->resolved = NULL; - r->resolved_index = 0; - /* TODO(klempner): Make grpc_resolve_address respect deadline */ - grpc_resolve_address(r->setup->target, "https", on_resolved, r); +static void subchannel_factory_unref(grpc_subchannel_factory *scf) { + subchannel_factory *f = (subchannel_factory *)scf; + if (gpr_unref(&f->refs)) { + GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base, + "subchannel_factory"); + grpc_channel_args_destroy(f->merge_args); + grpc_mdctx_unref(f->mdctx); + gpr_free(f); + } } -static void done_setup(void *sp) { - setup *s = sp; - gpr_free((void *)s->target); - grpc_security_connector_unref(&s->security_connector->base); - gpr_free(s); +static grpc_subchannel *subchannel_factory_create_subchannel( + grpc_subchannel_factory *scf, grpc_subchannel_args *args) { + subchannel_factory *f = (subchannel_factory *)scf; + connector *c = gpr_malloc(sizeof(*c)); + grpc_channel_args *final_args = + grpc_channel_args_merge(args->args, f->merge_args); + grpc_subchannel *s; + memset(c, 0, sizeof(*c)); + c->base.vtable = &connector_vtable; + c->security_connector = f->security_connector; + gpr_ref_init(&c->refs, 1); + args->mdctx = f->mdctx; + args->args = final_args; + s = grpc_subchannel_create(&c->base, args); + grpc_connector_unref(&c->base); + grpc_channel_args_destroy(final_args); + return s; } -static grpc_transport_setup_result complete_setup(void *channel_stack, - grpc_transport *transport, - grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = { - &grpc_client_auth_filter, &grpc_http_client_filter}; - return grpc_client_channel_transport_setup_complete( - channel_stack, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), - mdctx); -} +static const grpc_subchannel_factory_vtable subchannel_factory_vtable = { + subchannel_factory_ref, subchannel_factory_unref, + subchannel_factory_create_subchannel}; /* Create a secure client channel: Asynchronously: - resolve target @@ -211,13 +181,14 @@ static grpc_transport_setup_result complete_setup(void *channel_stack, grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, const char *target, const grpc_channel_args *args) { - setup *s; grpc_channel *channel; grpc_arg connector_arg; grpc_channel_args *args_copy; grpc_channel_args *new_args_from_connector; grpc_channel_security_connector *connector; grpc_mdctx *mdctx; + grpc_resolver *resolver; + subchannel_factory *f; #define MAX_FILTERS 3 const grpc_channel_filter *filters[MAX_FILTERS]; int n = 0; @@ -234,12 +205,10 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, } mdctx = grpc_mdctx_create(); - s = gpr_malloc(sizeof(setup)); connector_arg = grpc_security_connector_to_arg(&connector->base); args_copy = grpc_channel_args_copy_and_add( new_args_from_connector != NULL ? new_args_from_connector : args, - &connector_arg); - filters[n++] = &grpc_client_surface_filter; + &connector_arg, 1); /* TODO(census) if (grpc_channel_args_is_census_enabled(args)) { filters[n++] = &grpc_client_census_filter; @@ -247,18 +216,31 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, filters[n++] = &grpc_compress_filter; filters[n++] = &grpc_client_channel_filter; GPR_ASSERT(n <= MAX_FILTERS); + + f = gpr_malloc(sizeof(*f)); + f->base.vtable = &subchannel_factory_vtable; + gpr_ref_init(&f->refs, 1); + grpc_mdctx_ref(mdctx); + f->mdctx = mdctx; + GRPC_SECURITY_CONNECTOR_REF(&connector->base, "subchannel_factory"); + f->security_connector = connector; + f->merge_args = grpc_channel_args_copy(args_copy); + resolver = grpc_resolver_create(target, &f->base); + if (!resolver) { + return NULL; + } + channel = grpc_channel_create_from_filters(filters, n, args_copy, mdctx, 1); + grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel), + resolver); + GRPC_RESOLVER_UNREF(resolver, "create"); + grpc_subchannel_factory_unref(&f->base); + GRPC_SECURITY_CONNECTOR_UNREF(&connector->base, "channel_create"); + grpc_channel_args_destroy(args_copy); if (new_args_from_connector != NULL) { grpc_channel_args_destroy(new_args_from_connector); } - s->target = gpr_strdup(target); - s->setup_callback = complete_setup; - s->setup_user_data = grpc_channel_get_channel_stack(channel); - s->security_connector = connector; - grpc_client_setup_create_and_attach(grpc_channel_get_channel_stack(channel), - args, mdctx, initiate_setup, done_setup, - s); return channel; } |