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