1 // Copyright 2023 gRPC authors. 2 // 3 // Licensed under the Apache License, Version 2.0 (the "License"); 4 // you may not use this file except in compliance with the License. 5 // You may obtain a copy of the License at 6 // 7 // http://www.apache.org/licenses/LICENSE-2.0 8 // 9 // Unless required by applicable law or agreed to in writing, software 10 // distributed under the License is distributed on an "AS IS" BASIS, 11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 // See the License for the specific language governing permissions and 13 // limitations under the License. 14 15 #ifndef GRPC_TEST_CORE_END2END_FIXTURES_SOCKPAIR_FIXTURE_H 16 #define GRPC_TEST_CORE_END2END_FIXTURES_SOCKPAIR_FIXTURE_H 17 18 #include <utility> 19 20 #include "absl/functional/any_invocable.h" 21 #include "absl/status/status.h" 22 #include "absl/status/statusor.h" 23 #include "gtest/gtest.h" 24 25 #include <grpc/grpc.h> 26 #include <grpc/impl/channel_arg_names.h> 27 #include <grpc/status.h> 28 #include <grpc/support/log.h> 29 30 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" 31 #include "src/core/lib/channel/channel_args.h" 32 #include "src/core/lib/channel/channel_args_preconditioning.h" 33 #include "src/core/lib/channel/channelz.h" 34 #include "src/core/lib/config/core_configuration.h" 35 #include "src/core/lib/gprpp/ref_counted_ptr.h" 36 #include "src/core/lib/iomgr/endpoint.h" 37 #include "src/core/lib/iomgr/endpoint_pair.h" 38 #include "src/core/lib/iomgr/error.h" 39 #include "src/core/lib/iomgr/exec_ctx.h" 40 #include "src/core/lib/surface/channel.h" 41 #include "src/core/lib/surface/channel_create.h" 42 #include "src/core/lib/surface/channel_stack_type.h" 43 #include "src/core/lib/surface/completion_queue.h" 44 #include "src/core/lib/surface/server.h" 45 #include "src/core/lib/transport/transport.h" 46 #include "test/core/end2end/end2end_tests.h" 47 48 namespace grpc_core { 49 50 class SockpairFixture : public CoreTestFixture { 51 public: SockpairFixture(const ChannelArgs & ep_args)52 explicit SockpairFixture(const ChannelArgs& ep_args) 53 : ep_(grpc_iomgr_create_endpoint_pair("fixture", ep_args.ToC().get())) {} 54 ~SockpairFixture()55 ~SockpairFixture() override { 56 ExecCtx exec_ctx; 57 if (ep_.client != nullptr) { 58 grpc_endpoint_shutdown(ep_.client, absl::InternalError("done")); 59 grpc_endpoint_destroy(ep_.client); 60 } 61 if (ep_.server != nullptr) { 62 grpc_endpoint_shutdown(ep_.server, absl::InternalError("done")); 63 grpc_endpoint_destroy(ep_.server); 64 } 65 } 66 67 private: MutateClientArgs(ChannelArgs args)68 virtual ChannelArgs MutateClientArgs(ChannelArgs args) { return args; } MutateServerArgs(ChannelArgs args)69 virtual ChannelArgs MutateServerArgs(ChannelArgs args) { return args; } MakeServer(const ChannelArgs & in_args,grpc_completion_queue * cq,absl::AnyInvocable<void (grpc_server *)> & pre_server_start)70 grpc_server* MakeServer( 71 const ChannelArgs& in_args, grpc_completion_queue* cq, 72 absl::AnyInvocable<void(grpc_server*)>& pre_server_start) override { 73 auto args = MutateServerArgs(in_args); 74 ExecCtx exec_ctx; 75 Transport* transport; 76 auto* server = grpc_server_create(args.ToC().get(), nullptr); 77 grpc_server_register_completion_queue(server, cq, nullptr); 78 pre_server_start(server); 79 grpc_server_start(server); 80 auto server_channel_args = CoreConfiguration::Get() 81 .channel_args_preconditioning() 82 .PreconditionChannelArgs(args.ToC().get()); 83 auto* server_endpoint = std::exchange(ep_.server, nullptr); 84 EXPECT_NE(server_endpoint, nullptr); 85 transport = grpc_create_chttp2_transport(server_channel_args, 86 server_endpoint, false); 87 grpc_endpoint_add_to_pollset(server_endpoint, grpc_cq_pollset(cq)); 88 Server* core_server = Server::FromC(server); 89 grpc_error_handle error = core_server->SetupTransport( 90 transport, nullptr, core_server->channel_args(), nullptr); 91 if (error.ok()) { 92 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr, nullptr); 93 } else { 94 transport->Orphan(); 95 } 96 return server; 97 } MakeClient(const ChannelArgs & in_args,grpc_completion_queue *)98 grpc_channel* MakeClient(const ChannelArgs& in_args, 99 grpc_completion_queue*) override { 100 ExecCtx exec_ctx; 101 auto args = CoreConfiguration::Get() 102 .channel_args_preconditioning() 103 .PreconditionChannelArgs( 104 MutateClientArgs(in_args) 105 .Set(GRPC_ARG_DEFAULT_AUTHORITY, "test-authority") 106 .ToC() 107 .get()); 108 Transport* transport; 109 auto* client_endpoint = std::exchange(ep_.client, nullptr); 110 EXPECT_NE(client_endpoint, nullptr); 111 transport = grpc_create_chttp2_transport(args, client_endpoint, true); 112 auto channel = ChannelCreate("socketpair-target", args, 113 GRPC_CLIENT_DIRECT_CHANNEL, transport); 114 grpc_channel* client; 115 if (channel.ok()) { 116 client = channel->release()->c_ptr(); 117 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr, nullptr); 118 } else { 119 client = grpc_lame_client_channel_create( 120 nullptr, static_cast<grpc_status_code>(channel.status().code()), 121 "lame channel"); 122 transport->Orphan(); 123 } 124 GPR_ASSERT(client); 125 return client; 126 } 127 128 grpc_endpoint_pair ep_; 129 }; 130 } // namespace grpc_core 131 132 #endif // GRPC_TEST_CORE_END2END_FIXTURES_SOCKPAIR_FIXTURE_H 133