@ -18,23 +18,30 @@
# include <algorithm> // IWYU pragma: keep
# include <memory>
# include <string> // IWYU pragma: keep
# include <tuple>
# include <vector> // IWYU pragma: keep
# include "absl/functional/any_invocable.h"
# include "absl/status/statusor.h" // IWYU pragma: keep
# include "absl/strings/str_format.h" // IWYU pragma: keep
# include "absl/types/optional.h" // IWYU pragma: keep
# include "gmock/gmock.h"
# include "gtest/gtest.h"
# include <grpc/event_engine/event_engine.h>
# include <grpc/event_engine/memory_allocator.h>
# include <grpc/event_engine/slice.h> // IWYU pragma: keep
# include <grpc/event_engine/slice_buffer.h>
# include <grpc/grpc.h>
# include <grpc/status.h> // IWYU pragma: keep
# include "src/core/lib/gprpp/ref_counted_ptr.h"
# include "src/core/lib/iomgr/timer_manager.h"
# include "src/core/lib/promise/activity.h"
# include "src/core/lib/promise/event_engine_wakeup_scheduler.h"
# include "src/core/lib/promise/join.h"
# include "src/core/lib/promise/map.h"
# include "src/core/lib/promise/pipe.h"
# include "src/core/lib/promise/seq.h"
# include "src/core/lib/resource_quota/arena.h"
@ -42,12 +49,14 @@
# include "src/core/lib/resource_quota/resource_quota.h"
# include "src/core/lib/slice/slice.h"
# include "src/core/lib/slice/slice_buffer.h"
# include "src/core/lib/slice/slice_internal.h" // IWYU pragma: keep
# include "src/core/lib/transport/metadata_batch.h" // IWYU pragma: keep
# include "test/core/event_engine/fuzzing_event_engine/fuzzing_event_engine.h"
# include "test/core/event_engine/fuzzing_event_engine/fuzzing_event_engine.pb.h"
# include "test/core/promise/test_wakeup_schedulers.h"
using testing : : MockFunction ;
using testing : : Return ;
using testing : : Sequence ;
using testing : : StrictMock ;
using testing : : WithArgs ;
@ -101,19 +110,117 @@ class ClientTransportTest : public ::testing::Test {
return options ;
} ( ) ,
fuzzing_event_engine : : Actions ( ) ) ) ,
client_transport_ (
std : : make_unique < PromiseEndpoint > (
std : : unique_ptr < MockEndpoint > ( control_endpoint_ptr_ ) ,
SliceBuffer ( ) ) ,
std : : make_unique < PromiseEndpoint > (
std : : unique_ptr < MockEndpoint > ( data_endpoint_ptr_ ) ,
SliceBuffer ( ) ) ,
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
arena_ ( MakeScopedArena ( initial_arena_size , & memory_allocator_ ) ) ,
pipe_client_to_server_messages_ ( arena_ . get ( ) ) ,
pipe_client_to_server_messages_second_ ( arena_ . get ( ) ) { }
pipe_server_to_client_messages_ ( arena_ . get ( ) ) ,
pipe_server_intial_metadata_ ( arena_ . get ( ) ) ,
pipe_client_to_server_messages_second_ ( arena_ . get ( ) ) ,
pipe_server_to_client_messages_second_ ( arena_ . get ( ) ) ,
pipe_server_intial_metadata_second_ ( arena_ . get ( ) ) { }
// Expect how client transport will read from control/data endpoints with a
// test frame.
void AddReadExpectations ( int num_of_streams ) {
for ( int i = 0 ; i < num_of_streams ; i + + ) {
EXPECT_CALL ( control_endpoint_ , Read )
. InSequence ( control_endpoint_sequence )
. WillOnce ( WithArgs < 0 , 1 > (
[ this , i ] ( absl : : AnyInvocable < void ( absl : : Status ) > on_read ,
grpc_event_engine : : experimental : : SliceBuffer *
buffer ) mutable {
// Construct test frame for EventEngine read: headers (15
// bytes), message(16 bytes), message padding (48 byte),
// trailers (15 bytes).
const std : : string frame_header = {
static_cast < char > ( 0x80 ) , // frame type = fragment
0x03 , // flag = has header + has trailer
0x00 ,
0x00 ,
static_cast < char > ( i + 1 ) , // stream id = 1
0x00 ,
0x00 ,
0x00 ,
0x1a , // header length = 26
0x00 ,
0x00 ,
0x00 ,
0x08 , // message length = 8
0x00 ,
0x00 ,
0x00 ,
0x38 , // message padding =56
0x00 ,
0x00 ,
0x00 ,
0x0f , // trailer length = 15
0x00 ,
0x00 ,
0x00 } ;
// Schedule mock_endpoint to read buffer.
grpc_event_engine : : experimental : : Slice slice (
grpc_slice_from_cpp_string ( frame_header ) ) ;
buffer - > Append ( std : : move ( slice ) ) ;
// Execute read callback later to control when read starts.
read_callback . push_back ( std : : move ( on_read ) ) ;
// Return false to mock EventEngine read not finish.
return false ;
} ) ) ;
EXPECT_CALL ( control_endpoint_ , Read )
. InSequence ( control_endpoint_sequence )
. WillOnce ( WithArgs < 1 > (
[ ] ( grpc_event_engine : : experimental : : SliceBuffer * buffer ) {
// Encoded string of header ":path: /demo.Service/Step".
const std : : string header = {
0x10 , 0x05 , 0x3a , 0x70 , 0x61 , 0x74 , 0x68 , 0x12 , 0x2f ,
0x64 , 0x65 , 0x6d , 0x6f , 0x2e , 0x53 , 0x65 , 0x72 , 0x76 ,
0x69 , 0x63 , 0x65 , 0x2f , 0x53 , 0x74 , 0x65 , 0x70 } ;
// Encoded string of trailer "grpc-status: 0".
const std : : string trailers = { 0x10 , 0x0b , 0x67 , 0x72 , 0x70 ,
0x63 , 0x2d , 0x73 , 0x74 , 0x61 ,
0x74 , 0x75 , 0x73 , 0x01 , 0x30 } ;
// Schedule mock_endpoint to read buffer.
grpc_event_engine : : experimental : : Slice slice (
grpc_slice_from_cpp_string ( header + trailers ) ) ;
buffer - > Append ( std : : move ( slice ) ) ;
return true ;
} ) ) ;
}
EXPECT_CALL ( control_endpoint_ , Read )
. InSequence ( control_endpoint_sequence )
. WillOnce ( Return ( false ) ) ;
for ( int i = 0 ; i < num_of_streams ; i + + ) {
EXPECT_CALL ( data_endpoint_ , Read )
. InSequence ( data_endpoint_sequence )
. WillOnce ( WithArgs < 1 > (
[ this ] ( grpc_event_engine : : experimental : : SliceBuffer * buffer ) {
const std : : string message_padding = {
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 ,
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 ,
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 ,
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 ,
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 ,
0x00 , 0x00 , 0x00 , 0x00 , 0x00 , 0x00 } ;
grpc_event_engine : : experimental : : Slice slice (
grpc_slice_from_cpp_string ( message_padding + message ) ) ;
buffer - > Append ( std : : move ( slice ) ) ;
return true ;
} ) ) ;
}
}
// Initial ClientTransport with read expecations
void InitialClientTransport ( int num_of_streams ) {
// Read expectaions need to be added before transport initialization since
// reader_ activity loop is started in ClientTransport initialization,
AddReadExpectations ( num_of_streams ) ;
client_transport_ = std : : make_unique < ClientTransport > (
std : : make_unique < PromiseEndpoint > (
std : : unique_ptr < MockEndpoint > ( control_endpoint_ptr_ ) ,
SliceBuffer ( ) ) ,
std : : make_unique < PromiseEndpoint > (
std : : unique_ptr < MockEndpoint > ( data_endpoint_ptr_ ) , SliceBuffer ( ) ) ,
std : : static_pointer_cast < grpc_event_engine : : experimental : : EventEngine > (
event_engine_ ) ) ;
}
// Create client to server test messages.
std : : vector < MessageHandle > CreateMessages ( int num_of_messages ) {
std : : vector < MessageHandle > messages ;
for ( int i = 0 ; i < num_of_messages ; i + + ) {
@ -125,103 +232,238 @@ class ClientTransportTest : public ::testing::Test {
}
return messages ;
}
// Wait for last stream read to finish.
auto Wait ( ) {
return [ this ] ( ) mutable - > Poll < Result > {
MutexLock lock ( & mu_ ) ;
if ( last_stream_read_done_ ) {
return Result { } ;
} else {
waker_ = Activity : : current ( ) - > MakeNonOwningWaker ( ) ;
return Pending ( ) ;
}
} ;
}
// Wake up the pending Wait() promise.
void Wakeup ( ) {
MutexLock lock ( & mu_ ) ;
last_stream_read_done_ = true ;
waker_ . Wakeup ( ) ;
}
private :
struct Result { } ;
Mutex mu_ ;
Waker waker_ ABSL_GUARDED_BY ( mu_ ) ;
bool last_stream_read_done_ ABSL_GUARDED_BY ( mu_ ) = false ;
MockEndpoint * control_endpoint_ptr_ ;
MockEndpoint * data_endpoint_ptr_ ;
size_t initial_arena_size = 1024 ;
MemoryAllocator memory_allocator_ ;
Sequence control_endpoint_sequence ;
Sequence data_endpoint_sequence ;
protected :
MockEndpoint & control_endpoint_ ;
MockEndpoint & data_endpoint_ ;
std : : shared_ptr < grpc_event_engine : : experimental : : FuzzingEventEngine >
event_engine_ ;
ClientTransport client_transport_ ;
std : : unique_ptr < ClientTransport > client_transport_ ;
ScopedArenaPtr arena_ ;
Pipe < MessageHandle > pipe_client_to_server_messages_ ;
Pipe < MessageHandle > pipe_server_to_client_messages_ ;
Pipe < ServerMetadataHandle > pipe_server_intial_metadata_ ;
// Added for mutliple streams tests.
Pipe < MessageHandle > pipe_client_to_server_messages_second_ ;
const absl : : Status kDummyErrorStatus =
absl : : ErrnoToStatus ( 5566 , " just an error " ) ;
static constexpr size_t kDummyRequestSize = 5566u ;
Pipe < MessageHandle > pipe_server_to_client_messages_second_ ;
Pipe < ServerMetadataHandle > pipe_server_intial_metadata_second_ ;
std : : vector < absl : : AnyInvocable < void ( absl : : Status ) > > read_callback ;
// Added to verify received message payload.
const std : : string message = { 0x01 , 0x02 , 0x03 , 0x04 , 0x05 , 0x06 , 0x07 , 0x08 } ;
} ;
TEST_F ( ClientTransportTest , AddOneStream ) {
InitialClientTransport ( 1 ) ;
auto messages = CreateMessages ( 1 ) ;
ClientMetadataHandle md ;
auto args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_ . receiver , nullptr } ;
auto args = CallArgs { std : : move ( md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_ . sender ,
& pipe_client_to_server_messages_ . receiver ,
& pipe_server_to_client_messages_ . sender } ;
StrictMock < MockFunction < void ( absl : : Status ) > > on_done ;
EXPECT_CALL ( on_done , Call ( absl : : OkStatus ( ) ) ) ;
EXPECT_CALL ( control_endpoint_ , Write ) . WillOnce ( Return ( true ) ) ;
EXPECT_CALL ( data_endpoint_ , Write ) . WillOnce ( Return ( true ) ) ;
auto activity = MakeActivity (
Seq (
// Concurrently: send message into the pipe, and receive from the
// pipe.
Join ( Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
[ this ] {
this - > pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
client_transport_ . AddStream ( std : : move ( args ) ) ) ,
// Concurrently: write and read messages in client transport.
Join (
// Send messages to call_args.client_to_server_messages pipe,
// which will be eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
[ this ] {
pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Add first stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream ( std : : move ( args ) ) ,
[ this ] ( ) {
// Concurrently: start read from control endpoints.
read_callback [ 0 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream will finish with server trailers:
// "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] ( ) {
// Close pipes after receive message.
pipe_server_to_client_messages_ . sender . Close ( ) ;
pipe_server_intial_metadata_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
// Once complete, verify successful sending and the received value.
[ ] ( const std : : tuple < absl : : Status , absl : : Status > & ret ) {
[ ] ( const std : : tuple < absl : : Status , absl : : Status , absl : : Status > & ret ) {
EXPECT_TRUE ( std : : get < 0 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 1 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 2 > ( ret ) . ok ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
InlineWakeupScheduler ( ) ,
EventEngineWakeupScheduler (
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
[ & on_done ] ( absl : : Status status ) { on_done . Call ( std : : move ( status ) ) ; } ) ;
// Wait until ClientTransport's internal activities to finish.
event_engine_ - > TickUntilIdle ( ) ;
event_engine_ - > UnsetGlobalHooks ( ) ;
}
TEST_F ( ClientTransportTest , AddOneStreamWithEEFailed ) {
TEST_F ( ClientTransportTest , AddOneStreamWithWriteFailed ) {
InitialClientTransport ( 1 ) ;
auto messages = CreateMessages ( 1 ) ;
ClientMetadataHandle md ;
auto args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_ . receiver , nullptr } ;
auto args = CallArgs { std : : move ( md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_ . sender ,
& pipe_client_to_server_messages_ . receiver ,
& pipe_server_to_client_messages_ . sender } ;
StrictMock < MockFunction < void ( absl : : Status ) > > on_done ;
EXPECT_CALL ( on_done , Call ( absl : : OkStatus ( ) ) ) ;
EXPECT_CALL ( control_endpoint_ , Write )
. WillOnce (
WithArgs < 0 > ( [ this ] ( absl : : AnyInvocable < void ( absl : : Status ) > on_write ) {
on_write ( this - > kDummyErrorStatus ) ;
WithArgs < 0 > ( [ ] ( absl : : AnyInvocable < void ( absl : : Status ) > on_write ) {
on_write ( absl : : InternalError ( " control endpoint write failed. " ) ) ;
return false ;
} ) ) ;
EXPECT_CALL ( data_endpoint_ , Write )
. WillOnce (
WithArgs < 0 > ( [ this ] ( absl : : AnyInvocable < void ( absl : : Status ) > on_write ) {
on_write ( this - > kDummyErrorStatus ) ;
WithArgs < 0 > ( [ ] ( absl : : AnyInvocable < void ( absl : : Status ) > on_write ) {
on_write ( absl : : InternalError ( " control endpoint write failed. " ) ) ;
return false ;
} ) ) ;
auto activity = MakeActivity (
Seq (
// Concurrently: send message into the pipe, and receive from the
// pipe.
Join ( Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
[ this ] {
this - > pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
client_transport_ . AddStream ( std : : move ( args ) ) ) ,
// Concurrently: write and read messages in client transport.
Join (
// Send messages to call_args.client_to_server_messages pipe,
// which will be eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
[ this ] {
pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Add first stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream ( std : : move ( args ) ) ,
[ this ] ( ) {
// Start read from endpoints.
read_callback [ 0 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream will finish with server trailers:
// "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Close pipes after receive message.
pipe_server_to_client_messages_ . sender . Close ( ) ;
pipe_server_intial_metadata_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
// Once complete, verify successful sending and the received value.
[ ] ( const std : : tuple < absl : : Status , absl : : Status > & ret ) {
[ ] ( const std : : tuple < absl : : Status , absl : : Status , absl : : Status > & ret ) {
// TODO(ladynana): change these expectations to errors after the
// writer activity closes transport for EE failures.
// writer activity closes transport for write failures.
EXPECT_TRUE ( std : : get < 0 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 1 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 2 > ( ret ) . ok ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
InlineWakeupScheduler ( ) ,
EventEngineWakeupScheduler (
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
[ & on_done ] ( absl : : Status status ) { on_done . Call ( std : : move ( status ) ) ; } ) ;
// Wait until ClientTransport's internal activities to finish.
event_engine_ - > TickUntilIdle ( ) ;
@ -229,37 +471,90 @@ TEST_F(ClientTransportTest, AddOneStreamWithEEFailed) {
}
TEST_F ( ClientTransportTest , AddOneStreamMultipleMessages ) {
InitialClientTransport ( 1 ) ;
auto messages = CreateMessages ( 3 ) ;
ClientMetadataHandle md ;
auto args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_ . receiver , nullptr } ;
auto args = CallArgs { std : : move ( md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_ . sender ,
& pipe_client_to_server_messages_ . receiver ,
& pipe_server_to_client_messages_ . sender } ;
StrictMock < MockFunction < void ( absl : : Status ) > > on_done ;
EXPECT_CALL ( on_done , Call ( absl : : OkStatus ( ) ) ) ;
EXPECT_CALL ( control_endpoint_ , Write ) . Times ( 3 ) . WillRepeatedly ( Return ( true ) ) ;
EXPECT_CALL ( data_endpoint_ , Write ) . Times ( 3 ) . WillRepeatedly ( Return ( true ) ) ;
auto activity = MakeActivity (
Seq (
// Concurrently: send messages into the pipe, and receive from the
// pipe.
Join ( Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 1 ] ) ) ,
pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 2 ] ) ) ,
[ this ] {
this - > pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
client_transport_ . AddStream ( std : : move ( args ) ) ) ,
// Concurrently: write and read messages in client transport.
Join (
// Send messages to call_args.client_to_server_messages pipe,
// which will be eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 1 ] ) ) ,
pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 2 ] ) ) ,
[ this ] {
pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Add first stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream ( std : : move ( args ) ) ,
[ this ] ( ) {
// Start read from endpoints.
read_callback [ 0 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream finish with trailers "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Close pipes after receive message.
pipe_server_to_client_messages_ . sender . Close ( ) ;
pipe_server_intial_metadata_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
// Once complete, verify successful sending and the received value.
[ ] ( const std : : tuple < absl : : Status , absl : : Status > & ret ) {
[ ] ( const std : : tuple < absl : : Status , absl : : Status , absl : : Status > & ret ) {
EXPECT_TRUE ( std : : get < 0 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 1 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 2 > ( ret ) . ok ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
InlineWakeupScheduler ( ) ,
EventEngineWakeupScheduler (
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
[ & on_done ] ( absl : : Status status ) { on_done . Call ( std : : move ( status ) ) ; } ) ;
// Wait until ClientTransport's internal activities to finish.
event_engine_ - > TickUntilIdle ( ) ;
@ -267,51 +562,160 @@ TEST_F(ClientTransportTest, AddOneStreamMultipleMessages) {
}
TEST_F ( ClientTransportTest , AddMultipleStreams ) {
InitialClientTransport ( 2 ) ;
auto messages = CreateMessages ( 2 ) ;
ClientMetadataHandle md ;
auto first_stream_args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_ . receiver , nullptr } ;
auto second_stream_args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_second_ . receiver , nullptr } ;
ClientMetadataHandle first_stream_md ;
ClientMetadataHandle second_stream_md ;
auto first_stream_args =
CallArgs { std : : move ( first_stream_md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_ . sender ,
& pipe_client_to_server_messages_ . receiver ,
& pipe_server_to_client_messages_ . sender } ;
auto second_stream_args =
CallArgs { std : : move ( second_stream_md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_second_ . sender ,
& pipe_client_to_server_messages_second_ . receiver ,
& pipe_server_to_client_messages_second_ . sender } ;
StrictMock < MockFunction < void ( absl : : Status ) > > on_done ;
EXPECT_CALL ( on_done , Call ( absl : : OkStatus ( ) ) ) ;
EXPECT_CALL ( control_endpoint_ , Write ) . Times ( 2 ) . WillRepeatedly ( Return ( true ) ) ;
EXPECT_CALL ( data_endpoint_ , Write ) . Times ( 2 ) . WillRepeatedly ( Return ( true ) ) ;
auto activity = MakeActivity (
Seq (
// Concurrently: send messages into the pipe, and receive from the
// pipe.
// Concurrently: write and read messages from client transport.
Join (
// Send message to first stream pipe.
// Send messages to first stream's
// call_args.client_to_server_messages pipe, which will be
// eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
[ this ] {
pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Send message to second stream pipe.
// Send messages to second stream's
// call_args.client_to_server_messages pipe, which will be
// eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_second_ . sender . Push (
std : : move ( messages [ 1 ] ) ) ,
[ this ] {
pipe_client_to_server_messages_second_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive message from first stream pipe.
client_transport_ . AddStream ( std : : move ( first_stream_args ) ) ,
// Receive message from second stream pipe.
client_transport_ . AddStream ( std : : move ( second_stream_args ) ) ) ,
// Add first stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream (
std : : move ( first_stream_args ) ) ,
[ this ] {
read_callback [ 0 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream finish with trailers "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Add second stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream (
std : : move ( second_stream_args ) ) ,
Seq ( Wait ( ) ,
[ this ] {
// Wait until first stream read finished to start
// the second read.
read_callback [ 1 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream finish with trailers "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive first stream's messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Wake up the sencond stream read after first stream read
// finished.
Wakeup ( ) ;
// Close pipes after receive message.
pipe_server_to_client_messages_ . sender . Close ( ) ;
pipe_server_intial_metadata_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive second stream's messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_second_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_second_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Close pipes after receive message.
pipe_server_to_client_messages_second_ . sender . Close ( ) ;
pipe_server_intial_metadata_second_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
// Once complete, verify successful sending and the received value.
[ ] ( const std : : tuple < absl : : Status , absl : : Status , absl : : Status ,
absl : : Status > & ret ) {
absl : : Status , absl : : Status , absl : : Status > & ret ) {
EXPECT_TRUE ( std : : get < 0 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 1 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 2 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 3 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 4 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 5 > ( ret ) . ok ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
InlineWakeupScheduler ( ) ,
EventEngineWakeupScheduler (
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
[ & on_done ] ( absl : : Status status ) { on_done . Call ( std : : move ( status ) ) ; } ) ;
// Wait until ClientTransport's internal activities to finish.
event_engine_ - > TickUntilIdle ( ) ;
@ -319,24 +723,35 @@ TEST_F(ClientTransportTest, AddMultipleStreams) {
}
TEST_F ( ClientTransportTest , AddMultipleStreamsMultipleMessages ) {
InitialClientTransport ( 2 ) ;
auto messages = CreateMessages ( 6 ) ;
ClientMetadataHandle md ;
auto first_stream_args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_ . receiver , nullptr } ;
auto second_stream_args = CallArgs {
std : : move ( md ) , ClientInitialMetadataOutstandingToken : : Empty ( ) , nullptr ,
nullptr , & pipe_client_to_server_messages_second_ . receiver , nullptr } ;
ClientMetadataHandle first_stream_md ;
ClientMetadataHandle second_stream_md ;
auto first_stream_args =
CallArgs { std : : move ( first_stream_md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_ . sender ,
& pipe_client_to_server_messages_ . receiver ,
& pipe_server_to_client_messages_ . sender } ;
auto second_stream_args =
CallArgs { std : : move ( second_stream_md ) ,
ClientInitialMetadataOutstandingToken : : Empty ( ) ,
nullptr ,
& pipe_server_intial_metadata_second_ . sender ,
& pipe_client_to_server_messages_second_ . receiver ,
& pipe_server_to_client_messages_second_ . sender } ;
StrictMock < MockFunction < void ( absl : : Status ) > > on_done ;
EXPECT_CALL ( on_done , Call ( absl : : OkStatus ( ) ) ) ;
EXPECT_CALL ( control_endpoint_ , Write ) . Times ( 6 ) . WillRepeatedly ( Return ( true ) ) ;
EXPECT_CALL ( data_endpoint_ , Write ) . Times ( 6 ) . WillRepeatedly ( Return ( true ) ) ;
auto activity = MakeActivity (
Seq (
// Concurrently: send messages into the pipe, and receive from the
// pipe.
// Concurrently: write and read messages in client transport.
Join (
// Send messages to first stream pipe.
// Send messages to first stream's
// call_args.client_to_server_messages pipe, which will be
// eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_ . sender . Push (
std : : move ( messages [ 0 ] ) ) ,
pipe_client_to_server_messages_ . sender . Push (
@ -347,7 +762,9 @@ TEST_F(ClientTransportTest, AddMultipleStreamsMultipleMessages) {
pipe_client_to_server_messages_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Send messages to second stream pipe.
// Send messages to second stream's
// call_args.client_to_server_messages pipe, which will be
// eventually sent to control/data endpoints.
Seq ( pipe_client_to_server_messages_second_ . sender . Push (
std : : move ( messages [ 3 ] ) ) ,
pipe_client_to_server_messages_second_ . sender . Push (
@ -358,20 +775,116 @@ TEST_F(ClientTransportTest, AddMultipleStreamsMultipleMessages) {
pipe_client_to_server_messages_second_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive messages from first stream pipe.
client_transport_ . AddStream ( std : : move ( first_stream_args ) ) ,
// Receive messages from second stream pipe.
client_transport_ . AddStream ( std : : move ( second_stream_args ) ) ) ,
// Add first stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream (
std : : move ( first_stream_args ) ) ,
[ this ] {
read_callback [ 0 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream finish with trailers "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Add second stream with call_args into client transport.
Seq ( Join ( client_transport_ - > AddStream (
std : : move ( second_stream_args ) ) ,
Seq ( Wait ( ) ,
[ this ] {
// Wait until first stream read finished to start
// the second read.
read_callback [ 1 ] ( absl : : OkStatus ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
[ ] ( std : : tuple < absl : : StatusOr < ServerMetadataHandle > ,
absl : : Status >
ret ) {
// AddStream finish with trailers "grpc-status:0".
EXPECT_EQ ( std : : get < 0 > ( ret )
. value ( )
- > get ( GrpcStatusMetadata ( ) )
. value ( ) ,
grpc_status_code : : GRPC_STATUS_OK ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive first stream's messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Wake up the sencond stream read after first stream read
// finished.
Wakeup ( ) ;
// Close pipes after receive message.
pipe_server_to_client_messages_ . sender . Close ( ) ;
pipe_server_intial_metadata_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive second stream's messages from control/data endpoints.
Seq (
// Receive server initial metadata.
Map ( pipe_server_intial_metadata_second_ . receiver . Next ( ) ,
[ ] ( NextResult < ServerMetadataHandle > r ) {
// Expect value: ":path: /demo.Service/Step"
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( )
- > get_pointer ( HttpPathMetadata ( ) )
- > as_string_view ( ) ,
" /demo.Service/Step " ) ;
return absl : : OkStatus ( ) ;
} ) ,
// Receive server to client messages.
Map ( pipe_server_to_client_messages_second_ . receiver . Next ( ) ,
[ this ] ( NextResult < MessageHandle > r ) {
EXPECT_TRUE ( r . has_value ( ) ) ;
EXPECT_EQ ( r . value ( ) - > payload ( ) - > JoinIntoString ( ) ,
message ) ;
return absl : : OkStatus ( ) ;
} ) ,
[ this ] {
// Close pipes after receive message.
pipe_server_to_client_messages_second_ . sender . Close ( ) ;
pipe_server_intial_metadata_second_ . sender . Close ( ) ;
return absl : : OkStatus ( ) ;
} ) ) ,
// Once complete, verify successful sending and the received value.
[ ] ( const std : : tuple < absl : : Status , absl : : Status , absl : : Status ,
absl : : Status > & ret ) {
absl : : Status , absl : : Status , absl : : Status > & ret ) {
EXPECT_TRUE ( std : : get < 0 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 1 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 2 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 3 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 4 > ( ret ) . ok ( ) ) ;
EXPECT_TRUE ( std : : get < 5 > ( ret ) . ok ( ) ) ;
return absl : : OkStatus ( ) ;
} ) ,
InlineWakeupScheduler ( ) ,
EventEngineWakeupScheduler (
std : : static_pointer_cast <
grpc_event_engine : : experimental : : EventEngine > ( event_engine_ ) ) ,
[ & on_done ] ( absl : : Status status ) { on_done . Call ( std : : move ( status ) ) ; } ) ;
// Wait until ClientTransport's internal activities to finish.
event_engine_ - > TickUntilIdle ( ) ;
nanahpang
1 year ago
committed by
GitHub