| /* |
| * |
| * 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/fixtures/proxy.h" |
| |
| #include <string.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" |
| |
| struct grpc_end2end_proxy { |
| gpr_thd_id thd; |
| char *proxy_port; |
| char *server_port; |
| grpc_completion_queue *cq; |
| grpc_server *server; |
| grpc_channel *client; |
| |
| int shutdown; |
| |
| /* requested call */ |
| grpc_call *new_call; |
| grpc_call_details new_call_details; |
| grpc_metadata_array new_call_metadata; |
| }; |
| |
| typedef struct { |
| void (*func)(void *arg, int success); |
| void *arg; |
| } closure; |
| |
| typedef struct { |
| gpr_refcount refs; |
| grpc_end2end_proxy *proxy; |
| |
| grpc_call *c2p; |
| grpc_call *p2s; |
| |
| grpc_metadata_array c2p_initial_metadata; |
| grpc_metadata_array p2s_initial_metadata; |
| |
| grpc_byte_buffer *c2p_msg; |
| grpc_byte_buffer *p2s_msg; |
| |
| grpc_metadata_array p2s_trailing_metadata; |
| grpc_status_code p2s_status; |
| char *p2s_status_details; |
| size_t p2s_status_details_capacity; |
| |
| int c2p_server_cancelled; |
| } proxy_call; |
| |
| static void thread_main(void *arg); |
| static void request_call(grpc_end2end_proxy *proxy); |
| |
| grpc_end2end_proxy *grpc_end2end_proxy_create(const grpc_end2end_proxy_def *def, |
| grpc_channel_args *client_args, |
| grpc_channel_args *server_args) { |
| gpr_thd_options opt = gpr_thd_options_default(); |
| int proxy_port = grpc_pick_unused_port_or_die(); |
| int server_port = grpc_pick_unused_port_or_die(); |
| |
| grpc_end2end_proxy *proxy = gpr_malloc(sizeof(*proxy)); |
| memset(proxy, 0, sizeof(*proxy)); |
| |
| gpr_join_host_port(&proxy->proxy_port, "localhost", proxy_port); |
| gpr_join_host_port(&proxy->server_port, "localhost", server_port); |
| |
| gpr_log(GPR_DEBUG, "PROXY ADDR:%s BACKEND:%s", proxy->proxy_port, |
| proxy->server_port); |
| |
| proxy->cq = grpc_completion_queue_create(NULL); |
| proxy->server = def->create_server(proxy->proxy_port, server_args); |
| proxy->client = def->create_client(proxy->server_port, client_args); |
| |
| grpc_server_register_completion_queue(proxy->server, proxy->cq, NULL); |
| grpc_server_start(proxy->server); |
| |
| gpr_thd_options_set_joinable(&opt); |
| GPR_ASSERT(gpr_thd_new(&proxy->thd, thread_main, proxy, &opt)); |
| |
| request_call(proxy); |
| |
| return proxy; |
| } |
| |
| static closure *new_closure(void (*func)(void *arg, int success), void *arg) { |
| closure *cl = gpr_malloc(sizeof(*cl)); |
| cl->func = func; |
| cl->arg = arg; |
| return cl; |
| } |
| |
| static void shutdown_complete(void *arg, int success) { |
| grpc_end2end_proxy *proxy = arg; |
| proxy->shutdown = 1; |
| grpc_completion_queue_shutdown(proxy->cq); |
| } |
| |
| void grpc_end2end_proxy_destroy(grpc_end2end_proxy *proxy) { |
| grpc_server_shutdown_and_notify(proxy->server, proxy->cq, |
| new_closure(shutdown_complete, proxy)); |
| gpr_thd_join(proxy->thd); |
| gpr_free(proxy->proxy_port); |
| gpr_free(proxy->server_port); |
| grpc_server_destroy(proxy->server); |
| grpc_channel_destroy(proxy->client); |
| grpc_completion_queue_destroy(proxy->cq); |
| grpc_call_details_destroy(&proxy->new_call_details); |
| gpr_free(proxy); |
| } |
| |
| static void unrefpc(proxy_call *pc, const char *reason) { |
| if (gpr_unref(&pc->refs)) { |
| grpc_call_destroy(pc->c2p); |
| grpc_call_destroy(pc->p2s); |
| grpc_metadata_array_destroy(&pc->c2p_initial_metadata); |
| grpc_metadata_array_destroy(&pc->p2s_initial_metadata); |
| grpc_metadata_array_destroy(&pc->p2s_trailing_metadata); |
| gpr_free(pc->p2s_status_details); |
| gpr_free(pc); |
| } |
| } |
| |
| static void refpc(proxy_call *pc, const char *reason) { gpr_ref(&pc->refs); } |
| |
| static void on_c2p_sent_initial_metadata(void *arg, int success) { |
| proxy_call *pc = arg; |
| unrefpc(pc, "on_c2p_sent_initial_metadata"); |
| } |
| |
| static void on_p2s_recv_initial_metadata(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| if (!pc->proxy->shutdown) { |
| op.op = GRPC_OP_SEND_INITIAL_METADATA; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.send_initial_metadata.count = pc->p2s_initial_metadata.count; |
| op.data.send_initial_metadata.metadata = pc->p2s_initial_metadata.metadata; |
| refpc(pc, "on_c2p_sent_initial_metadata"); |
| err = grpc_call_start_batch( |
| pc->c2p, &op, 1, new_closure(on_c2p_sent_initial_metadata, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| |
| unrefpc(pc, "on_p2s_recv_initial_metadata"); |
| } |
| |
| static void on_p2s_sent_initial_metadata(void *arg, int success) { |
| proxy_call *pc = arg; |
| unrefpc(pc, "on_p2s_sent_initial_metadata"); |
| } |
| |
| static void on_c2p_recv_msg(void *arg, int success); |
| |
| static void on_p2s_sent_message(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| grpc_byte_buffer_destroy(pc->c2p_msg); |
| if (!pc->proxy->shutdown && success) { |
| op.op = GRPC_OP_RECV_MESSAGE; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.recv_message = &pc->c2p_msg; |
| refpc(pc, "on_c2p_recv_msg"); |
| err = grpc_call_start_batch(pc->c2p, &op, 1, |
| new_closure(on_c2p_recv_msg, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| |
| unrefpc(pc, "on_p2s_sent_message"); |
| } |
| |
| static void on_p2s_sent_close(void *arg, int success) { |
| proxy_call *pc = arg; |
| unrefpc(pc, "on_p2s_sent_close"); |
| } |
| |
| static void on_c2p_recv_msg(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| if (!pc->proxy->shutdown && success) { |
| if (pc->c2p_msg != NULL) { |
| op.op = GRPC_OP_SEND_MESSAGE; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.send_message = pc->c2p_msg; |
| refpc(pc, "on_p2s_sent_message"); |
| err = grpc_call_start_batch(pc->p2s, &op, 1, |
| new_closure(on_p2s_sent_message, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } else { |
| op.op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
| op.flags = 0; |
| op.reserved = NULL; |
| refpc(pc, "on_p2s_sent_close"); |
| err = grpc_call_start_batch(pc->p2s, &op, 1, |
| new_closure(on_p2s_sent_close, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| } |
| |
| unrefpc(pc, "on_c2p_recv_msg"); |
| } |
| |
| static void on_p2s_recv_msg(void *arg, int success); |
| |
| static void on_c2p_sent_message(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| grpc_byte_buffer_destroy(pc->p2s_msg); |
| if (!pc->proxy->shutdown && success) { |
| op.op = GRPC_OP_RECV_MESSAGE; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.recv_message = &pc->p2s_msg; |
| refpc(pc, "on_p2s_recv_msg"); |
| err = grpc_call_start_batch(pc->p2s, &op, 1, |
| new_closure(on_p2s_recv_msg, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| |
| unrefpc(pc, "on_c2p_sent_message"); |
| } |
| |
| static void on_p2s_recv_msg(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| if (!pc->proxy->shutdown && success && pc->p2s_msg) { |
| op.op = GRPC_OP_SEND_MESSAGE; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.send_message = pc->p2s_msg; |
| refpc(pc, "on_c2p_sent_message"); |
| err = grpc_call_start_batch(pc->c2p, &op, 1, |
| new_closure(on_c2p_sent_message, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| unrefpc(pc, "on_p2s_recv_msg"); |
| } |
| |
| static void on_c2p_sent_status(void *arg, int success) { |
| proxy_call *pc = arg; |
| unrefpc(pc, "on_c2p_sent_status"); |
| } |
| |
| static void on_p2s_status(void *arg, int success) { |
| proxy_call *pc = arg; |
| grpc_op op; |
| grpc_call_error err; |
| |
| if (!pc->proxy->shutdown) { |
| GPR_ASSERT(success); |
| op.op = GRPC_OP_SEND_STATUS_FROM_SERVER; |
| op.flags = 0; |
| op.reserved = NULL; |
| op.data.send_status_from_server.trailing_metadata_count = |
| pc->p2s_trailing_metadata.count; |
| op.data.send_status_from_server.trailing_metadata = |
| pc->p2s_trailing_metadata.metadata; |
| op.data.send_status_from_server.status = pc->p2s_status; |
| op.data.send_status_from_server.status_details = pc->p2s_status_details; |
| refpc(pc, "on_c2p_sent_status"); |
| err = grpc_call_start_batch(pc->c2p, &op, 1, |
| new_closure(on_c2p_sent_status, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| } |
| |
| unrefpc(pc, "on_p2s_status"); |
| } |
| |
| static void on_c2p_closed(void *arg, int success) { |
| proxy_call *pc = arg; |
| unrefpc(pc, "on_c2p_closed"); |
| } |
| |
| static void on_new_call(void *arg, int success) { |
| grpc_end2end_proxy *proxy = arg; |
| grpc_call_error err; |
| |
| if (success) { |
| grpc_op op; |
| proxy_call *pc = gpr_malloc(sizeof(*pc)); |
| memset(pc, 0, sizeof(*pc)); |
| pc->proxy = proxy; |
| GPR_SWAP(grpc_metadata_array, pc->c2p_initial_metadata, |
| proxy->new_call_metadata); |
| pc->c2p = proxy->new_call; |
| pc->p2s = grpc_channel_create_call( |
| proxy->client, pc->c2p, GRPC_PROPAGATE_DEFAULTS, proxy->cq, |
| proxy->new_call_details.method, proxy->new_call_details.host, |
| proxy->new_call_details.deadline, NULL); |
| gpr_ref_init(&pc->refs, 1); |
| |
| op.reserved = NULL; |
| |
| op.op = GRPC_OP_RECV_INITIAL_METADATA; |
| op.flags = 0; |
| op.data.recv_initial_metadata = &pc->p2s_initial_metadata; |
| refpc(pc, "on_p2s_recv_initial_metadata"); |
| err = grpc_call_start_batch( |
| pc->p2s, &op, 1, new_closure(on_p2s_recv_initial_metadata, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| op.op = GRPC_OP_SEND_INITIAL_METADATA; |
| op.flags = proxy->new_call_details.flags; |
| op.data.send_initial_metadata.count = pc->c2p_initial_metadata.count; |
| op.data.send_initial_metadata.metadata = pc->c2p_initial_metadata.metadata; |
| refpc(pc, "on_p2s_sent_initial_metadata"); |
| err = grpc_call_start_batch( |
| pc->p2s, &op, 1, new_closure(on_p2s_sent_initial_metadata, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| op.op = GRPC_OP_RECV_MESSAGE; |
| op.flags = 0; |
| op.data.recv_message = &pc->c2p_msg; |
| refpc(pc, "on_c2p_recv_msg"); |
| err = grpc_call_start_batch(pc->c2p, &op, 1, |
| new_closure(on_c2p_recv_msg, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| op.op = GRPC_OP_RECV_MESSAGE; |
| op.flags = 0; |
| op.data.recv_message = &pc->p2s_msg; |
| refpc(pc, "on_p2s_recv_msg"); |
| err = grpc_call_start_batch(pc->p2s, &op, 1, |
| new_closure(on_p2s_recv_msg, pc), NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| op.op = GRPC_OP_RECV_STATUS_ON_CLIENT; |
| op.flags = 0; |
| op.data.recv_status_on_client.trailing_metadata = |
| &pc->p2s_trailing_metadata; |
| op.data.recv_status_on_client.status = &pc->p2s_status; |
| op.data.recv_status_on_client.status_details = &pc->p2s_status_details; |
| op.data.recv_status_on_client.status_details_capacity = |
| &pc->p2s_status_details_capacity; |
| refpc(pc, "on_p2s_status"); |
| err = grpc_call_start_batch(pc->p2s, &op, 1, new_closure(on_p2s_status, pc), |
| NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| op.op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
| op.flags = 0; |
| op.data.recv_close_on_server.cancelled = &pc->c2p_server_cancelled; |
| refpc(pc, "on_c2p_closed"); |
| err = grpc_call_start_batch(pc->c2p, &op, 1, new_closure(on_c2p_closed, pc), |
| NULL); |
| GPR_ASSERT(err == GRPC_CALL_OK); |
| |
| request_call(proxy); |
| |
| unrefpc(pc, "init"); |
| } else { |
| GPR_ASSERT(proxy->new_call == NULL); |
| } |
| } |
| |
| static void request_call(grpc_end2end_proxy *proxy) { |
| proxy->new_call = NULL; |
| GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( |
| proxy->server, &proxy->new_call, |
| &proxy->new_call_details, |
| &proxy->new_call_metadata, proxy->cq, |
| proxy->cq, new_closure(on_new_call, proxy))); |
| } |
| |
| static void thread_main(void *arg) { |
| grpc_end2end_proxy *proxy = arg; |
| closure *cl; |
| for (;;) { |
| grpc_event ev = grpc_completion_queue_next( |
| proxy->cq, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL); |
| switch (ev.type) { |
| case GRPC_QUEUE_TIMEOUT: |
| gpr_log(GPR_ERROR, "Should never reach here"); |
| abort(); |
| case GRPC_QUEUE_SHUTDOWN: |
| return; |
| case GRPC_OP_COMPLETE: |
| cl = ev.tag; |
| cl->func(cl->arg, ev.success); |
| gpr_free(cl); |
| break; |
| } |
| } |
| } |
| |
| const char *grpc_end2end_proxy_get_client_target(grpc_end2end_proxy *proxy) { |
| return proxy->proxy_port; |
| } |
| |
| const char *grpc_end2end_proxy_get_server_port(grpc_end2end_proxy *proxy) { |
| return proxy->server_port; |
| } |