diff --git a/test/core/fling/client.c b/test/core/fling/client.c
index a91dfba9b0e07a475b53520f8af7393465361608..62626996314713619be059b4167dcef2050502db 100644
--- a/test/core/fling/client.c
+++ b/test/core/fling/client.c
@@ -49,40 +49,77 @@ static grpc_byte_buffer *the_buffer;
static grpc_channel *channel;
static grpc_completion_queue *cq;
static grpc_call *call;
-
-static void init_ping_pong_request(void) {}
+static grpc_op ops[6];
+static grpc_op stream_init_op;
+static grpc_op stream_step_ops[2];
+static grpc_metadata_array initial_metadata_recv;
+static grpc_metadata_array trailing_metadata_recv;
+static grpc_byte_buffer *response_payload_recv = NULL;
+static grpc_call_details call_details;
+static grpc_status_code status;
+static char *details = NULL;
+static size_t details_capacity = 0;
+static grpc_op *op;
+
+static void init_ping_pong_request(void) {
+ grpc_metadata_array_init(&initial_metadata_recv);
+ grpc_metadata_array_init(&trailing_metadata_recv);
+ grpc_call_details_init(&call_details);
+
+ op = ops;
+
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ op->data.send_message = the_buffer;
+ op++;
+ op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ op++;
+ op->op = GRPC_OP_RECV_INITIAL_METADATA;
+ op->data.recv_initial_metadata = &initial_metadata_recv;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message = &response_payload_recv;
+ 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->data.recv_status_on_client.status_details_capacity = &details_capacity;
+ op++;
+}
static void step_ping_pong_request(void) {
- call = grpc_channel_create_call_old(channel, "/Reflector/reflectUnary",
- "localhost", gpr_inf_future);
- GPR_ASSERT(grpc_call_invoke_old(call, cq, (void *)1, (void *)1,
- GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
- GPR_ASSERT(grpc_call_start_write_old(call, the_buffer, (void *)1,
- GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
- GPR_ASSERT(grpc_call_start_read_old(call, (void *)1) == GRPC_CALL_OK);
- GPR_ASSERT(grpc_call_writes_done_old(call, (void *)1) == GRPC_CALL_OK);
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+ call = grpc_channel_create_call(channel, cq, "/Reflector/reflectUnary",
+ "localhost", gpr_inf_future);
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call, ops, op - ops, (void *)1));
grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
grpc_call_destroy(call);
call = NULL;
}
static void init_ping_pong_stream(void) {
- call = grpc_channel_create_call_old(channel, "/Reflector/reflectStream",
- "localhost", gpr_inf_future);
- GPR_ASSERT(grpc_call_invoke_old(call, cq, (void *)1, (void *)1, 0) ==
- GRPC_CALL_OK);
+ call = grpc_channel_create_call(channel, cq, "/Reflector/reflectStream",
+ "localhost", gpr_inf_future);
+ stream_init_op.op = GRPC_OP_SEND_INITIAL_METADATA;
+ stream_init_op.data.send_initial_metadata.count = 0;
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call, &stream_init_op, 1, (void *)1));
grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+
+ grpc_metadata_array_init(&initial_metadata_recv);
+
+ stream_step_ops[0].op = GRPC_OP_SEND_MESSAGE;
+ stream_step_ops[0].data.send_message = the_buffer;
+ stream_step_ops[1].op = GRPC_OP_RECV_MESSAGE;
+ stream_step_ops[1].data.recv_message = &response_payload_recv;
}
static void step_ping_pong_stream(void) {
- GPR_ASSERT(grpc_call_start_write_old(call, the_buffer, (void *)1, 0) ==
- GRPC_CALL_OK);
- GPR_ASSERT(grpc_call_start_read_old(call, (void *)1) == GRPC_CALL_OK);
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call, stream_step_ops, 2, (void *)1));
grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
}
@@ -99,7 +136,8 @@ typedef struct {
static const scenario scenarios[] = {
{"ping-pong-request", init_ping_pong_request, step_ping_pong_request},
- {"ping-pong-stream", init_ping_pong_stream, step_ping_pong_stream}, };
+ {"ping-pong-stream", init_ping_pong_stream, step_ping_pong_stream},
+};
int main(int argc, char **argv) {
gpr_slice slice = gpr_slice_from_copied_string("x");
@@ -148,6 +186,7 @@ int main(int argc, char **argv) {
cq = grpc_completion_queue_create();
the_buffer = grpc_byte_buffer_create(&slice, payload_size);
histogram = gpr_histogram_create(0.01, 60e9);
+
sc.init();
for (i = 0; i < 1000; i++) {
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index ba5e96ddc04b537518359bc3fcafa1372e5d9288..bc52059b38aae840dd793949d7e493accfd01bb8 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -52,17 +52,118 @@
static grpc_completion_queue *cq;
static grpc_server *server;
+static grpc_call *call;
+static grpc_call_details call_details;
+static grpc_metadata_array request_metadata_recv;
+static grpc_metadata_array initial_metadata_send;
+static grpc_byte_buffer *payload_buffer = NULL;
+/* Used to drain the terminal read in unary calls. */
+static grpc_byte_buffer *terminal_buffer = NULL;
+
+static grpc_op read_op;
+static grpc_op metadata_send_op;
+static grpc_op write_op;
+static grpc_op status_op[2];
+static int was_cancelled = 2;
+static grpc_op unary_ops[6];
static int got_sigint = 0;
+static void *tag(gpr_intptr t) { return (void *)t; }
+
+typedef enum {
+ FLING_SERVER_NEW_REQUEST = 1,
+ FLING_SERVER_READ_FOR_UNARY,
+ FLING_SERVER_BATCH_OPS_FOR_UNARY,
+ FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING,
+ FLING_SERVER_READ_FOR_STREAMING,
+ FLING_SERVER_WRITE_FOR_STREAMING,
+ FLING_SERVER_SEND_STATUS_FOR_STREAMING
+} fling_server_tags;
+
typedef struct {
gpr_refcount pending_ops;
gpr_uint32 flags;
} call_state;
static void request_call(void) {
- call_state *s = gpr_malloc(sizeof(call_state));
- gpr_ref_init(&s->pending_ops, 2);
- grpc_server_request_call_old(server, s);
+ grpc_metadata_array_init(&request_metadata_recv);
+ grpc_call_details_init(&call_details);
+ grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
+ cq, tag(FLING_SERVER_NEW_REQUEST));
+}
+
+static void handle_unary_method(void) {
+ grpc_op *op;
+
+ grpc_metadata_array_init(&initial_metadata_send);
+
+ op = unary_ops;
+ op->op = GRPC_OP_SEND_INITIAL_METADATA;
+ op->data.send_initial_metadata.count = 0;
+ op++;
+ op->op = GRPC_OP_RECV_MESSAGE;
+ op->data.recv_message = &terminal_buffer;
+ op++;
+ op->op = GRPC_OP_SEND_MESSAGE;
+ if (payload_buffer == NULL) {
+ gpr_log(GPR_INFO, "NULL payload buffer !!!");
+ }
+ op->data.send_message = payload_buffer;
+ op++;
+ op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ op->data.send_status_from_server.status = GRPC_STATUS_OK;
+ op->data.send_status_from_server.trailing_metadata_count = 0;
+ op->data.send_status_from_server.status_details = "";
+ op++;
+ op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ op->data.recv_close_on_server.cancelled = &was_cancelled;
+ op++;
+
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call, unary_ops, op - unary_ops,
+ tag(FLING_SERVER_BATCH_OPS_FOR_UNARY)));
+}
+
+static void send_initial_metadata(void) {
+ grpc_metadata_array_init(&initial_metadata_send);
+ metadata_send_op.op = GRPC_OP_SEND_INITIAL_METADATA;
+ metadata_send_op.data.send_initial_metadata.count = 0;
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(
+ call, &metadata_send_op, 1,
+ tag(FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING)));
+}
+
+static void start_read_op(int t) {
+ /* Starting read at server */
+ read_op.op = GRPC_OP_RECV_MESSAGE;
+ read_op.data.recv_message = &payload_buffer;
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call, &read_op, 1, tag(t)));
+}
+
+static void start_write_op(void) {
+ /* Starting write at server */
+ write_op.op = GRPC_OP_SEND_MESSAGE;
+ if (payload_buffer == NULL) {
+ gpr_log(GPR_INFO, "NULL payload buffer !!!");
+ }
+ write_op.data.send_message = payload_buffer;
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call, &write_op, 1,
+ tag(FLING_SERVER_WRITE_FOR_STREAMING)));
+}
+
+static void start_send_status(void) {
+ status_op[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ status_op[0].data.send_status_from_server.status = GRPC_STATUS_OK;
+ status_op[0].data.send_status_from_server.trailing_metadata_count = 0;
+ status_op[0].data.send_status_from_server.status_details = "";
+ status_op[1].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ status_op[1].data.recv_close_on_server.cancelled = &was_cancelled;
+
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(
+ call, status_op, 2,
+ tag(FLING_SERVER_SEND_STATUS_FOR_STREAMING)));
}
static void sigint_handler(int x) { got_sigint = 1; }
@@ -133,43 +234,66 @@ int main(int argc, char **argv) {
if (!ev) continue;
s = ev->tag;
switch (ev->type) {
- case GRPC_SERVER_RPC_NEW:
- if (ev->call != NULL) {
- /* initial ops are already started in request_call */
- if (0 == strcmp(ev->data.server_rpc_new.method,
- "/Reflector/reflectStream")) {
- s->flags = 0;
- } else {
- s->flags = GRPC_WRITE_BUFFER_HINT;
- }
- grpc_call_server_accept_old(ev->call, cq, s);
- grpc_call_server_end_initial_metadata_old(ev->call, s->flags);
- GPR_ASSERT(grpc_call_start_read_old(ev->call, s) == GRPC_CALL_OK);
- request_call();
- } else {
- GPR_ASSERT(shutdown_started);
- gpr_free(s);
+ case GRPC_OP_COMPLETE:
+ switch ((gpr_intptr)s) {
+ case FLING_SERVER_NEW_REQUEST:
+ if (call != NULL) {
+ if (0 ==
+ strcmp(call_details.method, "/Reflector/reflectStream")) {
+ /* Received streaming call. Send metadata here. */
+ start_read_op(FLING_SERVER_READ_FOR_STREAMING);
+ send_initial_metadata();
+ } else {
+ /* Received unary call. Can do all ops in one batch. */
+ start_read_op(FLING_SERVER_READ_FOR_UNARY);
+ }
+ } else {
+ GPR_ASSERT(shutdown_started);
+ }
+ /* request_call();
+ */
+ break;
+ case FLING_SERVER_READ_FOR_STREAMING:
+ if (payload_buffer != NULL) {
+ /* Received payload from client. */
+ start_write_op();
+ } else {
+ /* Received end of stream from client. */
+ start_send_status();
+ }
+ break;
+ case FLING_SERVER_WRITE_FOR_STREAMING:
+ /* Write completed at server */
+ start_read_op(FLING_SERVER_READ_FOR_STREAMING);
+ break;
+ case FLING_SERVER_SEND_INIT_METADATA_FOR_STREAMING:
+ /* Metadata send completed at server */
+ break;
+ case FLING_SERVER_SEND_STATUS_FOR_STREAMING:
+ /* Send status and close completed at server */
+ grpc_call_destroy(call);
+ request_call();
+ break;
+ case FLING_SERVER_READ_FOR_UNARY:
+ /* Finished payload read for unary. Start all reamaining
+ * unary ops in a batch.
+ */
+ handle_unary_method();
+ break;
+ case FLING_SERVER_BATCH_OPS_FOR_UNARY:
+ /* Finished unary call. */
+ grpc_call_destroy(call);
+ request_call();
+ break;
}
break;
+ case GRPC_SERVER_RPC_NEW:
case GRPC_WRITE_ACCEPTED:
- GPR_ASSERT(ev->data.write_accepted == GRPC_OP_OK);
- GPR_ASSERT(grpc_call_start_read_old(ev->call, s) == GRPC_CALL_OK);
- break;
case GRPC_READ:
- if (ev->data.read) {
- GPR_ASSERT(grpc_call_start_write_old(ev->call, ev->data.read, s,
- s->flags) == GRPC_CALL_OK);
- } else {
- GPR_ASSERT(grpc_call_start_write_status_old(ev->call, GRPC_STATUS_OK,
- NULL, s) == GRPC_CALL_OK);
- }
- break;
case GRPC_FINISH_ACCEPTED:
case GRPC_FINISHED:
- if (gpr_unref(&s->pending_ops)) {
- grpc_call_destroy(ev->call);
- gpr_free(s);
- }
+ gpr_log(GPR_ERROR, "Unexpected event type.");
+ abort();
break;
case GRPC_QUEUE_SHUTDOWN:
GPR_ASSERT(shutdown_started);