I changed the c++ HelloWorld example to use Bidi Async server. The modified
proto is as follows.
// The greeting service definition.
service Greeter {
// Sends a greeting
rpc SayHello (stream HelloMessage) returns (stream HelloMessage) {}
}
// The response message containing the greetings
message HelloMessage {
string message = 1;
}
The sync server works fine. However, the async server does not work. It
does not receive anything after calling stream.read(). Eventually I got the
error below.
Server listening on 0.0.0.0:50051
Greeter server received:
Greeter server replied: Hello
pure virtual method called
terminate called without an active exception
Aborted
I made the async server code based on the original greeter_async_server.cc
with slight changes. I paste my code code below for your reference. Can you
please take a look which part is wrong? How can I make it work? Thanks.
class ServerImpl final {
public:
~ServerImpl() {
server_->Shutdown();
// Always shutdown the completion queue after the server.
cq_->Shutdown();
}
// There is no shutdown handling in this code.
void Run() {
std::string server_address("0.0.0.0:50051");
ServerBuilder builder;
// Listen on the given address without any authentication mechanism.
builder.AddListeningPort(server_address,
grpc::InsecureServerCredentials());
// Register "service_" as the instance through which we'll communicate
with
// clients. In this case it corresponds to an *asynchronous* service.
builder.RegisterService(&service_);
// Get hold of the completion queue used for the asynchronous
communication
// with the gRPC runtime.
cq_ = builder.AddCompletionQueue();
// Finally assemble the server.
server_ = builder.BuildAndStart();
std::cout << "Server listening on " << server_address << std::endl;
// Proceed to the server's main loop.
HandleRpcs();
}
private:
// Class encompasing the state and logic needed to serve a request.
class CallData {
public:
// Take in the "service" instance (in this case representing an
asynchronous
// server) and the completion queue "cq" used for asynchronous
communication
// with the gRPC runtime.
CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
: service_(service), cq_(cq), stream_(&ctx_), status_(CREATE) {
// Invoke the serving logic right away.
Proceed();
}
void Proceed() {
if (status_ == CREATE) {
// Make this instance progress to the PROCESS state.
status_ = PROCESS;
// As part of the initial CREATE state, we *request* that the system
// start processing SayHello requests. In this request, "this" acts
are
// the tag uniquely identifying the request (so that different
CallData
// instances can serve different requests concurrently), in this
case
// the memory address of this CallData instance.
service_->RequestSayHello(&ctx_, &stream_, cq_, cq_, this);
} else if (status_ == PROCESS) {
// Spawn a new CallData instance to serve new clients while we
process
// the one for this CallData. The instance will deallocate itself as
// part of its FINISH state.
new CallData(service_, cq_);
// The actual processing.
std::string prefix("Hello ");
stream_.Read(&request_, this);
std::cout << "Greeter server received: " << request_.message() <<
std::endl;
reply_.set_message(prefix + request_.message());
std::cout << "Greeter server replied: " << reply_.message() <<
std::endl;
stream_.Write(reply_, this);
// And we are done! Let the gRPC runtime know we've finished, using
the
// memory address of this instance as the uniquely identifying tag
for
// the event.
status_ = FINISH;
stream_.Finish(Status::OK, this);
} else {
GPR_ASSERT(status_ == FINISH);
// Once in the FINISH state, deallocate ourselves (CallData).
delete this;
}
}
private:
// The means of communication with the gRPC runtime for an asynchronous
// server.
Greeter::AsyncService* service_;
// The producer-consumer queue where for asynchronous server
notifications.
ServerCompletionQueue* cq_;
// Context for the rpc, allowing to tweak aspects of it such as the use
// of compression, authentication, as well as to send metadata back to
the
// client.
ServerContext ctx_;
// What we get from the client.
HelloMessage request_;
// What we send back to the client.
HelloMessage reply_;
// The means to get back to the client.
ServerAsyncReaderWriter<HelloMessage, HelloMessage> stream_;
// Let's implement a tiny state machine with the following states.
enum CallStatus { CREATE, PROCESS, FINISH };
CallStatus status_; // The current serving state.
};
// This can be run in multiple threads if needed.
void HandleRpcs() {
// Spawn a new CallData instance to serve new clients.
new CallData(&service_, cq_.get());
void* tag; // uniquely identifies a request.
bool ok;
while (true) {
// Block waiting to read the next event from the completion queue. The
// event is uniquely identified by its tag, which in this case is the
// memory address of a CallData instance.
// The return value of Next should always be checked. This return
value
// tells us whether there is any kind of event or cq_ is shutting
down.
GPR_ASSERT(cq_->Next(&tag, &ok));
GPR_ASSERT(ok);
static_cast<CallData*>(tag)->Proceed();
}
}
std::unique_ptr<ServerCompletionQueue> cq_;
Greeter::AsyncService service_;
std::unique_ptr<Server> server_;
};
int main(int argc, char** argv) {
ServerImpl server;
server.Run();
return 0;
}
--
You received this message because you are subscribed to the Google Groups
"grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/grpc-io.
To view this discussion on the web visit
https://groups.google.com/d/msgid/grpc-io/84950f23-0368-4870-a5f7-935d6588edc6%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
