diff --git a/BUILD b/BUILD
index 2f9bbbc5ac0c984fbae0056295e7809a5290c3f8..aa12700818bad5d68a0e4f311d535e54b7a976ba 100644
--- a/BUILD
+++ b/BUILD
@@ -1076,6 +1076,7 @@ grpc_cc_library(
],
hdrs = [
"third_party/objective_c/Cronet/bidirectional_stream_c.h",
+ "src/core/ext/transport/cronet/transport/cronet_transport.h",
],
language = "c",
public_hdrs = [
diff --git a/build.yaml b/build.yaml
index 2fbaa2e2262375af965702b23e3351576642172d..7853d68d2373878742ed266ab87af28b77fd5534 100644
--- a/build.yaml
+++ b/build.yaml
@@ -724,6 +724,7 @@ filegroups:
- include/grpc/grpc_security.h
- include/grpc/grpc_security_constants.h
headers:
+ - src/core/ext/transport/cronet/transport/cronet_transport.h
- third_party/objective_c/Cronet/bidirectional_stream_c.h
src:
- src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec
index 07b3b2bad5391ab593c4f4869ae6ab3ac65dcb62..8f9db71a53bf737ac54a32b57984c7b894c725b7 100644
--- a/gRPC-Core.podspec
+++ b/gRPC-Core.podspec
@@ -884,7 +884,8 @@ Pod::Spec.new do |s|
s.subspec 'Cronet-Interface' do |ss|
ss.header_mappings_dir = 'include/grpc'
- ss.source_files = 'include/grpc/grpc_cronet.h'
+ ss.source_files = 'include/grpc/grpc_cronet.h',
+ 'src/core/ext/transport/cronet/transport/cronet_transport.h'
end
s.subspec 'Cronet-Implementation' do |ss|
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index 6661547f6b60f4fb1e6ac136df01e0e93ac97297..da936bf02818c76fea4c79f3cfe5aa8e81845f66 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -230,6 +230,10 @@ typedef struct {
#define GRPC_ARG_LB_POLICY_NAME "grpc.lb_policy_name"
/** The grpc_socket_mutator instance that set the socket options. A pointer. */
#define GRPC_ARG_SOCKET_MUTATOR "grpc.socket_mutator"
+/** If non-zero, Cronet transport will coalesce packets to fewer frames when
+ * possible. */
+#define GRPC_ARG_USE_CRONET_PACKET_COALESCING \
+ "grpc.use_cronet_packet_coalescing"
/** \} */
/** Result of a grpc call. If the caller satisfies the prerequisites of a
diff --git a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
index 477cf07f45dbf81144b1c5c0b7dd298460c34817..b6e9e845df366ad08c3b345ec099c833d793c9d9 100644
--- a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
+++ b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
@@ -39,6 +39,7 @@
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
+#include "src/core/ext/transport/cronet/transport/cronet_transport.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/transport/transport_impl.h"
@@ -54,16 +55,14 @@ extern grpc_transport_vtable grpc_cronet_vtable;
GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
void *engine, const char *target, const grpc_channel_args *args,
void *reserved) {
- cronet_transport *ct = gpr_malloc(sizeof(cronet_transport));
- ct->base.vtable = &grpc_cronet_vtable;
- ct->engine = engine;
- ct->host = gpr_malloc(strlen(target) + 1);
- strcpy(ct->host, target);
gpr_log(GPR_DEBUG,
"grpc_create_cronet_transport: stream_engine = %p, target=%s", engine,
- ct->host);
+ target);
+
+ grpc_transport *ct =
+ grpc_create_cronet_transport(engine, target, args, reserved);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
return grpc_channel_create(&exec_ctx, target, args,
- GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
+ GRPC_CLIENT_DIRECT_CHANNEL, ct);
}
diff --git a/src/core/ext/transport/cronet/transport/cronet_api_dummy.c b/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
index da6c0b4fbcce5a956d147b116ec4cb3f237693ba..0dc6a5152fd29ade03ee27b21557137cf5373b1e 100644
--- a/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
+++ b/src/core/ext/transport/cronet/transport/cronet_api_dummy.c
@@ -80,4 +80,16 @@ void bidirectional_stream_cancel(bidirectional_stream* stream) {
GPR_ASSERT(0);
}
+void bidirectional_stream_disable_auto_flush(bidirectional_stream* stream,
+ bool disable_auto_flush) {
+ GPR_ASSERT(0);
+}
+
+void bidirectional_stream_delay_request_headers_until_flush(
+ bidirectional_stream* stream, bool delay_headers_until_flush) {
+ GPR_ASSERT(0);
+}
+
+void bidirectional_stream_flush(bidirectional_stream* stream) { GPR_ASSERT(0); }
+
#endif /* GRPC_COMPILE_WITH_CRONET */
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index d755b1f147fb6b7eb01857aeac61d22913caed66..01a03533daf16338068cfe7a114760e782d2ce6f 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -88,7 +88,7 @@ enum e_op_id {
/* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */
-static void on_request_headers_sent(bidirectional_stream *);
+static void on_stream_ready(bidirectional_stream *);
static void on_response_headers_received(
bidirectional_stream *, const bidirectional_stream_header_array *,
const char *);
@@ -100,7 +100,7 @@ static void on_succeeded(bidirectional_stream *);
static void on_failed(bidirectional_stream *, int);
static void on_canceled(bidirectional_stream *);
static bidirectional_stream_callback cronet_callbacks = {
- on_request_headers_sent,
+ on_stream_ready,
on_response_headers_received,
on_read_completed,
on_write_completed,
@@ -114,6 +114,7 @@ struct grpc_cronet_transport {
grpc_transport base; /* must be first element in this structure */
stream_engine *engine;
char *host;
+ bool use_packet_coalescing;
};
typedef struct grpc_cronet_transport grpc_cronet_transport;
@@ -152,6 +153,9 @@ struct op_state {
bool state_callback_received[OP_NUM_OPS];
bool fail_state;
bool flush_read;
+ bool flush_cronet_when_ready;
+ bool pending_write_for_trailer;
+ bool unprocessed_send_message;
grpc_error *cancel_error;
/* data structure for storing data coming from server */
struct read_state rs;
@@ -175,7 +179,7 @@ struct op_storage {
struct stream_obj {
struct op_and_state *oas;
grpc_transport_stream_op *curr_op;
- grpc_cronet_transport curr_ct;
+ grpc_cronet_transport *curr_ct;
grpc_stream *curr_gs;
bidirectional_stream *cbs;
bidirectional_stream_header_array header_array;
@@ -274,6 +278,9 @@ static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) {
new_op->next = storage->head;
storage->head = new_op;
storage->num_pending_ops++;
+ if (op->send_message) {
+ s->state.unprocessed_send_message = true;
+ }
CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op,
storage->num_pending_ops);
gpr_mu_unlock(&s->mu);
@@ -406,9 +413,10 @@ static void on_succeeded(bidirectional_stream *stream) {
/*
Cronet callback
*/
-static void on_request_headers_sent(bidirectional_stream *stream) {
- CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
+static void on_stream_ready(bidirectional_stream *stream) {
+ CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
stream_obj *s = (stream_obj *)stream->annotation;
+ grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
gpr_mu_lock(&s->mu);
s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true;
s->state.state_callback_received[OP_SEND_INITIAL_METADATA] = true;
@@ -417,6 +425,14 @@ static void on_request_headers_sent(bidirectional_stream *stream) {
gpr_free(s->header_array.headers);
s->header_array.headers = NULL;
}
+ /* Send the initial metadata on wire if there is no SEND_MESSAGE or
+ * SEND_TRAILING_METADATA ops pending */
+ if (t->use_packet_coalescing) {
+ if (s->state.flush_cronet_when_ready) {
+ CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+ bidirectional_stream_flush(stream);
+ }
+ }
gpr_mu_unlock(&s->mu);
execute_from_storage(s);
}
@@ -528,6 +544,7 @@ static void on_response_trailers_received(
CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
trailers);
stream_obj *s = (stream_obj *)stream->annotation;
+ grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
gpr_mu_lock(&s->mu);
memset(&s->state.rs.trailing_metadata, 0,
sizeof(s->state.rs.trailing_metadata));
@@ -558,6 +575,10 @@ static void on_response_trailers_received(
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
s->state.state_callback_received[OP_SEND_MESSAGE] = false;
bidirectional_stream_write(s->cbs, "", 0, true);
+ if (t->use_packet_coalescing) {
+ CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+ bidirectional_stream_flush(s->cbs);
+ }
s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true;
gpr_mu_unlock(&s->mu);
@@ -607,7 +628,7 @@ static void convert_metadata_to_cronet_headers(
curr = curr->next;
num_headers_available++;
}
- /* Allocate enough memory. It is freed in the on_request_headers_sent callback
+ /* Allocate enough memory. It is freed in the on_stream_ready callback
*/
bidirectional_stream_header *headers =
(bidirectional_stream_header *)gpr_malloc(
@@ -687,8 +708,10 @@ static bool header_has_authority(grpc_linked_mdelem *head) {
executed. This is the heart of the state machine.
*/
static bool op_can_be_run(grpc_transport_stream_op *curr_op,
- struct op_state *stream_state,
- struct op_state *op_state, enum e_op_id op_id) {
+ struct stream_obj *s, struct op_state *op_state,
+ enum e_op_id op_id) {
+ struct op_state *stream_state = &s->state;
+ grpc_cronet_transport *t = s->curr_ct;
bool result = true;
/* When call is canceled, every op can be run, except under following
conditions
@@ -755,12 +778,14 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
result = false;
/* we haven't sent message yet */
- else if (curr_op->send_message &&
+ else if (stream_state->unprocessed_send_message &&
!stream_state->state_op_done[OP_SEND_MESSAGE])
result = false;
/* we haven't got on_write_completed for the send yet */
else if (stream_state->state_op_done[OP_SEND_MESSAGE] &&
- !stream_state->state_callback_received[OP_SEND_MESSAGE])
+ !stream_state->state_callback_received[OP_SEND_MESSAGE] &&
+ !(t->use_packet_coalescing &&
+ stream_state->pending_write_for_trailer))
result = false;
} else if (op_id == OP_CANCEL_ERROR) {
/* already executed */
@@ -833,24 +858,28 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
struct op_and_state *oas) {
grpc_transport_stream_op *stream_op = &oas->op;
struct stream_obj *s = oas->s;
+ grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
struct op_state *stream_state = &s->state;
enum e_op_result result = NO_ACTION_POSSIBLE;
if (stream_op->send_initial_metadata &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_SEND_INITIAL_METADATA)) {
+ op_can_be_run(stream_op, s, &oas->state, OP_SEND_INITIAL_METADATA)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_INITIAL_METADATA", oas);
/* Start new cronet stream. It is destroyed in on_succeeded, on_canceled,
* on_failed */
GPR_ASSERT(s->cbs == NULL);
GPR_ASSERT(!stream_state->state_op_done[OP_SEND_INITIAL_METADATA]);
- s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
- &cronet_callbacks);
+ s->cbs =
+ bidirectional_stream_create(t->engine, s->curr_gs, &cronet_callbacks);
CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs);
+ if (t->use_packet_coalescing) {
+ bidirectional_stream_disable_auto_flush(s->cbs, true);
+ bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
+ }
char *url = NULL;
const char *method = "POST";
s->header_array.headers = NULL;
convert_metadata_to_cronet_headers(
- stream_op->send_initial_metadata->list.head, s->curr_ct.host, &url,
+ stream_op->send_initial_metadata->list.head, t->host, &url,
&s->header_array.headers, &s->header_array.count, &method);
s->header_array.capacity = s->header_array.count;
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url);
@@ -862,30 +891,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
gpr_free((void *)s->header_array.headers[header_index].value);
}
stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
- result = ACTION_TAKEN_WITH_CALLBACK;
- } else if (stream_op->recv_initial_metadata &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_RECV_INITIAL_METADATA)) {
- CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_INITIAL_METADATA", oas);
- if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
- grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
- GRPC_ERROR_NONE);
- } else if (stream_state->state_callback_received[OP_FAILED]) {
- grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
- GRPC_ERROR_NONE);
- } else {
- grpc_chttp2_incoming_metadata_buffer_publish(
- exec_ctx, &oas->s->state.rs.initial_metadata,
- stream_op->recv_initial_metadata);
- grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
- GRPC_ERROR_NONE);
+ if (t->use_packet_coalescing) {
+ if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
+ s->state.flush_cronet_when_ready = true;
+ }
}
- stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
- result = ACTION_TAKEN_NO_CALLBACK;
+ result = ACTION_TAKEN_WITH_CALLBACK;
} else if (stream_op->send_message &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_SEND_MESSAGE)) {
+ op_can_be_run(stream_op, s, &oas->state, OP_SEND_MESSAGE)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_MESSAGE", oas);
+ stream_state->unprocessed_send_message = false;
if (stream_state->state_callback_received[OP_FAILED]) {
result = NO_ACTION_POSSIBLE;
CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
@@ -916,16 +931,63 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer,
(int)write_buffer_size, false);
- result = ACTION_TAKEN_WITH_CALLBACK;
+ if (t->use_packet_coalescing) {
+ if (!stream_op->send_trailing_metadata) {
+ CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+ bidirectional_stream_flush(s->cbs);
+ result = ACTION_TAKEN_WITH_CALLBACK;
+ } else {
+ stream_state->pending_write_for_trailer = true;
+ result = ACTION_TAKEN_NO_CALLBACK;
+ }
+ } else {
+ result = ACTION_TAKEN_WITH_CALLBACK;
+ }
} else {
result = NO_ACTION_POSSIBLE;
}
}
stream_state->state_op_done[OP_SEND_MESSAGE] = true;
oas->state.state_op_done[OP_SEND_MESSAGE] = true;
+ } else if (stream_op->send_trailing_metadata &&
+ op_can_be_run(stream_op, s, &oas->state,
+ OP_SEND_TRAILING_METADATA)) {
+ CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_TRAILING_METADATA", oas);
+ if (stream_state->state_callback_received[OP_FAILED]) {
+ result = NO_ACTION_POSSIBLE;
+ CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
+ } else {
+ CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
+ stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
+ bidirectional_stream_write(s->cbs, "", 0, true);
+ if (t->use_packet_coalescing) {
+ CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+ bidirectional_stream_flush(s->cbs);
+ }
+ result = ACTION_TAKEN_WITH_CALLBACK;
+ }
+ stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
+ } else if (stream_op->recv_initial_metadata &&
+ op_can_be_run(stream_op, s, &oas->state,
+ OP_RECV_INITIAL_METADATA)) {
+ CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_INITIAL_METADATA", oas);
+ if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
+ grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+ GRPC_ERROR_NONE);
+ } else if (stream_state->state_callback_received[OP_FAILED]) {
+ grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+ GRPC_ERROR_NONE);
+ } else {
+ grpc_chttp2_incoming_metadata_buffer_publish(
+ exec_ctx, &oas->s->state.rs.initial_metadata,
+ stream_op->recv_initial_metadata);
+ grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+ GRPC_ERROR_NONE);
+ }
+ stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
+ result = ACTION_TAKEN_NO_CALLBACK;
} else if (stream_op->recv_message &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_RECV_MESSAGE)) {
+ op_can_be_run(stream_op, s, &oas->state, OP_RECV_MESSAGE)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_MESSAGE", oas);
if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
CRONET_LOG(GPR_DEBUG, "Stream is cancelled.");
@@ -980,6 +1042,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_NONE);
stream_state->state_op_done[OP_RECV_MESSAGE] = true;
oas->state.state_op_done[OP_RECV_MESSAGE] = true;
+
+ /* Extra read to trigger on_succeed */
+ stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
+ stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
+ stream_state->rs.received_bytes = 0;
+ CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
+ stream_state->state_op_done[OP_READ_REQ_MADE] =
+ true; /* Indicates that at least one read request has been made */
+ bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer,
+ stream_state->rs.remaining_bytes);
result = ACTION_TAKEN_NO_CALLBACK;
}
} else if (stream_state->rs.remaining_bytes == 0) {
@@ -1027,7 +1099,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
result = ACTION_TAKEN_NO_CALLBACK;
}
} else if (stream_op->recv_trailing_metadata &&
- op_can_be_run(stream_op, stream_state, &oas->state,
+ op_can_be_run(stream_op, s, &oas->state,
OP_RECV_TRAILING_METADATA)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_TRAILING_METADATA", oas);
if (oas->s->state.rs.trailing_metadata_valid) {
@@ -1038,23 +1110,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
}
stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
result = ACTION_TAKEN_NO_CALLBACK;
- } else if (stream_op->send_trailing_metadata &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_SEND_TRAILING_METADATA)) {
- CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_TRAILING_METADATA", oas);
- if (stream_state->state_callback_received[OP_FAILED]) {
- result = NO_ACTION_POSSIBLE;
- CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
- } else {
- CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
- stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
- bidirectional_stream_write(s->cbs, "", 0, true);
- result = ACTION_TAKEN_WITH_CALLBACK;
- }
- stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
} else if (stream_op->cancel_error &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_CANCEL_ERROR)) {
+ op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_CANCEL_ERROR", oas);
CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs);
if (s->cbs) {
@@ -1068,8 +1125,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->cancel_error = GRPC_ERROR_REF(stream_op->cancel_error);
}
} else if (stream_op->on_complete &&
- op_can_be_run(stream_op, stream_state, &oas->state,
- OP_ON_COMPLETE)) {
+ op_can_be_run(stream_op, s, &oas->state, OP_ON_COMPLETE)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_ON_COMPLETE", oas);
if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
grpc_closure_sched(exec_ctx, stream_op->on_complete,
@@ -1133,6 +1189,12 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
sizeof(s->state.state_callback_received));
s->state.fail_state = s->state.flush_read = false;
s->state.cancel_error = NULL;
+ s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
+ s->state.unprocessed_send_message = false;
+
+ s->curr_gs = gs;
+ s->curr_ct = (grpc_cronet_transport *)gt;
+
gpr_mu_init(&s->mu);
return 0;
}
@@ -1148,8 +1210,6 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_transport_stream_op *op) {
CRONET_LOG(GPR_DEBUG, "perform_stream_op");
stream_obj *s = (stream_obj *)gs;
- s->curr_gs = gs;
- memcpy(&s->curr_ct, gt, sizeof(grpc_cronet_transport));
add_to_storage(s, op);
if (op->send_initial_metadata &&
header_has_authority(op->send_initial_metadata->list.head)) {
@@ -1197,14 +1257,58 @@ static grpc_endpoint *get_endpoint(grpc_exec_ctx *exec_ctx,
static void perform_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_transport_op *op) {}
-const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
- "cronet_http",
- init_stream,
- set_pollset_do_nothing,
- set_pollset_set_do_nothing,
- perform_stream_op,
- perform_op,
- destroy_stream,
- destroy_transport,
- get_peer,
- get_endpoint};
+static const grpc_transport_vtable grpc_cronet_vtable = {
+ sizeof(stream_obj),
+ "cronet_http",
+ init_stream,
+ set_pollset_do_nothing,
+ set_pollset_set_do_nothing,
+ perform_stream_op,
+ perform_op,
+ destroy_stream,
+ destroy_transport,
+ get_peer,
+ get_endpoint};
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+ const grpc_channel_args *args,
+ void *reserved) {
+ grpc_cronet_transport *ct = gpr_malloc(sizeof(grpc_cronet_transport));
+ if (!ct) {
+ goto error;
+ }
+ ct->base.vtable = &grpc_cronet_vtable;
+ ct->engine = engine;
+ ct->host = gpr_malloc(strlen(target) + 1);
+ if (!ct->host) {
+ goto error;
+ }
+ strcpy(ct->host, target);
+
+ ct->use_packet_coalescing = true;
+ if (args) {
+ for (size_t i = 0; i < args->num_args; i++) {
+ if (0 ==
+ strcmp(args->args[i].key, GRPC_ARG_USE_CRONET_PACKET_COALESCING)) {
+ if (args->args[i].type != GRPC_ARG_INTEGER) {
+ gpr_log(GPR_ERROR, "%s ignored: it must be an integer",
+ GRPC_ARG_USE_CRONET_PACKET_COALESCING);
+ } else {
+ ct->use_packet_coalescing = (args->args[i].value.integer != 0);
+ }
+ }
+ }
+ }
+
+ return &ct->base;
+
+error:
+ if (ct) {
+ if (ct->host) {
+ gpr_free(ct->host);
+ }
+ gpr_free(ct);
+ }
+
+ return NULL;
+}
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.h b/src/core/ext/transport/cronet/transport/cronet_transport.h
new file mode 100644
index 0000000000000000000000000000000000000000..169ce31fd7d64a78710c483f0ee60b2877ba6566
--- /dev/null
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.h
@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright 2016, 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.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+#define GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+
+#include "src/core/lib/transport/transport.h"
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+ const grpc_channel_args *args,
+ void *reserved);
+
+#endif /* GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H */
diff --git a/src/objective-c/tests/CronetUnitTests/CronetUnitTests.m b/src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
index 599f840567a2e0b2289bc6331d4da6d3d24b88e1..5e3c59f8b34fb51dfd9217879e81ffef3ce05fd3 100644
--- a/src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
+++ b/src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
@@ -32,13 +32,13 @@
*/
#import <XCTest/XCTest.h>
-#import <sys/socket.h>
#import <netinet/in.h>
+#import <sys/socket.h>
#import <Cronet/Cronet.h>
-#import <grpc/support/host_port.h>
-#import <grpc/grpc_cronet.h>
#import <grpc/grpc.h>
+#import <grpc/grpc_cronet.h>
+#import <grpc/support/host_port.h>
#import "test/core/end2end/cq_verifier.h"
#import "test/core/util/port.h"
@@ -49,16 +49,19 @@
#import "src/core/lib/support/env.h"
#import "src/core/lib/support/string.h"
#import "src/core/lib/support/tmpfile.h"
+#import "test/core/end2end/data/ssl_test_data.h"
#import "test/core/util/test_config.h"
+#import <BoringSSL/openssl/ssl.h>
+
static void drain_cq(grpc_completion_queue *cq) {
grpc_event ev;
do {
- ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5), NULL);
+ ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
+ NULL);
} while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
-
@interface CronetUnitTests : XCTestCase
@end
@@ -68,47 +71,99 @@ static void drain_cq(grpc_completion_queue *cq) {
+ (void)setUp {
[super setUp];
-/*** FILE *roots_file;
- size_t roots_size = strlen(test_root_cert);*/
-
char *argv[] = {"CoreCronetEnd2EndTests"};
grpc_test_init(1, argv);
grpc_init();
[Cronet setHttp2Enabled:YES];
+ [Cronet setSslKeyLogFileName:@"Documents/key"];
+ [Cronet enableTestCertVerifierForTesting];
NSURL *url = [[[NSFileManager defaultManager]
- URLsForDirectory:NSDocumentDirectory
- inDomains:NSUserDomainMask] lastObject];
+ URLsForDirectory:NSDocumentDirectory
+ inDomains:NSUserDomainMask] lastObject];
NSLog(@"Documents directory: %@", url);
[Cronet start];
[Cronet startNetLogToFile:@"Documents/cronet_netlog.json" logBytes:YES];
+
+ init_ssl();
}
+ (void)tearDown {
grpc_shutdown();
+ cleanup_ssl();
[super tearDown];
}
+void init_ssl(void) {
+ SSL_load_error_strings();
+ OpenSSL_add_ssl_algorithms();
+}
+
+void cleanup_ssl(void) { EVP_cleanup(); }
+
+int alpn_cb(SSL *ssl, const unsigned char **out, unsigned char *outlen,
+ const unsigned char *in, unsigned int inlen, void *arg) {
+ // Always select "h2" as the ALPN protocol to be used
+ *out = (const unsigned char *)"h2";
+ *outlen = 2;
+ return SSL_TLSEXT_ERR_OK;
+}
+
+void init_ctx(SSL_CTX *ctx) {
+ // Install server certificate
+ BIO *pem = BIO_new_mem_buf((void *)test_server1_cert,
+ (int)strlen(test_server1_cert));
+ X509 *cert = PEM_read_bio_X509_AUX(pem, NULL, NULL, "");
+ SSL_CTX_use_certificate(ctx, cert);
+ X509_free(cert);
+ BIO_free(pem);
+
+ // Install server private key
+ pem =
+ BIO_new_mem_buf((void *)test_server1_key, (int)strlen(test_server1_key));
+ EVP_PKEY *key = PEM_read_bio_PrivateKey(pem, NULL, NULL, "");
+ SSL_CTX_use_PrivateKey(ctx, key);
+ EVP_PKEY_free(key);
+ BIO_free(pem);
+
+ // Select cipher suite
+ SSL_CTX_set_cipher_list(ctx, "ECDHE-RSA-AES128-GCM-SHA256");
+
+ // Select ALPN protocol
+ SSL_CTX_set_alpn_select_cb(ctx, alpn_cb, NULL);
+}
+
+unsigned int parse_h2_length(const char *field) {
+ return ((unsigned int)(unsigned char)(field[0])) * 65536 +
+ ((unsigned int)(unsigned char)(field[1])) * 256 +
+ ((unsigned int)(unsigned char)(field[2]));
+}
+
- (void)testInternalError {
grpc_call *c;
grpc_slice request_payload_slice =
- grpc_slice_from_copied_string("hello world");
+ grpc_slice_from_copied_string("hello world");
grpc_byte_buffer *request_payload =
- grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
- grpc_metadata meta_c[2] = {
- {"key1", "val1", 4, 0, {{NULL, NULL, NULL, NULL}}},
- {"key2", "val2", 4, 0, {{NULL, NULL, NULL, NULL}}}};
+ grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
+ grpc_slice_from_static_string("val1"),
+ 0,
+ {{NULL, NULL, NULL, NULL}}},
+ {grpc_slice_from_static_string("key2"),
+ grpc_slice_from_static_string("val2"),
+ 0,
+ {{NULL, NULL, NULL, NULL}}}};
int port = grpc_pick_unused_port_or_die();
char *addr;
gpr_join_host_port(&addr, "127.0.0.1", port);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
- cronet_engine *cronetEngine = [Cronet getGlobalEngine];
- grpc_channel *client = grpc_cronet_secure_channel_create(cronetEngine, addr,
- NULL, NULL);
+ stream_engine *cronetEngine = [Cronet getGlobalEngine];
+ grpc_channel *client =
+ grpc_cronet_secure_channel_create(cronetEngine, addr, NULL, NULL);
cq_verifier *cqv = cq_verifier_create(cq);
grpc_op ops[6];
@@ -120,12 +175,11 @@ static void drain_cq(grpc_completion_queue *cq) {
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
- char *details = NULL;
- size_t details_capacity = 0;
+ grpc_slice details;
- c = grpc_channel_create_call(
- client, NULL, GRPC_PROPAGATE_DEFAULTS, cq, "/foo",
- NULL, deadline, NULL);
+ c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
+ grpc_slice_from_static_string("/foo"), NULL,
+ deadline, NULL);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
@@ -164,35 +218,40 @@ static void drain_cq(grpc_completion_queue *cq) {
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
- op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op->flags = 0;
op->reserved = NULL;
op++;
- error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void*)1, NULL);
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
- dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
- int sl = socket(AF_INET, SOCK_STREAM, 0);
- GPR_ASSERT(sl >= 0);
- struct sockaddr_in s_addr;
- memset(&s_addr, 0, sizeof(s_addr));
- s_addr.sin_family = AF_INET;
- s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
- s_addr.sin_port = htons(port);
- bind(sl, (struct sockaddr*)&s_addr, sizeof(s_addr));
- listen(sl, 5);
- int s = accept(sl, NULL, NULL);
- sleep(1);
- close(s);
- close(sl);
- });
-
- CQ_EXPECT_COMPLETION(cqv, (void*)1, 1);
+ dispatch_async(
+ dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
+ int sl = socket(AF_INET, SOCK_STREAM, 0);
+ GPR_ASSERT(sl >= 0);
+
+ // Make and TCP endpoint to accept the connection
+ struct sockaddr_in s_addr;
+ memset(&s_addr, 0, sizeof(s_addr));
+ s_addr.sin_family = AF_INET;
+ s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+ s_addr.sin_port = htons(port);
+ GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
+ GPR_ASSERT(0 == listen(sl, 5));
+ int s = accept(sl, NULL, NULL);
+ GPR_ASSERT(s >= 0);
+
+ // Close the connection after 1 second to trigger Cronet's on_failed()
+ sleep(1);
+ close(s);
+ close(sl);
+ });
+
+ CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
- gpr_free(details);
+ grpc_slice_unref(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
@@ -204,11 +263,195 @@ static void drain_cq(grpc_completion_queue *cq) {
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload_recv);
-
+
+ grpc_channel_destroy(client);
+ grpc_completion_queue_shutdown(cq);
+ drain_cq(cq);
+ grpc_completion_queue_destroy(cq);
+}
+
+- (void)packetCoalescing:(BOOL)useCoalescing {
+ grpc_arg arg;
+ arg.key = GRPC_ARG_USE_CRONET_PACKET_COALESCING;
+ arg.type = GRPC_ARG_INTEGER;
+ arg.value.integer = useCoalescing ? 1 : 0;
+ grpc_channel_args *args = grpc_channel_args_copy_and_add(NULL, &arg, 1);
+ grpc_call *c;
+ grpc_slice request_payload_slice =
+ grpc_slice_from_copied_string("hello world");
+ grpc_byte_buffer *request_payload =
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+ gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
+ grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
+ grpc_slice_from_static_string("val1"),
+ 0,
+ {{NULL, NULL, NULL, NULL}}},
+ {grpc_slice_from_static_string("key2"),
+ grpc_slice_from_static_string("val2"),
+ 0,
+ {{NULL, NULL, NULL, NULL}}}};
+
+ int port = grpc_pick_unused_port_or_die();
+ char *addr;
+ gpr_join_host_port(&addr, "127.0.0.1", port);
+ grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+ stream_engine *cronetEngine = [Cronet getGlobalEngine];
+ grpc_channel *client =
+ grpc_cronet_secure_channel_create(cronetEngine, addr, args, NULL);
+
+ cq_verifier *cqv = cq_verifier_create(cq);
+ grpc_op ops[6];
+ grpc_op *op;
+ grpc_metadata_array initial_metadata_recv;
+ grpc_metadata_array trailing_metadata_recv;
+ grpc_metadata_array request_metadata_recv;
+ grpc_byte_buffer *response_payload_recv = NULL;
+ grpc_call_details call_details;
+ grpc_status_code status;
+ grpc_call_error error;
+ grpc_slice details;
+
+ c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
+ grpc_slice_from_static_string("/foo"), NULL,
+ deadline, NULL);
+ GPR_ASSERT(c);
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+
+ memset(ops, 0, sizeof(ops));
+ op = ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 2;
+ op->data.send_initial_metadata.metadata = meta_c;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message.send_message = request_payload;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message.recv_message = &response_payload_recv;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+ op->data.recv_status_on_client.status = &status;
+ op->data.recv_status_on_client.status_details = &details;
+ op->flags = 0;
+ op->reserved = NULL;
+ op++;
+ error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
+ GPR_ASSERT(GRPC_CALL_OK == error);
+
+ __weak XCTestExpectation *expectation = [self expectationWithDescription:@"Coalescing"];
+
+ dispatch_async(
+ dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
+ int sl = socket(AF_INET, SOCK_STREAM, 0);
+ GPR_ASSERT(sl >= 0);
+ struct sockaddr_in s_addr;
+ memset(&s_addr, 0, sizeof(s_addr));
+ s_addr.sin_family = AF_INET;
+ s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+ s_addr.sin_port = htons(port);
+ GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
+ GPR_ASSERT(0 == listen(sl, 5));
+ int s = accept(sl, NULL, NULL);
+ GPR_ASSERT(s >= 0);
+ struct timeval tv;
+ tv.tv_sec = 2;
+ tv.tv_usec = 0;
+ setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+
+ // Make an TLS endpoint to receive Cronet's transmission
+ SSL_CTX *ctx = SSL_CTX_new(TLSv1_2_server_method());
+ init_ctx(ctx);
+ SSL *ssl = SSL_new(ctx);
+ SSL_set_fd(ssl, s);
+ SSL_accept(ssl);
+
+ const char magic[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
+
+ char buf[4096];
+ long len;
+ BOOL coalesced = NO;
+ while ((len = SSL_read(ssl, buf, sizeof(buf))) > 0) {
+ gpr_log(GPR_DEBUG, "Read len: %ld", len);
+
+ // Analyze the HTTP/2 frames in the same TLS PDU to identify if
+ // coalescing is successful
+ unsigned int p = 0;
+ while (p < len) {
+ if (len - p >= 24 && 0 == memcmp(&buf[p], magic, 24)) {
+ p += 24;
+ continue;
+ }
+
+ if (buf[p + 3] == 0 && // Type is DATA
+ parse_h2_length(&buf[p]) == 0x10 && // Length is correct
+ (buf[p + 4] & 1) != 0 && // EOS bit is set
+ 0 == memcmp("hello world", &buf[p + 14],
+ 11)) { // Message is correct
+ coalesced = YES;
+ break;
+ }
+ p += (parse_h2_length(&buf[p]) + 9);
+ }
+ if (coalesced) {
+ break;
+ }
+ }
+
+ XCTAssert(coalesced == useCoalescing);
+ SSL_free(ssl);
+ SSL_CTX_free(ctx);
+ close(s);
+ close(sl);
+ [expectation fulfill];
+ });
+
+ CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
+ cq_verify(cqv);
+
+ grpc_slice_unref(details);
+ grpc_metadata_array_destroy(&initial_metadata_recv);
+ grpc_metadata_array_destroy(&trailing_metadata_recv);
+ grpc_metadata_array_destroy(&request_metadata_recv);
+ grpc_call_details_destroy(&call_details);
+
+ grpc_call_destroy(c);
+
+ cq_verifier_destroy(cqv);
+
+ grpc_byte_buffer_destroy(request_payload);
+ grpc_byte_buffer_destroy(response_payload_recv);
+
grpc_channel_destroy(client);
grpc_completion_queue_shutdown(cq);
drain_cq(cq);
grpc_completion_queue_destroy(cq);
+
+ [self waitForExpectationsWithTimeout:4 handler:nil];
+}
+
+- (void)testPacketCoalescing {
+ [self packetCoalescing:YES];
+ [self packetCoalescing:NO];
}
@end
diff --git a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
index 4a6b332dfd1066a0359806a3d594c9f9817aefc7..32b35ef333abe476d1ba736c02c4153da65b2ec1 100644
--- a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
+++ b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
@@ -1474,6 +1474,7 @@
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_COMPILE_WITH_CRONET=1",
+ "GRPC_CRONET_WITH_PACKET_COALESCING=1",
);
INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;
diff --git a/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme b/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
index 5524a27ffde9ed627f4590e3a5375b5c0dbd6cc0..49dc3faa3d093223c0ee7b34f4d0337e85f61207 100644
--- a/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
+++ b/src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
@@ -59,6 +59,16 @@
ReferencedContainer = "container:Tests.xcodeproj">
</BuildableReference>
</TestableReference>
+ <TestableReference
+ skipped = "NO">
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "5EAD6D231E27047400002378"
+ BuildableName = "CronetUnitTests.xctest"
+ BlueprintName = "CronetUnitTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </TestableReference>
</Testables>
<MacroExpansion>
<BuildableReference
@@ -100,6 +110,15 @@
savedToolIdentifier = ""
useCustomWorkingDirectory = "NO"
debugDocumentVersioning = "YES">
+ <MacroExpansion>
+ <BuildableReference
+ BuildableIdentifier = "primary"
+ BlueprintIdentifier = "63423F431B150A5F006CF63C"
+ BuildableName = "AllTests.xctest"
+ BlueprintName = "AllTests"
+ ReferencedContainer = "container:Tests.xcodeproj">
+ </BuildableReference>
+ </MacroExpansion>
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">
diff --git a/templates/gRPC-Core.podspec.template b/templates/gRPC-Core.podspec.template
index 82b2ec2cf4ce5dd3492381efd2ad96221833cf5c..f05ee4c05f538bf9bf39d78916901d07902297ba 100644
--- a/templates/gRPC-Core.podspec.template
+++ b/templates/gRPC-Core.podspec.template
@@ -161,7 +161,8 @@
s.subspec 'Cronet-Interface' do |ss|
ss.header_mappings_dir = 'include/grpc'
- ss.source_files = 'include/grpc/grpc_cronet.h'
+ ss.source_files = 'include/grpc/grpc_cronet.h',
+ 'src/core/ext/transport/cronet/transport/cronet_transport.h'
end
s.subspec 'Cronet-Implementation' do |ss|
diff --git a/test/core/end2end/gen_build_yaml.py b/test/core/end2end/gen_build_yaml.py
index bcb7136eaa2d1b4bf63a09c557f85a7eea91cc8e..5071299545c12d9ce8c075ab04d83ee5e7b12564 100755
--- a/test/core/end2end/gen_build_yaml.py
+++ b/test/core/end2end/gen_build_yaml.py
@@ -91,6 +91,7 @@ LOWCPU = 0.1
# maps test names to options
END2END_TESTS = {
+ 'authority_not_supported': default_test_options,
'bad_hostname': default_test_options,
'binary_metadata': default_test_options,
'resource_quota_server': default_test_options._replace(large_writes=True,
@@ -142,7 +143,6 @@ END2END_TESTS = {
'simple_request': default_test_options,
'streaming_error_response': default_test_options,
'trailing_metadata': default_test_options,
- 'authority_not_supported': default_test_options,
'write_buffering': default_test_options,
'write_buffering_at_end': default_test_options,
}
diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json
index 0d5aeb233c4a3e5ad740edfbb159df4b6da38345..8756398003fe822f3fba37a97e30aeecc83a0133 100644
--- a/tools/run_tests/generated/sources_and_headers.json
+++ b/tools/run_tests/generated/sources_and_headers.json
@@ -8029,6 +8029,7 @@
"include/grpc/grpc_cronet.h",
"include/grpc/grpc_security.h",
"include/grpc/grpc_security_constants.h",
+ "src/core/ext/transport/cronet/transport/cronet_transport.h",
"third_party/objective_c/Cronet/bidirectional_stream_c.h"
],
"is_filegroup": true,
@@ -8040,7 +8041,8 @@
"include/grpc/grpc_security_constants.h",
"src/core/ext/transport/cronet/client/secure/cronet_channel_create.c",
"src/core/ext/transport/cronet/transport/cronet_api_dummy.c",
- "src/core/ext/transport/cronet/transport/cronet_transport.c"
+ "src/core/ext/transport/cronet/transport/cronet_transport.c",
+ "src/core/ext/transport/cronet/transport/cronet_transport.h"
],
"third_party": false,
"type": "filegroup"