diff options
Diffstat (limited to 'test/core/end2end/fixtures/http_proxy_fixture.cc')
-rw-r--r-- | test/core/end2end/fixtures/http_proxy_fixture.cc | 177 |
1 files changed, 76 insertions, 101 deletions
diff --git a/test/core/end2end/fixtures/http_proxy_fixture.cc b/test/core/end2end/fixtures/http_proxy_fixture.cc index ac0c953a79..c49a866a6c 100644 --- a/test/core/end2end/fixtures/http_proxy_fixture.cc +++ b/test/core/end2end/fixtures/http_proxy_fixture.cc @@ -104,24 +104,21 @@ static void proxy_connection_ref(proxy_connection* conn, const char* reason) { } // Helper function to destroy the proxy connection. -static void proxy_connection_unref(grpc_exec_ctx* exec_ctx, - proxy_connection* conn, const char* reason) { +static void proxy_connection_unref(proxy_connection* conn, const char* reason) { if (gpr_unref(&conn->refcount)) { gpr_log(GPR_DEBUG, "endpoints: %p %p", conn->client_endpoint, conn->server_endpoint); - grpc_endpoint_destroy(exec_ctx, conn->client_endpoint); + grpc_endpoint_destroy(conn->client_endpoint); if (conn->server_endpoint != nullptr) { - grpc_endpoint_destroy(exec_ctx, conn->server_endpoint); + grpc_endpoint_destroy(conn->server_endpoint); } - grpc_pollset_set_destroy(exec_ctx, conn->pollset_set); - grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_read_buffer); - grpc_slice_buffer_destroy_internal(exec_ctx, - &conn->client_deferred_write_buffer); - grpc_slice_buffer_destroy_internal(exec_ctx, &conn->client_write_buffer); - grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_read_buffer); - grpc_slice_buffer_destroy_internal(exec_ctx, - &conn->server_deferred_write_buffer); - grpc_slice_buffer_destroy_internal(exec_ctx, &conn->server_write_buffer); + grpc_pollset_set_destroy(conn->pollset_set); + grpc_slice_buffer_destroy_internal(&conn->client_read_buffer); + grpc_slice_buffer_destroy_internal(&conn->client_deferred_write_buffer); + grpc_slice_buffer_destroy_internal(&conn->client_write_buffer); + grpc_slice_buffer_destroy_internal(&conn->server_read_buffer); + grpc_slice_buffer_destroy_internal(&conn->server_deferred_write_buffer); + grpc_slice_buffer_destroy_internal(&conn->server_write_buffer); grpc_http_parser_destroy(&conn->http_parser); grpc_http_request_destroy(&conn->http_request); gpr_unref(&conn->proxy->users); @@ -131,28 +128,24 @@ static void proxy_connection_unref(grpc_exec_ctx* exec_ctx, // Helper function to shut down the proxy connection. // Does NOT take ownership of a reference to error. -static void proxy_connection_failed(grpc_exec_ctx* exec_ctx, - proxy_connection* conn, bool is_client, +static void proxy_connection_failed(proxy_connection* conn, bool is_client, const char* prefix, grpc_error* error) { const char* msg = grpc_error_string(error); gpr_log(GPR_INFO, "%s: %s", prefix, msg); - grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint, - GRPC_ERROR_REF(error)); + grpc_endpoint_shutdown(conn->client_endpoint, GRPC_ERROR_REF(error)); if (conn->server_endpoint != nullptr) { - grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint, - GRPC_ERROR_REF(error)); + grpc_endpoint_shutdown(conn->server_endpoint, GRPC_ERROR_REF(error)); } - proxy_connection_unref(exec_ctx, conn, "conn_failed"); + proxy_connection_unref(conn, "conn_failed"); } // Callback for writing proxy data to the client. -static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_client_write_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; conn->client_is_writing = false; if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy client write", error); return; } @@ -164,22 +157,20 @@ static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice_buffer_move_into(&conn->client_deferred_write_buffer, &conn->client_write_buffer); conn->client_is_writing = true; - grpc_endpoint_write(exec_ctx, conn->client_endpoint, - &conn->client_write_buffer, + grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer, &conn->on_client_write_done); } else { // No more writes. Unref the connection. - proxy_connection_unref(exec_ctx, conn, "write_done"); + proxy_connection_unref(conn, "write_done"); } } // Callback for writing proxy data to the backend server. -static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_server_write_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; conn->server_is_writing = false; if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, false /* is_client */, + proxy_connection_failed(conn, false /* is_client */, "HTTP proxy server write", error); return; } @@ -191,22 +182,20 @@ static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice_buffer_move_into(&conn->server_deferred_write_buffer, &conn->server_write_buffer); conn->server_is_writing = true; - grpc_endpoint_write(exec_ctx, conn->server_endpoint, - &conn->server_write_buffer, + grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer, &conn->on_server_write_done); } else { // No more writes. Unref the connection. - proxy_connection_unref(exec_ctx, conn, "server_write"); + proxy_connection_unref(conn, "server_write"); } } // Callback for reading data from the client, which will be proxied to // the backend server. -static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_client_read_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy client read", error); return; } @@ -224,22 +213,20 @@ static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg, &conn->server_write_buffer); proxy_connection_ref(conn, "client_read"); conn->server_is_writing = true; - grpc_endpoint_write(exec_ctx, conn->server_endpoint, - &conn->server_write_buffer, + grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer, &conn->on_server_write_done); } // Read more data. - grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer, + grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer, &conn->on_client_read_done); } // Callback for reading data from the backend server, which will be // proxied to the client. -static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_server_read_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, false /* is_client */, + proxy_connection_failed(conn, false /* is_client */, "HTTP proxy server read", error); return; } @@ -257,22 +244,20 @@ static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg, &conn->client_write_buffer); proxy_connection_ref(conn, "server_read"); conn->client_is_writing = true; - grpc_endpoint_write(exec_ctx, conn->client_endpoint, - &conn->client_write_buffer, + grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer, &conn->on_client_write_done); } // Read more data. - grpc_endpoint_read(exec_ctx, conn->server_endpoint, &conn->server_read_buffer, + grpc_endpoint_read(conn->server_endpoint, &conn->server_read_buffer, &conn->on_server_read_done); } // Callback to write the HTTP response for the CONNECT request. -static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_write_response_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; conn->client_is_writing = false; if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy write response", error); return; } @@ -283,17 +268,16 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg, // for the other one. proxy_connection_ref(conn, "client_read"); proxy_connection_ref(conn, "server_read"); - proxy_connection_unref(exec_ctx, conn, "write_response"); - grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer, + proxy_connection_unref(conn, "write_response"); + grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer, &conn->on_client_read_done); - grpc_endpoint_read(exec_ctx, conn->server_endpoint, &conn->server_read_buffer, + grpc_endpoint_read(conn->server_endpoint, &conn->server_read_buffer, &conn->on_server_read_done); } // Callback to connect to the backend server specified by the HTTP // CONNECT request. -static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_server_connect_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; if (error != GRPC_ERROR_NONE) { // TODO(roth): Technically, in this case, we should handle the error @@ -301,7 +285,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg, // connection failed. However, for the purposes of this test code, // it's fine to pretend this is a client-side error, which will // cause the client connection to be dropped. - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy server connect", error); return; } @@ -312,8 +296,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice_from_copied_string("HTTP/1.0 200 connected\r\n\r\n"); grpc_slice_buffer_add(&conn->client_write_buffer, slice); conn->client_is_writing = true; - grpc_endpoint_write(exec_ctx, conn->client_endpoint, - &conn->client_write_buffer, + grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer, &conn->on_write_response_done); } @@ -322,8 +305,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg, * Basic <base64_encoded_expected_cred> * Returns true if it matches, false otherwise */ -static bool proxy_auth_header_matches(grpc_exec_ctx* exec_ctx, - char* proxy_auth_header_val, +static bool proxy_auth_header_matches(char* proxy_auth_header_val, char* expected_cred) { GPR_ASSERT(proxy_auth_header_val != nullptr); GPR_ASSERT(expected_cred != nullptr); @@ -331,11 +313,10 @@ static bool proxy_auth_header_matches(grpc_exec_ctx* exec_ctx, return false; } proxy_auth_header_val += 6; - grpc_slice decoded_slice = - grpc_base64_decode(exec_ctx, proxy_auth_header_val, 0); + grpc_slice decoded_slice = grpc_base64_decode(proxy_auth_header_val, 0); const bool header_matches = grpc_slice_str_cmp(decoded_slice, expected_cred) == 0; - grpc_slice_unref_internal(exec_ctx, decoded_slice); + grpc_slice_unref_internal(decoded_slice); return header_matches; } @@ -345,13 +326,12 @@ static bool proxy_auth_header_matches(grpc_exec_ctx* exec_ctx, // the client indicating that the request failed. However, for the purposes // of this test code, it's fine to pretend this is a client-side error, // which will cause the client connection to be dropped. -static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void on_read_request_done(void* arg, grpc_error* error) { proxy_connection* conn = (proxy_connection*)arg; gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn, grpc_error_string(error)); if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy read request", error); return; } @@ -361,7 +341,7 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, error = grpc_http_parser_parse( &conn->http_parser, conn->client_read_buffer.slices[i], nullptr); if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy request parse", error); GRPC_ERROR_UNREF(error); return; @@ -371,8 +351,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice_buffer_reset_and_unref(&conn->client_read_buffer); // If we're not done reading the request, read more data. if (conn->http_parser.state != GRPC_HTTP_BODY) { - grpc_endpoint_read(exec_ctx, conn->client_endpoint, - &conn->client_read_buffer, &conn->on_read_request_done); + grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer, + &conn->on_read_request_done); return; } // Make sure we got a CONNECT request. @@ -382,7 +362,7 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, conn->http_request.method); error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy read request", error); GRPC_ERROR_UNREF(error); return; @@ -395,15 +375,14 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, for (size_t i = 0; i < conn->http_request.hdr_count; i++) { if (strcmp(conn->http_request.hdrs[i].key, "Proxy-Authorization") == 0) { client_authenticated = proxy_auth_header_matches( - exec_ctx, conn->http_request.hdrs[i].value, - proxy_auth_arg->value.string); + conn->http_request.hdrs[i].value, proxy_auth_arg->value.string); break; } } if (!client_authenticated) { const char* msg = "HTTP Connect could not verify authentication"; error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(msg); - proxy_connection_failed(exec_ctx, conn, true /* is_client */, + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy read request", error); GRPC_ERROR_UNREF(error); return; @@ -414,8 +393,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, error = grpc_blocking_resolve_address(conn->http_request.path, "80", &resolved_addresses); if (error != GRPC_ERROR_NONE) { - proxy_connection_failed(exec_ctx, conn, true /* is_client */, - "HTTP proxy DNS lookup", error); + proxy_connection_failed(conn, true /* is_client */, "HTTP proxy DNS lookup", + error); GRPC_ERROR_UNREF(error); return; } @@ -423,15 +402,15 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, // Connect to requested address. // The connection callback inherits our reference to conn. const grpc_millis deadline = - grpc_exec_ctx_now(exec_ctx) + 10 * GPR_MS_PER_SEC; - grpc_tcp_client_connect(exec_ctx, &conn->on_server_connect_done, - &conn->server_endpoint, conn->pollset_set, nullptr, + grpc_core::ExecCtx::Get()->Now() + 10 * GPR_MS_PER_SEC; + grpc_tcp_client_connect(&conn->on_server_connect_done, &conn->server_endpoint, + conn->pollset_set, nullptr, &resolved_addresses->addrs[0], deadline); grpc_resolved_addresses_destroy(resolved_addresses); } -static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, - grpc_endpoint* endpoint, grpc_pollset* accepting_pollset, +static void on_accept(void* arg, grpc_endpoint* endpoint, + grpc_pollset* accepting_pollset, grpc_tcp_server_acceptor* acceptor) { gpr_free(acceptor); grpc_end2end_http_proxy* proxy = (grpc_end2end_http_proxy*)arg; @@ -442,8 +421,8 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, conn->proxy = proxy; gpr_ref_init(&conn->refcount, 1); conn->pollset_set = grpc_pollset_set_create(); - grpc_pollset_set_add_pollset(exec_ctx, conn->pollset_set, proxy->pollset); - grpc_endpoint_add_to_pollset_set(exec_ctx, endpoint, conn->pollset_set); + grpc_pollset_set_add_pollset(conn->pollset_set, proxy->pollset); + grpc_endpoint_add_to_pollset_set(endpoint, conn->pollset_set); GRPC_CLOSURE_INIT(&conn->on_read_request_done, on_read_request_done, conn, grpc_combiner_scheduler(conn->proxy->combiner)); GRPC_CLOSURE_INIT(&conn->on_server_connect_done, on_server_connect_done, conn, @@ -468,7 +447,7 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, grpc_slice_buffer_init(&conn->server_write_buffer); grpc_http_parser_init(&conn->http_parser, GRPC_HTTP_REQUEST, &conn->http_request); - grpc_endpoint_read(exec_ctx, conn->client_endpoint, &conn->client_read_buffer, + grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer, &conn->on_read_request_done); } @@ -478,24 +457,23 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, static void thread_main(void* arg) { grpc_end2end_http_proxy* proxy = (grpc_end2end_http_proxy*)arg; - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_core::ExecCtx _local_exec_ctx; do { gpr_ref(&proxy->users); grpc_pollset_worker* worker = nullptr; gpr_mu_lock(proxy->mu); GRPC_LOG_IF_ERROR( "grpc_pollset_work", - grpc_pollset_work(&exec_ctx, proxy->pollset, &worker, - grpc_exec_ctx_now(&exec_ctx) + GPR_MS_PER_SEC)); + grpc_pollset_work(proxy->pollset, &worker, + grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC)); gpr_mu_unlock(proxy->mu); - grpc_exec_ctx_flush(&exec_ctx); + grpc_core::ExecCtx::Get()->Flush(); } while (!gpr_unref(&proxy->users)); - grpc_exec_ctx_finish(&exec_ctx); } grpc_end2end_http_proxy* grpc_end2end_http_proxy_create( grpc_channel_args* args) { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_core::ExecCtx _local_exec_ctx; grpc_end2end_http_proxy* proxy = (grpc_end2end_http_proxy*)gpr_malloc(sizeof(*proxy)); memset(proxy, 0, sizeof(*proxy)); @@ -507,8 +485,8 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create( gpr_log(GPR_INFO, "Proxy address: %s", proxy->proxy_name); // Create TCP server. proxy->channel_args = grpc_channel_args_copy(args); - grpc_error* error = grpc_tcp_server_create( - &exec_ctx, nullptr, proxy->channel_args, &proxy->server); + grpc_error* error = + grpc_tcp_server_create(nullptr, proxy->channel_args, &proxy->server); GPR_ASSERT(error == GRPC_ERROR_NONE); // Bind to port. grpc_resolved_address resolved_addr; @@ -523,9 +501,8 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create( // Start server. proxy->pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size()); grpc_pollset_init(proxy->pollset, &proxy->mu); - grpc_tcp_server_start(&exec_ctx, proxy->server, &proxy->pollset, 1, on_accept, - proxy); - grpc_exec_ctx_finish(&exec_ctx); + grpc_tcp_server_start(proxy->server, &proxy->pollset, 1, on_accept, proxy); + // Start proxy thread. gpr_thd_options opt = gpr_thd_options_default(); gpr_thd_options_set_joinable(&opt); @@ -533,27 +510,25 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create( return proxy; } -static void destroy_pollset(grpc_exec_ctx* exec_ctx, void* arg, - grpc_error* error) { +static void destroy_pollset(void* arg, grpc_error* error) { grpc_pollset* pollset = (grpc_pollset*)arg; - grpc_pollset_destroy(exec_ctx, pollset); + grpc_pollset_destroy(pollset); gpr_free(pollset); } void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) { gpr_unref(&proxy->users); // Signal proxy thread to shutdown. - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_core::ExecCtx _local_exec_ctx; gpr_thd_join(proxy->thd); - grpc_tcp_server_shutdown_listeners(&exec_ctx, proxy->server); - grpc_tcp_server_unref(&exec_ctx, proxy->server); + grpc_tcp_server_shutdown_listeners(proxy->server); + grpc_tcp_server_unref(proxy->server); gpr_free(proxy->proxy_name); - grpc_channel_args_destroy(&exec_ctx, proxy->channel_args); - grpc_pollset_shutdown(&exec_ctx, proxy->pollset, + grpc_channel_args_destroy(proxy->channel_args); + grpc_pollset_shutdown(proxy->pollset, GRPC_CLOSURE_CREATE(destroy_pollset, proxy->pollset, grpc_schedule_on_exec_ctx)); - GRPC_COMBINER_UNREF(&exec_ctx, proxy->combiner, "test"); + GRPC_COMBINER_UNREF(proxy->combiner, "test"); gpr_free(proxy); - grpc_exec_ctx_finish(&exec_ctx); } const char* grpc_end2end_http_proxy_get_proxy_name( |