1 /*
2 *
3 * Copyright 2016 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include "test/core/end2end/fixtures/http_proxy_fixture.h"
20
21 #include "src/core/lib/iomgr/sockaddr.h"
22
23 #include <string.h>
24
25 #include <grpc/grpc.h>
26 #include <grpc/slice_buffer.h>
27 #include <grpc/support/alloc.h>
28 #include <grpc/support/atm.h>
29 #include <grpc/support/log.h>
30 #include <grpc/support/string_util.h>
31 #include <grpc/support/sync.h>
32
33 #include "src/core/lib/channel/channel_args.h"
34 #include "src/core/lib/gpr/string.h"
35 #include "src/core/lib/gprpp/host_port.h"
36 #include "src/core/lib/gprpp/thd.h"
37 #include "src/core/lib/http/parser.h"
38 #include "src/core/lib/iomgr/closure.h"
39 #include "src/core/lib/iomgr/combiner.h"
40 #include "src/core/lib/iomgr/endpoint.h"
41 #include "src/core/lib/iomgr/error.h"
42 #include "src/core/lib/iomgr/exec_ctx.h"
43 #include "src/core/lib/iomgr/pollset.h"
44 #include "src/core/lib/iomgr/pollset_set.h"
45 #include "src/core/lib/iomgr/resolve_address.h"
46 #include "src/core/lib/iomgr/sockaddr_utils.h"
47 #include "src/core/lib/iomgr/tcp_client.h"
48 #include "src/core/lib/iomgr/tcp_server.h"
49 #include "src/core/lib/iomgr/timer.h"
50 #include "src/core/lib/slice/b64.h"
51 #include "src/core/lib/slice/slice_internal.h"
52 #include "test/core/util/port.h"
53
54 struct grpc_end2end_http_proxy {
grpc_end2end_http_proxygrpc_end2end_http_proxy55 grpc_end2end_http_proxy()
56 : server(nullptr),
57 channel_args(nullptr),
58 mu(nullptr),
59 pollset(nullptr),
60 combiner(nullptr) {
61 gpr_ref_init(&users, 1);
62 combiner = grpc_combiner_create();
63 }
64 grpc_core::UniquePtr<char> proxy_name;
65 grpc_core::Thread thd;
66 grpc_tcp_server* server;
67 grpc_channel_args* channel_args;
68 gpr_mu* mu;
69 grpc_pollset* pollset;
70 gpr_refcount users;
71
72 grpc_combiner* combiner;
73 };
74
75 //
76 // Connection handling
77 //
78
79 // proxy_connection structure is only accessed in the closures which are all
80 // scheduled under the same combiner lock. So there is no need for a mutex to
81 // protect this structure.
82 typedef struct proxy_connection {
83 grpc_end2end_http_proxy* proxy;
84
85 grpc_endpoint* client_endpoint;
86 grpc_endpoint* server_endpoint;
87
88 gpr_refcount refcount;
89
90 grpc_pollset_set* pollset_set;
91
92 // NOTE: All the closures execute under proxy->combiner lock. Which means
93 // there will not be any data-races between the closures
94 grpc_closure on_read_request_done;
95 grpc_closure on_server_connect_done;
96 grpc_closure on_write_response_done;
97 grpc_closure on_client_read_done;
98 grpc_closure on_client_write_done;
99 grpc_closure on_server_read_done;
100 grpc_closure on_server_write_done;
101
102 bool client_read_failed : 1;
103 bool client_write_failed : 1;
104 bool client_shutdown : 1;
105 bool server_read_failed : 1;
106 bool server_write_failed : 1;
107 bool server_shutdown : 1;
108
109 grpc_slice_buffer client_read_buffer;
110 grpc_slice_buffer client_deferred_write_buffer;
111 bool client_is_writing;
112 grpc_slice_buffer client_write_buffer;
113 grpc_slice_buffer server_read_buffer;
114 grpc_slice_buffer server_deferred_write_buffer;
115 bool server_is_writing;
116 grpc_slice_buffer server_write_buffer;
117
118 grpc_http_parser http_parser;
119 grpc_http_request http_request;
120 } proxy_connection;
121
proxy_connection_ref(proxy_connection * conn,const char * reason)122 static void proxy_connection_ref(proxy_connection* conn, const char* reason) {
123 gpr_ref(&conn->refcount);
124 }
125
126 // Helper function to destroy the proxy connection.
proxy_connection_unref(proxy_connection * conn,const char * reason)127 static void proxy_connection_unref(proxy_connection* conn, const char* reason) {
128 if (gpr_unref(&conn->refcount)) {
129 gpr_log(GPR_DEBUG, "endpoints: %p %p", conn->client_endpoint,
130 conn->server_endpoint);
131 grpc_endpoint_destroy(conn->client_endpoint);
132 if (conn->server_endpoint != nullptr) {
133 grpc_endpoint_destroy(conn->server_endpoint);
134 }
135 grpc_pollset_set_destroy(conn->pollset_set);
136 grpc_slice_buffer_destroy_internal(&conn->client_read_buffer);
137 grpc_slice_buffer_destroy_internal(&conn->client_deferred_write_buffer);
138 grpc_slice_buffer_destroy_internal(&conn->client_write_buffer);
139 grpc_slice_buffer_destroy_internal(&conn->server_read_buffer);
140 grpc_slice_buffer_destroy_internal(&conn->server_deferred_write_buffer);
141 grpc_slice_buffer_destroy_internal(&conn->server_write_buffer);
142 grpc_http_parser_destroy(&conn->http_parser);
143 grpc_http_request_destroy(&conn->http_request);
144 gpr_unref(&conn->proxy->users);
145 gpr_free(conn);
146 }
147 }
148
149 enum failure_type {
150 SETUP_FAILED, // To be used before we start proxying.
151 CLIENT_READ_FAILED,
152 CLIENT_WRITE_FAILED,
153 SERVER_READ_FAILED,
154 SERVER_WRITE_FAILED,
155 };
156
157 // Helper function to shut down the proxy connection.
proxy_connection_failed(proxy_connection * conn,failure_type failure,const char * prefix,grpc_error * error)158 static void proxy_connection_failed(proxy_connection* conn,
159 failure_type failure, const char* prefix,
160 grpc_error* error) {
161 gpr_log(GPR_INFO, "%s: %s", prefix, grpc_error_string(error));
162 // Decide whether we should shut down the client and server.
163 bool shutdown_client = false;
164 bool shutdown_server = false;
165 if (failure == SETUP_FAILED) {
166 shutdown_client = true;
167 shutdown_server = true;
168 } else {
169 if ((failure == CLIENT_READ_FAILED && conn->client_write_failed) ||
170 (failure == CLIENT_WRITE_FAILED && conn->client_read_failed) ||
171 (failure == SERVER_READ_FAILED && !conn->client_is_writing)) {
172 shutdown_client = true;
173 }
174 if ((failure == SERVER_READ_FAILED && conn->server_write_failed) ||
175 (failure == SERVER_WRITE_FAILED && conn->server_read_failed) ||
176 (failure == CLIENT_READ_FAILED && !conn->server_is_writing)) {
177 shutdown_server = true;
178 }
179 }
180 // If we decided to shut down either one and have not yet done so, do so.
181 if (shutdown_client && !conn->client_shutdown) {
182 grpc_endpoint_shutdown(conn->client_endpoint, GRPC_ERROR_REF(error));
183 conn->client_shutdown = true;
184 }
185 if (shutdown_server && !conn->server_shutdown &&
186 (conn->server_endpoint != nullptr)) {
187 grpc_endpoint_shutdown(conn->server_endpoint, GRPC_ERROR_REF(error));
188 conn->server_shutdown = true;
189 }
190 // Unref the connection.
191 proxy_connection_unref(conn, "conn_failed");
192 GRPC_ERROR_UNREF(error);
193 }
194
195 // Callback for writing proxy data to the client.
on_client_write_done(void * arg,grpc_error * error)196 static void on_client_write_done(void* arg, grpc_error* error) {
197 proxy_connection* conn = static_cast<proxy_connection*>(arg);
198 conn->client_is_writing = false;
199 if (error != GRPC_ERROR_NONE) {
200 proxy_connection_failed(conn, CLIENT_WRITE_FAILED,
201 "HTTP proxy client write", GRPC_ERROR_REF(error));
202 return;
203 }
204 // Clear write buffer (the data we just wrote).
205 grpc_slice_buffer_reset_and_unref(&conn->client_write_buffer);
206 // If more data was read from the server since we started this write,
207 // write that data now.
208 if (conn->client_deferred_write_buffer.length > 0) {
209 grpc_slice_buffer_move_into(&conn->client_deferred_write_buffer,
210 &conn->client_write_buffer);
211 conn->client_is_writing = true;
212 grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
213 &conn->on_client_write_done, nullptr);
214 } else {
215 // No more writes. Unref the connection.
216 proxy_connection_unref(conn, "write_done");
217 }
218 }
219
220 // Callback for writing proxy data to the backend server.
on_server_write_done(void * arg,grpc_error * error)221 static void on_server_write_done(void* arg, grpc_error* error) {
222 proxy_connection* conn = static_cast<proxy_connection*>(arg);
223 conn->server_is_writing = false;
224 if (error != GRPC_ERROR_NONE) {
225 proxy_connection_failed(conn, SERVER_WRITE_FAILED,
226 "HTTP proxy server write", GRPC_ERROR_REF(error));
227 return;
228 }
229 // Clear write buffer (the data we just wrote).
230 grpc_slice_buffer_reset_and_unref(&conn->server_write_buffer);
231 // If more data was read from the client since we started this write,
232 // write that data now.
233 if (conn->server_deferred_write_buffer.length > 0) {
234 grpc_slice_buffer_move_into(&conn->server_deferred_write_buffer,
235 &conn->server_write_buffer);
236 conn->server_is_writing = true;
237 grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer,
238 &conn->on_server_write_done, nullptr);
239 } else {
240 // No more writes. Unref the connection.
241 proxy_connection_unref(conn, "server_write");
242 }
243 }
244
245 // Callback for reading data from the client, which will be proxied to
246 // the backend server.
on_client_read_done(void * arg,grpc_error * error)247 static void on_client_read_done(void* arg, grpc_error* error) {
248 proxy_connection* conn = static_cast<proxy_connection*>(arg);
249 if (error != GRPC_ERROR_NONE) {
250 proxy_connection_failed(conn, CLIENT_READ_FAILED, "HTTP proxy client read",
251 GRPC_ERROR_REF(error));
252 return;
253 }
254 // If there is already a pending write (i.e., server_write_buffer is
255 // not empty), then move the read data into server_deferred_write_buffer,
256 // and the next write will be requested in on_server_write_done(), when
257 // the current write is finished.
258 //
259 // Otherwise, move the read data into the write buffer and write it.
260 if (conn->server_is_writing) {
261 grpc_slice_buffer_move_into(&conn->client_read_buffer,
262 &conn->server_deferred_write_buffer);
263 } else {
264 grpc_slice_buffer_move_into(&conn->client_read_buffer,
265 &conn->server_write_buffer);
266 proxy_connection_ref(conn, "client_read");
267 conn->server_is_writing = true;
268 grpc_endpoint_write(conn->server_endpoint, &conn->server_write_buffer,
269 &conn->on_server_write_done, nullptr);
270 }
271 // Read more data.
272 grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer,
273 &conn->on_client_read_done, /*urgent=*/false);
274 }
275
276 // Callback for reading data from the backend server, which will be
277 // proxied to the client.
on_server_read_done(void * arg,grpc_error * error)278 static void on_server_read_done(void* arg, grpc_error* error) {
279 proxy_connection* conn = static_cast<proxy_connection*>(arg);
280 if (error != GRPC_ERROR_NONE) {
281 proxy_connection_failed(conn, SERVER_READ_FAILED, "HTTP proxy server read",
282 GRPC_ERROR_REF(error));
283 return;
284 }
285 // If there is already a pending write (i.e., client_write_buffer is
286 // not empty), then move the read data into client_deferred_write_buffer,
287 // and the next write will be requested in on_client_write_done(), when
288 // the current write is finished.
289 //
290 // Otherwise, move the read data into the write buffer and write it.
291 if (conn->client_is_writing) {
292 grpc_slice_buffer_move_into(&conn->server_read_buffer,
293 &conn->client_deferred_write_buffer);
294 } else {
295 grpc_slice_buffer_move_into(&conn->server_read_buffer,
296 &conn->client_write_buffer);
297 proxy_connection_ref(conn, "server_read");
298 conn->client_is_writing = true;
299 grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
300 &conn->on_client_write_done, nullptr);
301 }
302 // Read more data.
303 grpc_endpoint_read(conn->server_endpoint, &conn->server_read_buffer,
304 &conn->on_server_read_done, /*urgent=*/false);
305 }
306
307 // Callback to write the HTTP response for the CONNECT request.
on_write_response_done(void * arg,grpc_error * error)308 static void on_write_response_done(void* arg, grpc_error* error) {
309 proxy_connection* conn = static_cast<proxy_connection*>(arg);
310 conn->client_is_writing = false;
311 if (error != GRPC_ERROR_NONE) {
312 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy write response",
313 GRPC_ERROR_REF(error));
314 return;
315 }
316 // Clear write buffer.
317 grpc_slice_buffer_reset_and_unref(&conn->client_write_buffer);
318 // Start reading from both client and server. One of the read
319 // requests inherits our ref to conn, but we need to take a new ref
320 // for the other one.
321 proxy_connection_ref(conn, "client_read");
322 proxy_connection_ref(conn, "server_read");
323 proxy_connection_unref(conn, "write_response");
324 grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer,
325 &conn->on_client_read_done, /*urgent=*/false);
326 grpc_endpoint_read(conn->server_endpoint, &conn->server_read_buffer,
327 &conn->on_server_read_done, /*urgent=*/false);
328 }
329
330 // Callback to connect to the backend server specified by the HTTP
331 // CONNECT request.
on_server_connect_done(void * arg,grpc_error * error)332 static void on_server_connect_done(void* arg, grpc_error* error) {
333 proxy_connection* conn = static_cast<proxy_connection*>(arg);
334 if (error != GRPC_ERROR_NONE) {
335 // TODO(roth): Technically, in this case, we should handle the error
336 // by returning an HTTP response to the client indicating that the
337 // connection failed. However, for the purposes of this test code,
338 // it's fine to pretend this is a client-side error, which will
339 // cause the client connection to be dropped.
340 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy server connect",
341 GRPC_ERROR_REF(error));
342 return;
343 }
344 // We've established a connection, so send back a 200 response code to
345 // the client.
346 // The write callback inherits our reference to conn.
347 grpc_slice slice =
348 grpc_slice_from_copied_string("HTTP/1.0 200 connected\r\n\r\n");
349 grpc_slice_buffer_add(&conn->client_write_buffer, slice);
350 conn->client_is_writing = true;
351 grpc_endpoint_write(conn->client_endpoint, &conn->client_write_buffer,
352 &conn->on_write_response_done, nullptr);
353 }
354
355 /**
356 * Parses the proxy auth header value to check if it matches :-
357 * Basic <base64_encoded_expected_cred>
358 * Returns true if it matches, false otherwise
359 */
proxy_auth_header_matches(char * proxy_auth_header_val,char * expected_cred)360 static bool proxy_auth_header_matches(char* proxy_auth_header_val,
361 char* expected_cred) {
362 GPR_ASSERT(proxy_auth_header_val != nullptr);
363 GPR_ASSERT(expected_cred != nullptr);
364 if (strncmp(proxy_auth_header_val, "Basic ", 6) != 0) {
365 return false;
366 }
367 proxy_auth_header_val += 6;
368 grpc_slice decoded_slice = grpc_base64_decode(proxy_auth_header_val, 0);
369 const bool header_matches =
370 grpc_slice_str_cmp(decoded_slice, expected_cred) == 0;
371 grpc_slice_unref_internal(decoded_slice);
372 return header_matches;
373 }
374
375 // Callback to read the HTTP CONNECT request.
376 // TODO(roth): Technically, for any of the failure modes handled by this
377 // function, we should handle the error by returning an HTTP response to
378 // the client indicating that the request failed. However, for the purposes
379 // of this test code, it's fine to pretend this is a client-side error,
380 // which will cause the client connection to be dropped.
on_read_request_done(void * arg,grpc_error * error)381 static void on_read_request_done(void* arg, grpc_error* error) {
382 proxy_connection* conn = static_cast<proxy_connection*>(arg);
383 gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn,
384 grpc_error_string(error));
385 if (error != GRPC_ERROR_NONE) {
386 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy read request",
387 GRPC_ERROR_REF(error));
388 return;
389 }
390 // Read request and feed it to the parser.
391 for (size_t i = 0; i < conn->client_read_buffer.count; ++i) {
392 if (GRPC_SLICE_LENGTH(conn->client_read_buffer.slices[i]) > 0) {
393 error = grpc_http_parser_parse(
394 &conn->http_parser, conn->client_read_buffer.slices[i], nullptr);
395 if (error != GRPC_ERROR_NONE) {
396 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy request parse",
397 GRPC_ERROR_REF(error));
398 GRPC_ERROR_UNREF(error);
399 return;
400 }
401 }
402 }
403 grpc_slice_buffer_reset_and_unref(&conn->client_read_buffer);
404 // If we're not done reading the request, read more data.
405 if (conn->http_parser.state != GRPC_HTTP_BODY) {
406 grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer,
407 &conn->on_read_request_done, /*urgent=*/false);
408 return;
409 }
410 // Make sure we got a CONNECT request.
411 if (strcmp(conn->http_request.method, "CONNECT") != 0) {
412 char* msg;
413 gpr_asprintf(&msg, "HTTP proxy got request method %s",
414 conn->http_request.method);
415 error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
416 gpr_free(msg);
417 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy read request",
418 GRPC_ERROR_REF(error));
419 GRPC_ERROR_UNREF(error);
420 return;
421 }
422 // If proxy auth is being used, check if the header is present and as expected
423 const grpc_arg* proxy_auth_arg = grpc_channel_args_find(
424 conn->proxy->channel_args, GRPC_ARG_HTTP_PROXY_AUTH_CREDS);
425 char* proxy_auth_str = grpc_channel_arg_get_string(proxy_auth_arg);
426 if (proxy_auth_str != nullptr) {
427 bool client_authenticated = false;
428 for (size_t i = 0; i < conn->http_request.hdr_count; i++) {
429 if (strcmp(conn->http_request.hdrs[i].key, "Proxy-Authorization") == 0) {
430 client_authenticated = proxy_auth_header_matches(
431 conn->http_request.hdrs[i].value, proxy_auth_str);
432 break;
433 }
434 }
435 if (!client_authenticated) {
436 const char* msg = "HTTP Connect could not verify authentication";
437 error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(msg);
438 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy read request",
439 GRPC_ERROR_REF(error));
440 GRPC_ERROR_UNREF(error);
441 return;
442 }
443 }
444 // Resolve address.
445 grpc_resolved_addresses* resolved_addresses = nullptr;
446 error = grpc_blocking_resolve_address(conn->http_request.path, "80",
447 &resolved_addresses);
448 if (error != GRPC_ERROR_NONE) {
449 proxy_connection_failed(conn, SETUP_FAILED, "HTTP proxy DNS lookup",
450 GRPC_ERROR_REF(error));
451 GRPC_ERROR_UNREF(error);
452 return;
453 }
454 GPR_ASSERT(resolved_addresses->naddrs >= 1);
455 // Connect to requested address.
456 // The connection callback inherits our reference to conn.
457 const grpc_millis deadline =
458 grpc_core::ExecCtx::Get()->Now() + 10 * GPR_MS_PER_SEC;
459 grpc_tcp_client_connect(&conn->on_server_connect_done, &conn->server_endpoint,
460 conn->pollset_set, nullptr,
461 &resolved_addresses->addrs[0], deadline);
462 grpc_resolved_addresses_destroy(resolved_addresses);
463 }
464
on_accept(void * arg,grpc_endpoint * endpoint,grpc_pollset * accepting_pollset,grpc_tcp_server_acceptor * acceptor)465 static void on_accept(void* arg, grpc_endpoint* endpoint,
466 grpc_pollset* accepting_pollset,
467 grpc_tcp_server_acceptor* acceptor) {
468 gpr_free(acceptor);
469 grpc_end2end_http_proxy* proxy = static_cast<grpc_end2end_http_proxy*>(arg);
470 // Instantiate proxy_connection.
471 proxy_connection* conn =
472 static_cast<proxy_connection*>(gpr_zalloc(sizeof(*conn)));
473 gpr_ref(&proxy->users);
474 conn->client_endpoint = endpoint;
475 conn->proxy = proxy;
476 gpr_ref_init(&conn->refcount, 1);
477 conn->pollset_set = grpc_pollset_set_create();
478 grpc_pollset_set_add_pollset(conn->pollset_set, proxy->pollset);
479 grpc_endpoint_add_to_pollset_set(endpoint, conn->pollset_set);
480 GRPC_CLOSURE_INIT(&conn->on_read_request_done, on_read_request_done, conn,
481 grpc_combiner_scheduler(conn->proxy->combiner));
482 GRPC_CLOSURE_INIT(&conn->on_server_connect_done, on_server_connect_done, conn,
483 grpc_combiner_scheduler(conn->proxy->combiner));
484 GRPC_CLOSURE_INIT(&conn->on_write_response_done, on_write_response_done, conn,
485 grpc_combiner_scheduler(conn->proxy->combiner));
486 GRPC_CLOSURE_INIT(&conn->on_client_read_done, on_client_read_done, conn,
487 grpc_combiner_scheduler(conn->proxy->combiner));
488 GRPC_CLOSURE_INIT(&conn->on_client_write_done, on_client_write_done, conn,
489 grpc_combiner_scheduler(conn->proxy->combiner));
490 GRPC_CLOSURE_INIT(&conn->on_server_read_done, on_server_read_done, conn,
491 grpc_combiner_scheduler(conn->proxy->combiner));
492 GRPC_CLOSURE_INIT(&conn->on_server_write_done, on_server_write_done, conn,
493 grpc_combiner_scheduler(conn->proxy->combiner));
494 grpc_slice_buffer_init(&conn->client_read_buffer);
495 grpc_slice_buffer_init(&conn->client_deferred_write_buffer);
496 conn->client_is_writing = false;
497 grpc_slice_buffer_init(&conn->client_write_buffer);
498 grpc_slice_buffer_init(&conn->server_read_buffer);
499 grpc_slice_buffer_init(&conn->server_deferred_write_buffer);
500 conn->server_is_writing = false;
501 grpc_slice_buffer_init(&conn->server_write_buffer);
502 grpc_http_parser_init(&conn->http_parser, GRPC_HTTP_REQUEST,
503 &conn->http_request);
504 grpc_endpoint_read(conn->client_endpoint, &conn->client_read_buffer,
505 &conn->on_read_request_done, /*urgent=*/false);
506 }
507
508 //
509 // Proxy class
510 //
511
thread_main(void * arg)512 static void thread_main(void* arg) {
513 grpc_end2end_http_proxy* proxy = static_cast<grpc_end2end_http_proxy*>(arg);
514 grpc_core::ExecCtx exec_ctx;
515 do {
516 gpr_ref(&proxy->users);
517 grpc_pollset_worker* worker = nullptr;
518 gpr_mu_lock(proxy->mu);
519 GRPC_LOG_IF_ERROR(
520 "grpc_pollset_work",
521 grpc_pollset_work(proxy->pollset, &worker,
522 grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC));
523 gpr_mu_unlock(proxy->mu);
524 grpc_core::ExecCtx::Get()->Flush();
525 } while (!gpr_unref(&proxy->users));
526 }
527
grpc_end2end_http_proxy_create(grpc_channel_args * args)528 grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(
529 grpc_channel_args* args) {
530 grpc_core::ExecCtx exec_ctx;
531 grpc_end2end_http_proxy* proxy = grpc_core::New<grpc_end2end_http_proxy>();
532 // Construct proxy address.
533 const int proxy_port = grpc_pick_unused_port_or_die();
534 grpc_core::JoinHostPort(&proxy->proxy_name, "localhost", proxy_port);
535 gpr_log(GPR_INFO, "Proxy address: %s", proxy->proxy_name.get());
536 // Create TCP server.
537 proxy->channel_args = grpc_channel_args_copy(args);
538 grpc_error* error =
539 grpc_tcp_server_create(nullptr, proxy->channel_args, &proxy->server);
540 GPR_ASSERT(error == GRPC_ERROR_NONE);
541 // Bind to port.
542 grpc_resolved_address resolved_addr;
543 grpc_sockaddr_in* addr =
544 reinterpret_cast<grpc_sockaddr_in*>(resolved_addr.addr);
545 memset(&resolved_addr, 0, sizeof(resolved_addr));
546 addr->sin_family = GRPC_AF_INET;
547 grpc_sockaddr_set_port(&resolved_addr, proxy_port);
548 int port;
549 error = grpc_tcp_server_add_port(proxy->server, &resolved_addr, &port);
550 GPR_ASSERT(error == GRPC_ERROR_NONE);
551 GPR_ASSERT(port == proxy_port);
552 // Start server.
553 proxy->pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
554 grpc_pollset_init(proxy->pollset, &proxy->mu);
555 grpc_tcp_server_start(proxy->server, &proxy->pollset, 1, on_accept, proxy);
556
557 // Start proxy thread.
558 proxy->thd = grpc_core::Thread("grpc_http_proxy", thread_main, proxy);
559 proxy->thd.Start();
560 return proxy;
561 }
562
destroy_pollset(void * arg,grpc_error * error)563 static void destroy_pollset(void* arg, grpc_error* error) {
564 grpc_pollset* pollset = static_cast<grpc_pollset*>(arg);
565 grpc_pollset_destroy(pollset);
566 gpr_free(pollset);
567 }
568
grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy * proxy)569 void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) {
570 gpr_unref(&proxy->users); // Signal proxy thread to shutdown.
571 grpc_core::ExecCtx exec_ctx;
572 proxy->thd.Join();
573 grpc_tcp_server_shutdown_listeners(proxy->server);
574 grpc_tcp_server_unref(proxy->server);
575 grpc_channel_args_destroy(proxy->channel_args);
576 grpc_pollset_shutdown(proxy->pollset,
577 GRPC_CLOSURE_CREATE(destroy_pollset, proxy->pollset,
578 grpc_schedule_on_exec_ctx));
579 GRPC_COMBINER_UNREF(proxy->combiner, "test");
580 grpc_core::Delete(proxy);
581 }
582
grpc_end2end_http_proxy_get_proxy_name(grpc_end2end_http_proxy * proxy)583 const char* grpc_end2end_http_proxy_get_proxy_name(
584 grpc_end2end_http_proxy* proxy) {
585 return proxy->proxy_name.get();
586 }
587