1 /*
2 * Copyright 2014 Google Inc. All rights reserved.
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include <grpc++/grpc++.h>
18
19 #include <thread>
20
21 #include "monster_test.grpc.fb.h"
22 #include "monster_test_generated.h"
23 #include "test_assert.h"
24
25 using namespace MyGame::Example;
26 using flatbuffers::FlatBufferBuilder;
27 using flatbuffers::grpc::MessageBuilder;
28
29 void message_builder_tests();
30
31 // The callback implementation of our server, that derives from the generated
32 // code. It implements all rpcs specified in the FlatBuffers schema.
33 class ServiceImpl final : public MyGame::Example::MonsterStorage::Service {
Store(::grpc::ServerContext * context,const flatbuffers::grpc::Message<Monster> * request,flatbuffers::grpc::Message<Stat> * response)34 virtual ::grpc::Status Store(
35 ::grpc::ServerContext *context,
36 const flatbuffers::grpc::Message<Monster> *request,
37 flatbuffers::grpc::Message<Stat> *response) override {
38 // Create a response from the incoming request name.
39 fbb_.Clear();
40 auto stat_offset = CreateStat(
41 fbb_, fbb_.CreateString("Hello, " + request->GetRoot()->name()->str()));
42 fbb_.Finish(stat_offset);
43 // Transfer ownership of the message to gRPC
44 *response = fbb_.ReleaseMessage<Stat>();
45 return grpc::Status::OK;
46 }
Retrieve(::grpc::ServerContext * context,const flatbuffers::grpc::Message<Stat> * request,::grpc::ServerWriter<flatbuffers::grpc::Message<Monster>> * writer)47 virtual ::grpc::Status Retrieve(
48 ::grpc::ServerContext *context,
49 const flatbuffers::grpc::Message<Stat> *request,
50 ::grpc::ServerWriter<flatbuffers::grpc::Message<Monster>> *writer)
51 override {
52 for (int i = 0; i < 5; i++) {
53 fbb_.Clear();
54 // Create 5 monsters for resposne.
55 auto monster_offset =
56 CreateMonster(fbb_, 0, 0, 0,
57 fbb_.CreateString(request->GetRoot()->id()->str() +
58 " No." + std::to_string(i)));
59 fbb_.Finish(monster_offset);
60
61 flatbuffers::grpc::Message<Monster> monster =
62 fbb_.ReleaseMessage<Monster>();
63
64 // Send monster to client using streaming.
65 writer->Write(monster);
66 }
67 return grpc::Status::OK;
68 }
69
70 private:
71 flatbuffers::grpc::MessageBuilder fbb_;
72 };
73
74 // Track the server instance, so we can terminate it later.
75 grpc::Server *server_instance = nullptr;
76 // Mutex to protec this variable.
77 std::mutex wait_for_server;
78 std::condition_variable server_instance_cv;
79
80 // This function implements the server thread.
RunServer()81 void RunServer() {
82 auto server_address = "0.0.0.0:50051";
83 // Callback interface we implemented above.
84 ServiceImpl service;
85 grpc::ServerBuilder builder;
86 builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
87 builder.RegisterService(&service);
88
89 // Start the server. Lock to change the variable we're changing.
90 wait_for_server.lock();
91 server_instance = builder.BuildAndStart().release();
92 wait_for_server.unlock();
93 server_instance_cv.notify_one();
94
95 std::cout << "Server listening on " << server_address << std::endl;
96 // This will block the thread and serve requests.
97 server_instance->Wait();
98 }
99
StoreRPC(MonsterStorage::Stub * stub)100 template<class Builder> void StoreRPC(MonsterStorage::Stub *stub) {
101 Builder fbb;
102 grpc::ClientContext context;
103 // Build a request with the name set.
104 auto monster_offset = CreateMonster(fbb, 0, 0, 0, fbb.CreateString("Fred"));
105 MessageBuilder mb(std::move(fbb));
106 mb.Finish(monster_offset);
107 auto request = mb.ReleaseMessage<Monster>();
108 flatbuffers::grpc::Message<Stat> response;
109
110 // The actual RPC.
111 auto status = stub->Store(&context, request, &response);
112
113 if (status.ok()) {
114 auto resp = response.GetRoot()->id();
115 std::cout << "RPC response: " << resp->str() << std::endl;
116 } else {
117 std::cout << "RPC failed" << std::endl;
118 }
119 }
120
RetrieveRPC(MonsterStorage::Stub * stub)121 template<class Builder> void RetrieveRPC(MonsterStorage::Stub *stub) {
122 Builder fbb;
123 grpc::ClientContext context;
124 fbb.Clear();
125 auto stat_offset = CreateStat(fbb, fbb.CreateString("Fred"));
126 fbb.Finish(stat_offset);
127 auto request = MessageBuilder(std::move(fbb)).ReleaseMessage<Stat>();
128
129 flatbuffers::grpc::Message<Monster> response;
130 auto stream = stub->Retrieve(&context, request);
131 while (stream->Read(&response)) {
132 auto resp = response.GetRoot()->name();
133 std::cout << "RPC Streaming response: " << resp->str() << std::endl;
134 }
135 }
136
grpc_server_test()137 int grpc_server_test() {
138 // Launch server.
139 std::thread server_thread(RunServer);
140
141 // wait for server to spin up.
142 std::unique_lock<std::mutex> lock(wait_for_server);
143 while (!server_instance) server_instance_cv.wait(lock);
144
145 // Now connect the client.
146 auto channel = grpc::CreateChannel("localhost:50051",
147 grpc::InsecureChannelCredentials());
148 auto stub = MyGame::Example::MonsterStorage::NewStub(channel);
149
150 StoreRPC<MessageBuilder>(stub.get());
151 StoreRPC<FlatBufferBuilder>(stub.get());
152
153 RetrieveRPC<MessageBuilder>(stub.get());
154 RetrieveRPC<FlatBufferBuilder>(stub.get());
155
156 #if !FLATBUFFERS_GRPC_DISABLE_AUTO_VERIFICATION
157 {
158 // Test that an invalid request errors out correctly
159 grpc::ClientContext context;
160 flatbuffers::grpc::Message<Monster> request; // simulate invalid message
161 flatbuffers::grpc::Message<Stat> response;
162 auto status = stub->Store(&context, request, &response);
163 // The rpc status should be INTERNAL to indicate a verification error. This
164 // matches the protobuf gRPC status code for an unparseable message.
165 assert(!status.ok());
166 assert(status.error_code() == ::grpc::StatusCode::INTERNAL);
167 assert(strcmp(status.error_message().c_str(),
168 "Message verification failed") == 0);
169 }
170 #endif
171
172 server_instance->Shutdown();
173
174 server_thread.join();
175
176 delete server_instance;
177
178 return 0;
179 }
180
main(int,const char * [])181 int main(int /*argc*/, const char * /*argv*/[]) {
182 message_builder_tests();
183 grpc_server_test();
184
185 if (!testing_fails) {
186 TEST_OUTPUT_LINE("ALL TESTS PASSED");
187 return 0;
188 } else {
189 TEST_OUTPUT_LINE("%d FAILED TESTS", testing_fails);
190 return 1;
191 }
192 }
193