diff --git a/test/core/fling/server.c b/test/core/fling/server.c index ba5e96ddc04b537518359bc3fcafa1372e5d9288..fae4cc89871662a2590b641c8595e1948418a94a 100644 --- a/test/core/fling/server.c +++ b/test/core/fling/server.c @@ -52,17 +52,77 @@ 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; + + +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 int got_sigint = 0; +static void *tag(gpr_intptr t) { return (void *)t; } + 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(101)); +} + +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(3))); +} + +static void start_read_op(void) { + /* 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(1))); +} + +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(2))); +} + + +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(4))); } static void sigint_handler(int x) { got_sigint = 1; } @@ -133,6 +193,49 @@ int main(int argc, char **argv) { if (!ev) continue; s = ev->tag; switch (ev->type) { + case GRPC_OP_COMPLETE: + switch ((gpr_intptr) s) { + case 101: + if(call != NULL) { + if (0 == strcmp(call_details.method, + "/Reflector/reflectStream")) { + /* Received streaming call. Send metadata here. */ + } else { + /* Received unary call. Can do all ops in one batch. */ + } + start_read_op(); + send_initial_metadata(); + } + else { + GPR_ASSERT(shutdown_started); + } + /* request_call(); + */ + break; + case 1: + if (payload_buffer != NULL) { + /* Received payload from client. */ + start_write_op(); + } + else { + /* Received end of stream from client. */ + start_send_status(); + } + break; + case 2: + /* Write completed at server */ + start_read_op(); + break; + case 3: + /* Metadata send completed at server */ + break; + case 4: + /* Send status and close completed at server */ + grpc_call_destroy(call); + request_call(); + break; + } + break; case GRPC_SERVER_RPC_NEW: if (ev->call != NULL) { /* initial ops are already started in request_call */