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