1 //
2 //
3 // Copyright 2016 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 #include "test/core/util/mock_endpoint.h"
20
21 #include "absl/status/status.h"
22 #include "absl/strings/string_view.h"
23
24 #include <grpc/slice_buffer.h>
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/log.h>
27 #include <grpc/support/sync.h>
28
29 #include "src/core/lib/gprpp/debug_location.h"
30 #include "src/core/lib/iomgr/closure.h"
31 #include "src/core/lib/iomgr/error.h"
32 #include "src/core/lib/iomgr/exec_ctx.h"
33 #include "src/core/lib/iomgr/iomgr_fwd.h"
34 #include "src/core/lib/iomgr/sockaddr.h"
35
36 typedef struct mock_endpoint {
37 grpc_endpoint base;
38 gpr_mu mu;
39 void (*on_write)(grpc_slice slice);
40 grpc_slice_buffer read_buffer;
41 grpc_slice_buffer* on_read_out;
42 grpc_closure* on_read;
43 bool put_reads_done;
44 bool destroyed;
45 } mock_endpoint;
46
me_read(grpc_endpoint * ep,grpc_slice_buffer * slices,grpc_closure * cb,bool,int)47 static void me_read(grpc_endpoint* ep, grpc_slice_buffer* slices,
48 grpc_closure* cb, bool /*urgent*/,
49 int /*min_progress_size*/) {
50 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
51 gpr_mu_lock(&m->mu);
52 if (m->read_buffer.count > 0) {
53 grpc_slice_buffer_swap(&m->read_buffer, slices);
54 grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb, absl::OkStatus());
55 } else if (m->put_reads_done) {
56 grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb,
57 absl::UnavailableError("reads done"));
58 } else {
59 m->on_read = cb;
60 m->on_read_out = slices;
61 }
62 gpr_mu_unlock(&m->mu);
63 }
64
me_write(grpc_endpoint * ep,grpc_slice_buffer * slices,grpc_closure * cb,void *,int)65 static void me_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
66 grpc_closure* cb, void* /*arg*/, int /*max_frame_size*/) {
67 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
68 for (size_t i = 0; i < slices->count; i++) {
69 m->on_write(slices->slices[i]);
70 }
71 grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb, absl::OkStatus());
72 }
73
me_add_to_pollset(grpc_endpoint *,grpc_pollset *)74 static void me_add_to_pollset(grpc_endpoint* /*ep*/,
75 grpc_pollset* /*pollset*/) {}
76
me_add_to_pollset_set(grpc_endpoint *,grpc_pollset_set *)77 static void me_add_to_pollset_set(grpc_endpoint* /*ep*/,
78 grpc_pollset_set* /*pollset*/) {}
79
me_delete_from_pollset_set(grpc_endpoint *,grpc_pollset_set *)80 static void me_delete_from_pollset_set(grpc_endpoint* /*ep*/,
81 grpc_pollset_set* /*pollset*/) {}
82
me_shutdown(grpc_endpoint * ep,grpc_error_handle why)83 static void me_shutdown(grpc_endpoint* ep, grpc_error_handle why) {
84 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
85 gpr_mu_lock(&m->mu);
86 if (m->on_read) {
87 grpc_core::ExecCtx::Run(
88 DEBUG_LOCATION, m->on_read,
89 GRPC_ERROR_CREATE_REFERENCING("Endpoint Shutdown", &why, 1));
90 m->on_read = nullptr;
91 }
92 gpr_mu_unlock(&m->mu);
93 }
94
destroy(mock_endpoint * m)95 static void destroy(mock_endpoint* m) {
96 grpc_slice_buffer_destroy(&m->read_buffer);
97 gpr_mu_destroy(&m->mu);
98 gpr_free(m);
99 }
100
me_destroy(grpc_endpoint * ep)101 static void me_destroy(grpc_endpoint* ep) {
102 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
103 m->destroyed = true;
104 if (m->put_reads_done) {
105 destroy(m);
106 }
107 }
108
grpc_mock_endpoint_finish_put_reads(grpc_endpoint * ep)109 void grpc_mock_endpoint_finish_put_reads(grpc_endpoint* ep) {
110 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
111 m->put_reads_done = true;
112 if (m->destroyed) {
113 destroy(m);
114 }
115 }
116
me_get_peer(grpc_endpoint *)117 static absl::string_view me_get_peer(grpc_endpoint* /*ep*/) {
118 return "fake:mock_endpoint";
119 }
120
me_get_local_address(grpc_endpoint *)121 static absl::string_view me_get_local_address(grpc_endpoint* /*ep*/) {
122 return "fake:mock_endpoint";
123 }
124
me_get_fd(grpc_endpoint *)125 static int me_get_fd(grpc_endpoint* /*ep*/) { return -1; }
126
me_can_track_err(grpc_endpoint *)127 static bool me_can_track_err(grpc_endpoint* /*ep*/) { return false; }
128
129 static const grpc_endpoint_vtable vtable = {me_read,
130 me_write,
131 me_add_to_pollset,
132 me_add_to_pollset_set,
133 me_delete_from_pollset_set,
134 me_shutdown,
135 me_destroy,
136 me_get_peer,
137 me_get_local_address,
138 me_get_fd,
139 me_can_track_err};
140
grpc_mock_endpoint_create(void (* on_write)(grpc_slice slice))141 grpc_endpoint* grpc_mock_endpoint_create(void (*on_write)(grpc_slice slice)) {
142 mock_endpoint* m = static_cast<mock_endpoint*>(gpr_malloc(sizeof(*m)));
143 m->base.vtable = &vtable;
144 grpc_slice_buffer_init(&m->read_buffer);
145 gpr_mu_init(&m->mu);
146 m->on_write = on_write;
147 m->on_read = nullptr;
148 m->put_reads_done = false;
149 m->destroyed = false;
150 return &m->base;
151 }
152
grpc_mock_endpoint_put_read(grpc_endpoint * ep,grpc_slice slice)153 void grpc_mock_endpoint_put_read(grpc_endpoint* ep, grpc_slice slice) {
154 mock_endpoint* m = reinterpret_cast<mock_endpoint*>(ep);
155 gpr_mu_lock(&m->mu);
156 GPR_ASSERT(!m->put_reads_done);
157 if (m->on_read != nullptr) {
158 grpc_slice_buffer_add(m->on_read_out, slice);
159 grpc_core::ExecCtx::Run(DEBUG_LOCATION, m->on_read, absl::OkStatus());
160 m->on_read = nullptr;
161 } else {
162 grpc_slice_buffer_add(&m->read_buffer, slice);
163 }
164 gpr_mu_unlock(&m->mu);
165 }
166