@ -36,41 +36,29 @@
static const size_t kMaxPayloadSizeForGet = 2048 ; static const size_t kMaxPayloadSizeForGet = 2048 ;
typedef struct call_data { typedef struct call_data {
// State for handling send_initial_metadata ops.
grpc_linked_mdelem method ; grpc_linked_mdelem method ;
grpc_linked_mdelem scheme ; grpc_linked_mdelem scheme ;
grpc_linked_mdelem authority ; grpc_linked_mdelem authority ;
grpc_linked_mdelem te_trailers ; grpc_linked_mdelem te_trailers ;
grpc_linked_mdelem content_type ; grpc_linked_mdelem content_type ;
grpc_linked_mdelem user_agent ; grpc_linked_mdelem user_agent ;
// State for handling recv_initial_metadata ops.
grpc_metadata_batch * recv_initial_metadata ; grpc_metadata_batch * recv_initial_metadata ;
grpc_closure * original_recv_initial_metadata_ready ;
grpc_closure recv_initial_metadata_ready ;
// State for handling recv_trailing_metadata ops.
grpc_metadata_batch * recv_trailing_metadata ; grpc_metadata_batch * recv_trailing_metadata ;
uint8_t * payload_bytes ; grpc_closure * original_recv_trailing_metadata_on_complete ;
grpc_closure recv_trailing_metadata_on_complete ;
/* Vars to read data off of send_message */ // State for handling send_message ops.
grpc_transport_stream_op_batch * send_op ; grpc_transport_stream_op_batch * send_message_batch ;
uint32_t send_length ; size_t send_message_bytes_read ;
uint32_t send_flags ; grpc_byte_stream_cache send_message_cache ;
grpc_slice incoming_slice ; grpc_caching_byte_stream send_message_caching_stream ;
grpc_slice_buffer_stream replacement_stream ; grpc_closure on_send_message_next_done ;
grpc_slice_buffer slices ; grpc_closure * original_send_message_on_complete ;
/* flag that indicates that all slices of send_messages aren't availble */ grpc_closure send_message_on_complete ;
bool send_message_blocked ;
/** Closure to call when finished with the hc_on_recv hook */
grpc_closure * on_done_recv_initial_metadata ;
grpc_closure * on_done_recv_trailing_metadata ;
grpc_closure * on_complete ;
grpc_closure * post_send ;
/** Receive closures are chained: we inject this closure as the on_done_recv
up - call on transport_op , and remember to call our on_done_recv member
after handling it . */
grpc_closure hc_on_recv_initial_metadata ;
grpc_closure hc_on_recv_trailing_metadata ;
grpc_closure hc_on_complete ;
grpc_closure got_slice ;
grpc_closure send_done ;
} call_data ; } call_data ;
typedef struct channel_data { typedef struct channel_data {
@ -148,7 +136,7 @@ static grpc_error *client_filter_incoming_metadata(grpc_exec_ctx *exec_ctx,
return GRPC_ERROR_NONE ; return GRPC_ERROR_NONE ;
} }
static void hc_on_ recv_initial_metadata( grpc_exec_ctx * exec_ctx , static void recv_initial_metadata_ready ( grpc_exec_ctx * exec_ctx ,
void * user_data , grpc_error * error ) { void * user_data , grpc_error * error ) {
grpc_call_element * elem = user_data ; grpc_call_element * elem = user_data ;
call_data * calld = elem - > call_data ; call_data * calld = elem - > call_data ;
@ -158,11 +146,13 @@ static void hc_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx,
} else { } else {
GRPC_ERROR_REF ( error ) ; GRPC_ERROR_REF ( error ) ;
} }
GRPC_CLOSURE_RUN ( exec_ctx , calld - > on_done_recv_initial_metadata , error ) ; GRPC_CLOSURE_RUN ( exec_ctx , calld - > original_recv_initial_metadata_ready ,
error ) ;
} }
static void hc_on_recv_trailing_metadata ( grpc_exec_ctx * exec_ctx , static void recv_trailing_metadata_on_complete ( grpc_exec_ctx * exec_ctx ,
void * user_data , grpc_error * error ) { void * user_data ,
grpc_error * error ) {
grpc_call_element * elem = user_data ; grpc_call_element * elem = user_data ;
call_data * calld = elem - > call_data ; call_data * calld = elem - > call_data ;
if ( error = = GRPC_ERROR_NONE ) { if ( error = = GRPC_ERROR_NONE ) {
@ -171,302 +161,288 @@ static void hc_on_recv_trailing_metadata(grpc_exec_ctx *exec_ctx,
} else { } else {
GRPC_ERROR_REF ( error ) ; GRPC_ERROR_REF ( error ) ;
} }
GRPC_CLOSURE_RUN ( exec_ctx , calld - > on_done_recv_trailing_metadata , error ) ; GRPC_CLOSURE_RUN ( exec_ctx , calld - > original_recv_trailing_metadata_on_complete ,
error ) ;
} }
static void hc _on_complete( grpc_exec_ctx * exec_ctx , void * user_data , static void send_message _on_complete( grpc_exec_ctx * exec_ctx , void * arg ,
grpc_error * error ) { grpc_error * error ) {
grpc_call_element * elem = user_data ; grpc_call_element * elem = ( grpc_call_element * ) arg ;
call_data * calld = elem - > call_data ; call_data * calld = ( call_data * ) elem - > call_data ;
if ( calld - > payload_bytes ) { grpc_byte_stream_cache_destroy ( exec_ctx , & calld - > send_message_cache ) ;
gpr_free ( calld - > payload_bytes ) ; GRPC_CLOSURE_RUN ( exec_ctx , calld - > original_send_message_on_complete ,
calld - > payload_bytes = NULL ; GRPC_ERROR_REF ( error ) ) ;
}
calld - > on_complete - > cb ( exec_ctx , calld - > on_complete - > cb_arg , error ) ;
}
static void send_done ( grpc_exec_ctx * exec_ctx , void * elemp , grpc_error * error ) {
grpc_call_element * elem = elemp ;
call_data * calld = elem - > call_data ;
grpc_slice_buffer_reset_and_unref_internal ( exec_ctx , & calld - > slices ) ;
calld - > post_send - > cb ( exec_ctx , calld - > post_send - > cb_arg , error ) ;
}
static void remove_if_present ( grpc_exec_ctx * exec_ctx ,
grpc_metadata_batch * batch ,
grpc_metadata_batch_callouts_index idx ) {
if ( batch - > idx . array [ idx ] ! = NULL ) {
grpc_metadata_batch_remove ( exec_ctx , batch , batch - > idx . array [ idx ] ) ;
}
} }
static void continue_send_message ( grpc_exec_ctx * exec_ctx , // Pulls a slice from the send_message byte stream, updating
grpc_call_element * elem ) { // calld->send_message_bytes_read.
call_data * calld = elem - > call_data ; static grpc_error * pull_slice_from_send_message ( grpc_exec_ctx * exec_ctx ,
uint8_t * wrptr = calld - > payload_bytes ; call_data * calld ) {
while ( grpc_byte_stream_next ( grpc_slice incoming_slice ;
exec_ctx , calld - > send_op - > payload - > send_message . send_message , ~ ( size_t ) 0 , grpc_error * error = grpc_byte_stream_pull (
& calld - > got_slice ) ) { exec_ctx , & calld - > send_message_caching_stream . base , & incoming_slice ) ;
grpc_byte_stream_pull ( exec_ctx , if ( error = = GRPC_ERROR_NONE ) {
calld - > send_op - > payload - > send_message . send_message , calld - > send_message_bytes_read + = GRPC_SLICE_LENGTH ( incoming_slice ) ;
& calld - > incoming_slice ) ; grpc_slice_unref_internal ( exec_ctx , incoming_slice ) ;
if ( GRPC_SLICE_LENGTH ( calld - > incoming_slice ) > 0 ) { }
memcpy ( wrptr , GRPC_SLICE_START_PTR ( calld - > incoming_slice ) , return error ;
GRPC_SLICE_LENGTH ( calld - > incoming_slice ) ) ; }
}
wrptr + = GRPC_SLICE_LENGTH ( calld - > incoming_slice ) ; // Reads as many slices as possible from the send_message byte stream.
grpc_slice_buffer_add ( & calld - > slices , calld - > incoming_slice ) ; // Upon successful return, if calld->send_message_bytes_read ==
if ( calld - > send_length = = calld - > slices . length ) { // calld->send_message_caching_stream.base.length, then we have completed
calld - > send_message_blocked = false ; // reading from the byte stream; otherwise, an async read has been dispatched
// and on_send_message_next_done() will be invoked when it is complete.
static grpc_error * read_all_available_send_message_data ( grpc_exec_ctx * exec_ctx ,
call_data * calld ) {
while ( grpc_byte_stream_next ( exec_ctx ,
& calld - > send_message_caching_stream . base ,
~ ( size_t ) 0 , & calld - > on_send_message_next_done ) ) {
grpc_error * error = pull_slice_from_send_message ( exec_ctx , calld ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ;
if ( calld - > send_message_bytes_read = =
calld - > send_message_caching_stream . base . length ) {
break ; break ;
} }
} }
return GRPC_ERROR_NONE ;
} }
static void got_slice ( grpc_exec_ctx * exec_ctx , void * elemp , grpc_error * error ) { // Async callback for grpc_byte_stream_next().
grpc_call_element * elem = elemp ; static void on_send_message_next_done ( grpc_exec_ctx * exec_ctx , void * arg ,
call_data * calld = elem - > call_data ; grpc_error * error ) {
calld - > send_message_blocked = false ; grpc_call_element * elem = ( grpc_call_element * ) arg ;
if ( GRPC_ERROR_NONE ! = call_data * calld = ( call_data * ) elem - > call_data ;
grpc_byte_stream_pull ( exec_ctx , if ( error ! = GRPC_ERROR_NONE ) {
calld - > send_op - > payload - > send_message . send_message , grpc_transport_stream_op_batch_finish_with_failure (
& calld - > incoming_slice ) ) { exec_ctx , calld - > send_message_batch , error ) ;
/* Should never reach here */ return ;
abort ( ) ;
}
grpc_slice_buffer_add ( & calld - > slices , calld - > incoming_slice ) ;
if ( calld - > send_length = = calld - > slices . length ) {
/* Pass down the original send_message op that was blocked.*/
grpc_slice_buffer_stream_init ( & calld - > replacement_stream , & calld - > slices ,
calld - > send_flags ) ;
calld - > send_op - > payload - > send_message . send_message =
& calld - > replacement_stream . base ;
calld - > post_send = calld - > send_op - > on_complete ;
calld - > send_op - > on_complete = & calld - > send_done ;
grpc_call_next_op ( exec_ctx , elem , calld - > send_op ) ;
} else {
continue_send_message ( exec_ctx , elem ) ;
}
} }
error = pull_slice_from_send_message ( exec_ctx , calld ) ;
static grpc_error * hc_mutate_op ( grpc_exec_ctx * exec_ctx , if ( error ! = GRPC_ERROR_NONE ) {
grpc_transport_stream_op_batch_finish_with_failure (
exec_ctx , calld - > send_message_batch , error ) ;
return ;
}
// There may or may not be more to read, but we don't care. If we got
// here, then we know that all of the data was not available
// synchronously, so we were not able to do a cached call. Instead,
// we just reset the byte stream and then send down the batch as-is.
grpc_caching_byte_stream_reset ( & calld - > send_message_caching_stream ) ;
grpc_call_next_op ( exec_ctx , elem , calld - > send_message_batch ) ;
}
static char * slice_buffer_to_string ( grpc_slice_buffer * slice_buffer ) {
char * payload_bytes = gpr_malloc ( slice_buffer - > length + 1 ) ;
size_t offset = 0 ;
for ( size_t i = 0 ; i < slice_buffer - > count ; + + i ) {
memcpy ( payload_bytes + offset ,
GRPC_SLICE_START_PTR ( slice_buffer - > slices [ i ] ) ,
GRPC_SLICE_LENGTH ( slice_buffer - > slices [ i ] ) ) ;
offset + = GRPC_SLICE_LENGTH ( slice_buffer - > slices [ i ] ) ;
}
* ( payload_bytes + offset ) = ' \0 ' ;
return payload_bytes ;
}
// Modifies the path entry in the batch's send_initial_metadata to
// append the base64-encoded query for a GET request.
static grpc_error * update_path_for_get ( grpc_exec_ctx * exec_ctx ,
grpc_call_element * elem , grpc_call_element * elem ,
grpc_transport_stream_op_batch * op ) { grpc_transport_stream_op_batch * batch ) {
/* grab pointers to our data from the call element */ call_data * calld = ( call_data * ) elem - > call_data ;
call_data * calld = elem - > call_data ;
channel_data * channeld = elem - > channel_data ;
grpc_error * error ;
if ( op - > send_initial_metadata ) {
/* Decide which HTTP VERB to use. We use GET if the request is marked
cacheable , and the operation contains both initial metadata and send
message , and the payload is below the size threshold , and all the data
for this request is immediately available . */
grpc_mdelem method = GRPC_MDELEM_METHOD_POST ;
if ( op - > send_message & &
( op - > payload - > send_initial_metadata . send_initial_metadata_flags &
GRPC_INITIAL_METADATA_CACHEABLE_REQUEST ) & &
op - > payload - > send_message . send_message - > length <
channeld - > max_payload_size_for_get ) {
method = GRPC_MDELEM_METHOD_GET ;
/* The following write to calld->send_message_blocked isn't racy with
reads in hc_start_transport_op ( which deals with SEND_MESSAGE ops ) because
being here means ops - > send_message is not NULL , which is primarily
guarding the read there . */
calld - > send_message_blocked = true ;
} else if ( op - > payload - > send_initial_metadata . send_initial_metadata_flags &
GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST ) {
method = GRPC_MDELEM_METHOD_PUT ;
}
/* Attempt to read the data from send_message and create a header field. */
if ( grpc_mdelem_eq ( method , GRPC_MDELEM_METHOD_GET ) ) {
/* allocate memory to hold the entire payload */
calld - > payload_bytes =
gpr_malloc ( op - > payload - > send_message . send_message - > length ) ;
/* read slices of send_message and copy into payload_bytes */
calld - > send_op = op ;
calld - > send_length = op - > payload - > send_message . send_message - > length ;
calld - > send_flags = op - > payload - > send_message . send_message - > flags ;
continue_send_message ( exec_ctx , elem ) ;
if ( calld - > send_message_blocked = = false ) {
/* when all the send_message data is available, then modify the path
* MDELEM by appending base64 encoded query to the path */
const int k_url_safe = 1 ;
const int k_multi_line = 0 ;
const unsigned char k_query_separator = ' ? ' ;
grpc_slice path_slice = grpc_slice path_slice =
GRPC_MDVALUE ( op - > payload - > send_initial_metadata GRPC_MDVALUE ( batch - > payload - > send_initial_metadata . send_initial_metadata
. send_initial_metadata - > idx . named . path - > md ) ; - > idx . named . path - > md ) ;
/* sum up individual component's lengths and allocate enough memory to
/* sum up individual component's lengths and allocate enough memory to
* hold combined path + query */ * hold combined path + query */
size_t estimated_len = GRPC_SLICE_LENGTH ( path_slice ) ; size_t estimated_len = GRPC_SLICE_LENGTH ( path_slice ) ;
estimated_len + + ; /* for the '?' */ estimated_len + + ; /* for the '?' */
estimated_len + = grpc_base64_estimate_encoded_size ( estimated_len + = grpc_base64_estimate_encoded_size (
op - > payload - > send_message . send_message - > length , k_url_safe , batch - > payload - > send_message . send_message - > length , true /* url_safe */ ,
k_multi_line ) ; false /* multi_line */ ) ;
grpc_slice path_with_query_slice = GRPC_SLICE_MALLOC ( estimated_len ) ; grpc_slice path_with_query_slice = GRPC_SLICE_MALLOC ( estimated_len ) ;
/* memcopy individual pieces into this slice */ /* memcopy individual pieces into this slice */
uint8_t * write_ptr = char * write_ptr = ( char * ) GRPC_SLICE_START_PTR ( path_with_query_slice ) ;
( uint8_t * ) GRPC_SLICE_START_PTR ( path_with_query_slice ) ; char * original_path = ( char * ) GRPC_SLICE_START_PTR ( path_slice ) ;
uint8_t * original_path = ( uint8_t * ) GRPC_SLICE_START_PTR ( path_slice ) ;
memcpy ( write_ptr , original_path , GRPC_SLICE_LENGTH ( path_slice ) ) ; memcpy ( write_ptr , original_path , GRPC_SLICE_LENGTH ( path_slice ) ) ;
write_ptr + = GRPC_SLICE_LENGTH ( path_slice ) ; write_ptr + = GRPC_SLICE_LENGTH ( path_slice ) ;
* write_ptr + + = ' ? ' ;
* write_ptr = k_query_separator ; char * payload_bytes =
write_ptr + + ; /* for the '?' */ slice_buffer_to_string ( & calld - > send_message_cache . cache_buffer ) ;
grpc_base64_encode_core ( ( char * ) write_ptr , payload_bytes ,
grpc_base64_encode_core ( ( char * ) write_ptr , calld - > payload_bytes , batch - > payload - > send_message . send_message - > length ,
op - > payload - > send_message . send_message - > length , true /* url_safe */ , false /* multi_line */ ) ;
k_url_safe , k_multi_line ) ; gpr_free ( payload_bytes ) ;
/* remove trailing unused memory and add trailing 0 to terminate string */
/* remove trailing unused memory and add trailing 0 to terminate string
*/
char * t = ( char * ) GRPC_SLICE_START_PTR ( path_with_query_slice ) ; char * t = ( char * ) GRPC_SLICE_START_PTR ( path_with_query_slice ) ;
/* safe to use strlen since base64_encode will always add '\0' */ /* safe to use strlen since base64_encode will always add '\0' */
path_with_query_slice = path_with_query_slice =
grpc_slice_sub_no_ref ( path_with_query_slice , 0 , strlen ( t ) ) ; grpc_slice_sub_no_ref ( path_with_query_slice , 0 , strlen ( t ) ) ;
/* substitute previous path with the new path+query */ /* substitute previous path with the new path+query */
grpc_mdelem mdelem_path_and_query = grpc_mdelem_from_slices ( grpc_mdelem mdelem_path_and_query =
exec_ctx , GRPC_MDSTR_PATH , path_with_query_slice ) ; grpc_mdelem_from_slices ( exec_ctx , GRPC_MDSTR_PATH , path_with_query_slice ) ;
grpc_metadata_batch * b = grpc_metadata_batch * b =
op - > payload - > send_initial_metadata . send_initial_metadata ; batch - > payload - > send_initial_metadata . send_initial_metadata ;
error = grpc_metadata_batch_substitute ( exec_ctx , b , b - > idx . named . path , return grpc_metadata_batch_substitute ( exec_ctx , b , b - > idx . named . path ,
mdelem_path_and_query ) ; mdelem_path_and_query ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; }
static void remove_if_present ( grpc_exec_ctx * exec_ctx ,
grpc_metadata_batch * batch ,
grpc_metadata_batch_callouts_index idx ) {
if ( batch - > idx . array [ idx ] ! = NULL ) {
grpc_metadata_batch_remove ( exec_ctx , batch , batch - > idx . array [ idx ] ) ;
}
}
static void hc_start_transport_stream_op_batch (
grpc_exec_ctx * exec_ctx , grpc_call_element * elem ,
grpc_transport_stream_op_batch * batch ) {
call_data * calld = elem - > call_data ;
channel_data * channeld = elem - > channel_data ;
GPR_TIMER_BEGIN ( " hc_start_transport_stream_op_batch " , 0 ) ;
GRPC_CALL_LOG_OP ( GPR_INFO , elem , batch ) ;
if ( batch - > recv_initial_metadata ) {
/* substitute our callback for the higher callback */
calld - > recv_initial_metadata =
batch - > payload - > recv_initial_metadata . recv_initial_metadata ;
calld - > original_recv_initial_metadata_ready =
batch - > payload - > recv_initial_metadata . recv_initial_metadata_ready ;
batch - > payload - > recv_initial_metadata . recv_initial_metadata_ready =
& calld - > recv_initial_metadata_ready ;
}
calld - > on_complete = op - > on_complete ; if ( batch - > recv_trailing_metadata ) {
op - > on_complete = & calld - > hc_on_complete ; /* substitute our callback for the higher callback */
op - > send_message = false ; calld - > recv_trailing_metadata =
batch - > payload - > recv_trailing_metadata . recv_trailing_metadata ;
calld - > original_recv_trailing_metadata_on_complete = batch - > on_complete ;
batch - > on_complete = & calld - > recv_trailing_metadata_on_complete ;
}
grpc_error * error = GRPC_ERROR_NONE ;
bool batch_will_be_handled_asynchronously = false ;
if ( batch - > send_initial_metadata ) {
// Decide which HTTP VERB to use. We use GET if the request is marked
// cacheable, and the operation contains both initial metadata and send
// message, and the payload is below the size threshold, and all the data
// for this request is immediately available.
grpc_mdelem method = GRPC_MDELEM_METHOD_POST ;
if ( batch - > send_message & &
( batch - > payload - > send_initial_metadata . send_initial_metadata_flags &
GRPC_INITIAL_METADATA_CACHEABLE_REQUEST ) & &
batch - > payload - > send_message . send_message - > length <
channeld - > max_payload_size_for_get ) {
calld - > send_message_bytes_read = 0 ;
grpc_byte_stream_cache_init ( & calld - > send_message_cache ,
batch - > payload - > send_message . send_message ) ;
grpc_caching_byte_stream_init ( & calld - > send_message_caching_stream ,
& calld - > send_message_cache ) ;
batch - > payload - > send_message . send_message =
& calld - > send_message_caching_stream . base ;
calld - > original_send_message_on_complete = batch - > on_complete ;
batch - > on_complete = & calld - > send_message_on_complete ;
calld - > send_message_batch = batch ;
error = read_all_available_send_message_data ( exec_ctx , calld ) ;
if ( error ! = GRPC_ERROR_NONE ) goto done ;
// If all the data has been read, then we can use GET.
if ( calld - > send_message_bytes_read = =
calld - > send_message_caching_stream . base . length ) {
method = GRPC_MDELEM_METHOD_GET ;
error = update_path_for_get ( exec_ctx , elem , batch ) ;
if ( error ! = GRPC_ERROR_NONE ) goto done ;
batch - > send_message = false ;
grpc_byte_stream_destroy ( exec_ctx ,
& calld - > send_message_caching_stream . base ) ;
} else { } else {
/* Not all data is available. Fall back to POST. */ // Not all data is available. The batch will be sent down
// asynchronously in on_send_message_next_done().
batch_will_be_handled_asynchronously = true ;
// Fall back to POST.
gpr_log ( GPR_DEBUG , gpr_log ( GPR_DEBUG ,
" Request is marked Cacheable but not all data is available. \
" Request is marked Cacheable but not all data is available. "
Falling back to POST " ); " Falling back to POST " ) ;
method = GRPC_MDELEM_METHOD_POST ;
} }
} else if ( batch - > payload - > send_initial_metadata
. send_initial_metadata_flags &
GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST ) {
method = GRPC_MDELEM_METHOD_PUT ;
} }
remove_if_present ( exec_ctx , remove_if_present (
op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
GRPC_BATCH_METHOD ) ; GRPC_BATCH_METHOD ) ;
remove_if_present ( exec_ctx , remove_if_present (
op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
GRPC_BATCH_SCHEME ) ; GRPC_BATCH_SCHEME ) ;
remove_if_present ( exec_ctx , remove_if_present (
op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
GRPC_BATCH_TE ) ; GRPC_BATCH_TE ) ;
remove_if_present ( exec_ctx , remove_if_present (
op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
GRPC_BATCH_CONTENT_TYPE ) ; GRPC_BATCH_CONTENT_TYPE ) ;
remove_if_present ( exec_ctx , remove_if_present (
op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
GRPC_BATCH_USER_AGENT ) ; GRPC_BATCH_USER_AGENT ) ;
/* Send : prefixed headers, which have to be before any application
/* Send : prefixed headers, which have to be before any application
layer headers . */ layer headers . */
error = grpc_metadata_batch_add_head ( error = grpc_metadata_batch_add_head (
exec_ctx , op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
& calld - > method , method ) ; & calld - > method , method ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; if ( error ! = GRPC_ERROR_NONE ) goto done ;
error = grpc_metadata_batch_add_head ( error = grpc_metadata_batch_add_head (
exec_ctx , op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
& calld - > scheme , channeld - > static_scheme ) ; & calld - > scheme , channeld - > static_scheme ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; if ( error ! = GRPC_ERROR_NONE ) goto done ;
error = grpc_metadata_batch_add_tail ( error = grpc_metadata_batch_add_tail (
exec_ctx , op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
& calld - > te_trailers , GRPC_MDELEM_TE_TRAILERS ) ; & calld - > te_trailers , GRPC_MDELEM_TE_TRAILERS ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; if ( error ! = GRPC_ERROR_NONE ) goto done ;
error = grpc_metadata_batch_add_tail ( error = grpc_metadata_batch_add_tail (
exec_ctx , op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
& calld - > content_type , GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC ) ; & calld - > content_type , GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; if ( error ! = GRPC_ERROR_NONE ) goto done ;
error = grpc_metadata_batch_add_tail ( error = grpc_metadata_batch_add_tail (
exec_ctx , op - > payload - > send_initial_metadata . send_initial_metadata , exec_ctx , batch - > payload - > send_initial_metadata . send_initial_metadata ,
& calld - > user_agent , GRPC_MDELEM_REF ( channeld - > user_agent ) ) ; & calld - > user_agent , GRPC_MDELEM_REF ( channeld - > user_agent ) ) ;
if ( error ! = GRPC_ERROR_NONE ) return error ; if ( error ! = GRPC_ERROR_NONE ) goto done ;
}
if ( op - > recv_initial_metadata ) {
/* substitute our callback for the higher callback */
calld - > recv_initial_metadata =
op - > payload - > recv_initial_metadata . recv_initial_metadata ;
calld - > on_done_recv_initial_metadata =
op - > payload - > recv_initial_metadata . recv_initial_metadata_ready ;
op - > payload - > recv_initial_metadata . recv_initial_metadata_ready =
& calld - > hc_on_recv_initial_metadata ;
}
if ( op - > recv_trailing_metadata ) {
/* substitute our callback for the higher callback */
calld - > recv_trailing_metadata =
op - > payload - > recv_trailing_metadata . recv_trailing_metadata ;
calld - > on_done_recv_trailing_metadata = op - > on_complete ;
op - > on_complete = & calld - > hc_on_recv_trailing_metadata ;
}
return GRPC_ERROR_NONE ;
} }
static void hc_start_transport_op ( grpc_exec_ctx * exec_ctx , done :
grpc_call_element * elem ,
grpc_transport_stream_op_batch * op ) {
GPR_TIMER_BEGIN ( " hc_start_transport_op " , 0 ) ;
GRPC_CALL_LOG_OP ( GPR_INFO , elem , op ) ;
grpc_error * error = hc_mutate_op ( exec_ctx , elem , op ) ;
if ( error ! = GRPC_ERROR_NONE ) { if ( error ! = GRPC_ERROR_NONE ) {
grpc_transport_stream_op_batch_finish_with_failure ( exec_ctx , op , error ) ; grpc_transport_stream_op_batch_finish_with_failure (
} else { exec_ctx , calld - > send_message_batch , error ) ;
call_data * calld = elem - > call_data ; } else if ( ! batch_will_be_handled_asynchronously ) {
if ( op - > send_message & & calld - > send_message_blocked ) { grpc_call_next_op ( exec_ctx , elem , batch ) ;
/* Don't forward the op. send_message contains slices that aren't ready
yet . The call will be forwarded by the op_complete of slice read call .
*/
} else {
grpc_call_next_op ( exec_ctx , elem , op ) ;
}
} }
GPR_TIMER_END ( " hc_start_transport_op " , 0 ) ; GPR_TIMER_END ( " hc_start_transport_stream_op_batch " , 0 ) ;
} }
/* Constructor for call_data */ /* Constructor for call_data */
static grpc_error * init_call_elem ( grpc_exec_ctx * exec_ctx , static grpc_error * init_call_elem ( grpc_exec_ctx * exec_ctx ,
grpc_call_element * elem , grpc_call_element * elem ,
const grpc_call_element_args * args ) { const grpc_call_element_args * args ) {
call_data * calld = elem - > call_data ; call_data * calld = ( call_data * ) elem - > call_data ;
calld - > on_done_recv_initial_metadata = NULL ; GRPC_CLOSURE_INIT ( & calld - > recv_initial_metadata_ready ,
calld - > on_done_recv_trailing_metadata = NULL ; recv_initial_metadata_ready , elem ,
calld - > on_complete = NULL ;
calld - > payload_bytes = NULL ;
calld - > send_message_blocked = false ;
grpc_slice_buffer_init ( & calld - > slices ) ;
GRPC_CLOSURE_INIT ( & calld - > hc_on_recv_initial_metadata ,
hc_on_recv_initial_metadata , elem ,
grpc_schedule_on_exec_ctx ) ;
GRPC_CLOSURE_INIT ( & calld - > hc_on_recv_trailing_metadata ,
hc_on_recv_trailing_metadata , elem ,
grpc_schedule_on_exec_ctx ) ;
GRPC_CLOSURE_INIT ( & calld - > hc_on_complete , hc_on_complete , elem ,
grpc_schedule_on_exec_ctx ) ; grpc_schedule_on_exec_ctx ) ;
GRPC_CLOSURE_INIT ( & calld - > got_slice , got_slice , elem , GRPC_CLOSURE_INIT ( & calld - > recv_trailing_metadata_on_complete ,
grpc_schedule_on_exec_ctx ) ; recv_trailing_metadata_on_complete , elem ,
GRPC_CLOSURE_INIT ( & calld - > send_done , send_done , elem ,
grpc_schedule_on_exec_ctx ) ; grpc_schedule_on_exec_ctx ) ;
GRPC_CLOSURE_INIT ( & calld - > send_message_on_complete , send_message_on_complete ,
elem , grpc_schedule_on_exec_ctx ) ;
GRPC_CLOSURE_INIT ( & calld - > on_send_message_next_done ,
on_send_message_next_done , elem , grpc_schedule_on_exec_ctx ) ;
return GRPC_ERROR_NONE ; return GRPC_ERROR_NONE ;
} }
/* Destructor for call_data */ /* Destructor for call_data */
static void destroy_call_elem ( grpc_exec_ctx * exec_ctx , grpc_call_element * elem , static void destroy_call_elem ( grpc_exec_ctx * exec_ctx , grpc_call_element * elem ,
const grpc_call_final_info * final_info , const grpc_call_final_info * final_info ,
grpc_closure * ignored ) { grpc_closure * ignored ) { }
call_data * calld = elem - > call_data ;
grpc_slice_buffer_destroy_internal ( exec_ctx , & calld - > slices ) ;
}
static grpc_mdelem scheme_from_args ( const grpc_channel_args * args ) { static grpc_mdelem scheme_from_args ( const grpc_channel_args * args ) {
unsigned i ; unsigned i ;
@ -580,7 +556,7 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
} }
const grpc_channel_filter grpc_http_client_filter = { const grpc_channel_filter grpc_http_client_filter = {
hc_start_transport_op , hc_start_transport_stream_ op_batch ,
grpc_channel_next_op , grpc_channel_next_op ,
sizeof ( call_data ) , sizeof ( call_data ) ,
init_call_elem , init_call_elem ,
Yash Tibrewal
7 years ago