Merge pull request #7018 from vjpai/fc_unary
Support server-side streaming of a unary RPC
diff --git a/include/grpc++/ext/reflection.grpc.pb.h b/include/grpc++/ext/reflection.grpc.pb.h
index 0b4ef86..064117e 100644
--- a/include/grpc++/ext/reflection.grpc.pb.h
+++ b/include/grpc++/ext/reflection.grpc.pb.h
@@ -74,6 +74,7 @@
#include <grpc++/impl/codegen/async_stream.h>
#include <grpc++/impl/codegen/async_unary_call.h>
+#include <grpc++/impl/codegen/method_handler_impl.h>
#include <grpc++/impl/codegen/proto_utils.h>
#include <grpc++/impl/codegen/rpc_method.h>
#include <grpc++/impl/codegen/service_type.h>
@@ -174,6 +175,7 @@
return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
}
};
+ typedef Service StreamedUnaryService;
};
} // namespace v1alpha
diff --git a/include/grpc++/impl/codegen/call.h b/include/grpc++/impl/codegen/call.h
index dfac177..df225d3 100644
--- a/include/grpc++/impl/codegen/call.h
+++ b/include/grpc++/impl/codegen/call.h
@@ -662,10 +662,10 @@
call_hook_->PerformOpsOnCall(ops, this);
}
- grpc_call* call() { return call_; }
- CompletionQueue* cq() { return cq_; }
+ grpc_call* call() const { return call_; }
+ CompletionQueue* cq() const { return cq_; }
- int max_message_size() { return max_message_size_; }
+ int max_message_size() const { return max_message_size_; }
private:
CallHook* call_hook_;
diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index 03009e0..78bc5ca 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -68,8 +68,10 @@
class ServerReader;
template <class W>
class ServerWriter;
+namespace internal {
template <class W, class R>
-class ServerReaderWriter;
+class ServerReaderWriterBody;
+}
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -178,15 +180,15 @@
template <class W>
friend class ::grpc::ServerWriter;
template <class W, class R>
- friend class ::grpc::ServerReaderWriter;
+ friend class ::grpc::internal::ServerReaderWriterBody;
template <class ServiceType, class RequestType, class ResponseType>
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ClientStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ServerStreamingHandler;
- template <class ServiceType, class RequestType, class ResponseType>
- friend class BidiStreamingHandler;
+ template <class Streamer, bool WriteNeeded>
+ friend class TemplatedBidiStreamingHandler;
friend class UnknownMethodHandler;
friend class ::grpc::Server;
friend class ::grpc::ServerContext;
diff --git a/include/grpc++/impl/codegen/method_handler_impl.h b/include/grpc++/impl/codegen/method_handler_impl.h
index 2f4be64..d989263 100644
--- a/include/grpc++/impl/codegen/method_handler_impl.h
+++ b/include/grpc++/impl/codegen/method_handler_impl.h
@@ -167,20 +167,22 @@
};
// A wrapper class of an application provided bidi-streaming handler.
-template <class ServiceType, class RequestType, class ResponseType>
-class BidiStreamingHandler : public MethodHandler {
+// This also applies to server-streamed implementation of a unary method
+// with the additional requirement that such methods must have done a
+// write for status to be ok
+// Since this is used by more than 1 class, the service is not passed in.
+// Instead, it is expected to be an implicitly-captured argument of func
+// (through bind or something along those lines)
+template <class Streamer, bool WriteNeeded>
+class TemplatedBidiStreamingHandler : public MethodHandler {
public:
- BidiStreamingHandler(
- std::function<Status(ServiceType*, ServerContext*,
- ServerReaderWriter<ResponseType, RequestType>*)>
- func,
- ServiceType* service)
- : func_(func), service_(service) {}
+ TemplatedBidiStreamingHandler(
+ std::function<Status(ServerContext*, Streamer*)> func)
+ : func_(func), write_needed_(WriteNeeded) {}
void RunHandler(const HandlerParameter& param) GRPC_FINAL {
- ServerReaderWriter<ResponseType, RequestType> stream(param.call,
- param.server_context);
- Status status = func_(service_, param.server_context, &stream);
+ Streamer stream(param.call, param.server_context);
+ Status status = func_(param.server_context, &stream);
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> ops;
if (!param.server_context->sent_initial_metadata_) {
@@ -189,6 +191,12 @@
if (param.server_context->compression_level_set()) {
ops.set_compression_level(param.server_context->compression_level());
}
+ if (write_needed_ && status.ok()) {
+ // If we needed a write but never did one, we need to mark the
+ // status as a fail
+ status = Status(StatusCode::INTERNAL,
+ "Service did not provide response message");
+ }
}
ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
param.call->PerformOps(&ops);
@@ -196,10 +204,36 @@
}
private:
- std::function<Status(ServiceType*, ServerContext*,
- ServerReaderWriter<ResponseType, RequestType>*)>
- func_;
- ServiceType* service_;
+ std::function<Status(ServerContext*, Streamer*)> func_;
+ const bool write_needed_;
+};
+
+template <class ServiceType, class RequestType, class ResponseType>
+class BidiStreamingHandler
+ : public TemplatedBidiStreamingHandler<
+ ServerReaderWriter<ResponseType, RequestType>, false> {
+ public:
+ BidiStreamingHandler(
+ std::function<Status(ServiceType*, ServerContext*,
+ ServerReaderWriter<ResponseType, RequestType>*)>
+ func,
+ ServiceType* service)
+ : TemplatedBidiStreamingHandler<
+ ServerReaderWriter<ResponseType, RequestType>, false>(std::bind(
+ func, service, std::placeholders::_1, std::placeholders::_2)) {}
+};
+
+template <class RequestType, class ResponseType>
+class StreamedUnaryHandler
+ : public TemplatedBidiStreamingHandler<
+ ServerUnaryStreamer<RequestType, ResponseType>, true> {
+ public:
+ explicit StreamedUnaryHandler(
+ std::function<Status(ServerContext*,
+ ServerUnaryStreamer<RequestType, ResponseType>*)>
+ func)
+ : TemplatedBidiStreamingHandler<
+ ServerUnaryStreamer<RequestType, ResponseType>, true>(func) {}
};
// Handle unknown method by returning UNIMPLEMENTED error.
diff --git a/include/grpc++/impl/codegen/rpc_method.h b/include/grpc++/impl/codegen/rpc_method.h
index 39cb4f7..4897428 100644
--- a/include/grpc++/impl/codegen/rpc_method.h
+++ b/include/grpc++/impl/codegen/rpc_method.h
@@ -60,11 +60,12 @@
const char* name() const { return name_; }
RpcType method_type() const { return method_type_; }
+ void SetMethodType(RpcType type) { method_type_ = type; }
void* channel_tag() const { return channel_tag_; }
private:
const char* const name_;
- const RpcType method_type_;
+ RpcType method_type_;
void* const channel_tag_;
};
diff --git a/include/grpc++/impl/codegen/rpc_service_method.h b/include/grpc++/impl/codegen/rpc_service_method.h
index 8b1f026..52124fb 100644
--- a/include/grpc++/impl/codegen/rpc_service_method.h
+++ b/include/grpc++/impl/codegen/rpc_service_method.h
@@ -82,6 +82,7 @@
// if MethodHandler is nullptr, then this is an async method
MethodHandler* handler() const { return handler_.get(); }
void ResetHandler() { handler_.reset(); }
+ void SetHandler(MethodHandler* handler) { handler_.reset(handler); }
private:
void* server_tag_;
diff --git a/include/grpc++/impl/codegen/server_context.h b/include/grpc++/impl/codegen/server_context.h
index c9d1f4d..379c9f7 100644
--- a/include/grpc++/impl/codegen/server_context.h
+++ b/include/grpc++/impl/codegen/server_context.h
@@ -65,8 +65,10 @@
class ServerReader;
template <class W>
class ServerWriter;
+namespace internal {
template <class W, class R>
-class ServerReaderWriter;
+class ServerReaderWriterBody;
+}
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -187,15 +189,15 @@
template <class W>
friend class ::grpc::ServerWriter;
template <class W, class R>
- friend class ::grpc::ServerReaderWriter;
+ friend class ::grpc::internal::ServerReaderWriterBody;
template <class ServiceType, class RequestType, class ResponseType>
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ClientStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ServerStreamingHandler;
- template <class ServiceType, class RequestType, class ResponseType>
- friend class BidiStreamingHandler;
+ template <class Streamer, bool WriteNeeded>
+ friend class TemplatedBidiStreamingHandler;
friend class UnknownMethodHandler;
friend class ::grpc::ClientContext;
diff --git a/include/grpc++/impl/codegen/service_type.h b/include/grpc++/impl/codegen/service_type.h
index c19dfc7..72b2225 100644
--- a/include/grpc++/impl/codegen/service_type.h
+++ b/include/grpc++/impl/codegen/service_type.h
@@ -147,6 +147,17 @@
methods_[index].reset();
}
+ void MarkMethodStreamedUnary(int index,
+ MethodHandler* streamed_unary_method) {
+ GPR_CODEGEN_ASSERT(methods_[index] && methods_[index]->handler() &&
+ "Cannot mark an async or generic method Streamed Unary");
+ methods_[index]->SetHandler(streamed_unary_method);
+
+ // From the server's point of view, streamed unary is a special
+ // case of BIDI_STREAMING that has 1 read and 1 write, in that order.
+ methods_[index]->SetMethodType(::grpc::RpcMethod::BIDI_STREAMING);
+ }
+
private:
friend class Server;
friend class ServerInterface;
diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h
index 7601cea..e1d4660 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -79,6 +79,9 @@
public:
virtual ~ReaderInterface() {}
+ /// Upper bound on the next message size available for reading on this stream
+ virtual bool NextMessageSize(uint32_t* sz) = 0;
+
/// Blocking read a message and parse to \a msg. Returns \a true on success.
/// This is thread-safe with respect to \a Write or \WritesDone methods on
/// the same stream. It should not be called concurrently with another \a
@@ -157,6 +160,11 @@
cq_.Pluck(&ops); /// status ignored
}
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = call_.max_message_size();
+ return true;
+ }
+
bool Read(R* msg) GRPC_OVERRIDE {
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops;
if (!context_->initial_metadata_received_) {
@@ -302,6 +310,11 @@
cq_.Pluck(&ops); // status ignored
}
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = call_.max_message_size();
+ return true;
+ }
+
bool Read(R* msg) GRPC_OVERRIDE {
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops;
if (!context_->initial_metadata_received_) {
@@ -369,6 +382,11 @@
call_->cq()->Pluck(&ops);
}
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = call_->max_message_size();
+ return true;
+ }
+
bool Read(R* msg) GRPC_OVERRIDE {
CallOpSet<CallOpRecvMessage<R>> ops;
ops.RecvMessage(msg);
@@ -434,12 +452,15 @@
public WriterInterface<W>,
public ReaderInterface<R> {};
+// Actual implementation of bi-directional streaming
+namespace internal {
template <class W, class R>
-class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
+class ServerReaderWriterBody GRPC_FINAL {
public:
- ServerReaderWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
+ ServerReaderWriterBody(Call* call, ServerContext* ctx)
+ : call_(call), ctx_(ctx) {}
- void SendInitialMetadata() GRPC_OVERRIDE {
+ void SendInitialMetadata() {
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
CallOpSet<CallOpSendInitialMetadata> ops;
@@ -453,15 +474,19 @@
call_->cq()->Pluck(&ops);
}
- bool Read(R* msg) GRPC_OVERRIDE {
+ bool NextMessageSize(uint32_t* sz) {
+ *sz = call_->max_message_size();
+ return true;
+ }
+
+ bool Read(R* msg) {
CallOpSet<CallOpRecvMessage<R>> ops;
ops.RecvMessage(msg);
call_->PerformOps(&ops);
return call_->cq()->Pluck(&ops) && ops.got_message;
}
- using WriterInterface<W>::Write;
- bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
+ bool Write(const W& msg, const WriteOptions& options) {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
if (!ops.SendMessage(msg, options).ok()) {
return false;
@@ -482,6 +507,76 @@
Call* const call_;
ServerContext* const ctx_;
};
+}
+
+// class to represent the user API for a bidirectional streaming call
+template <class W, class R>
+class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
+ public:
+ ServerReaderWriter(Call* call, ServerContext* ctx) : body_(call, ctx) {}
+
+ void SendInitialMetadata() GRPC_OVERRIDE { body_.SendInitialMetadata(); }
+
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ return body_.NextMessageSize(sz);
+ }
+
+ bool Read(R* msg) GRPC_OVERRIDE { return body_.Read(msg); }
+
+ using WriterInterface<W>::Write;
+ bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
+ return body_.Write(msg, options);
+ }
+
+ private:
+ internal::ServerReaderWriterBody<W, R> body_;
+};
+
+/// A class to represent a flow-controlled unary call. This is something
+/// of a hybrid between conventional unary and streaming. This is invoked
+/// through a unary call on the client side, but the server responds to it
+/// as though it were a single-ping-pong streaming call. The server can use
+/// the \a NextMessageSize method to determine an upper-bound on the size of
+/// the message.
+/// A key difference relative to streaming: ServerUnaryStreamer
+/// must have exactly 1 Read and exactly 1 Write, in that order, to function
+/// correctly. Otherwise, the RPC is in error.
+template <class RequestType, class ResponseType>
+class ServerUnaryStreamer GRPC_FINAL
+ : public ServerReaderWriterInterface<ResponseType, RequestType> {
+ public:
+ ServerUnaryStreamer(Call* call, ServerContext* ctx)
+ : body_(call, ctx), read_done_(false), write_done_(false) {}
+
+ void SendInitialMetadata() GRPC_OVERRIDE { body_.SendInitialMetadata(); }
+
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ return body_.NextMessageSize(sz);
+ }
+
+ bool Read(RequestType* request) GRPC_OVERRIDE {
+ if (read_done_) {
+ return false;
+ }
+ read_done_ = true;
+ return body_.Read(request);
+ }
+
+ using WriterInterface<ResponseType>::Write;
+ bool Write(const ResponseType& response,
+ const WriteOptions& options) GRPC_OVERRIDE {
+ if (write_done_ || !read_done_) {
+ return false;
+ }
+ write_done_ = true;
+ return body_.Write(response, options);
+ }
+
+ private:
+ internal::ServerReaderWriterBody<ResponseType, RequestType> body_;
+ bool read_done_;
+ bool write_done_;
+};
} // namespace grpc
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index c386115..d0c35ea 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -130,6 +130,7 @@
static const char *headers_strs[] = {
"grpc++/impl/codegen/async_stream.h",
"grpc++/impl/codegen/async_unary_call.h",
+ "grpc++/impl/codegen/method_handler_impl.h",
"grpc++/impl/codegen/proto_utils.h",
"grpc++/impl/codegen/rpc_method.h",
"grpc++/impl/codegen/service_type.h",
@@ -604,6 +605,57 @@
printer->Print(*vars, "};\n");
}
+void PrintHeaderServerMethodStreamedUnary(
+ Printer *printer, const Method *method,
+ std::map<grpc::string, grpc::string> *vars) {
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] = method->input_type_name();
+ (*vars)["Response"] = method->output_type_name();
+ if (method->NoStreaming()) {
+ printer->Print(*vars, "template <class BaseClass>\n");
+ printer->Print(*vars,
+ "class WithStreamedUnaryMethod_$Method$ : "
+ "public BaseClass {\n");
+ printer->Print(
+ " private:\n"
+ " void BaseClassMustBeDerivedFromService(const Service *service) "
+ "{}\n");
+ printer->Print(" public:\n");
+ printer->Indent();
+ printer->Print(*vars,
+ "WithStreamedUnaryMethod_$Method$() {\n"
+ " ::grpc::Service::MarkMethodStreamedUnary($Idx$,\n"
+ " new ::grpc::StreamedUnaryHandler< $Request$, "
+ "$Response$>(std::bind"
+ "(&WithStreamedUnaryMethod_$Method$<BaseClass>::"
+ "Streamed$Method$, this, std::placeholders::_1, "
+ "std::placeholders::_2)));\n"
+ "}\n");
+ printer->Print(*vars,
+ "~WithStreamedUnaryMethod_$Method$() GRPC_OVERRIDE {\n"
+ " BaseClassMustBeDerivedFromService(this);\n"
+ "}\n");
+ printer->Print(
+ *vars,
+ "// disable regular version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, const $Request$* request, "
+ "$Response$* response) GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ printer->Print(*vars,
+ "// replace default version of method with streamed unary\n"
+ "virtual ::grpc::Status Streamed$Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerUnaryStreamer< "
+ "$Request$,$Response$>* server_unary_streamer)"
+ " = 0;\n");
+ printer->Outdent();
+ printer->Print(*vars, "};\n");
+ }
+}
+
void PrintHeaderServerMethodGeneric(
Printer *printer, const Method *method,
std::map<grpc::string, grpc::string> *vars) {
@@ -770,6 +822,28 @@
PrintHeaderServerMethodGeneric(printer, service->method(i).get(), vars);
}
+ // Server side - Streamed Unary
+ for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["Idx"] = as_string(i);
+ PrintHeaderServerMethodStreamedUnary(printer, service->method(i).get(),
+ vars);
+ }
+
+ printer->Print("typedef ");
+ for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["method_name"] = service->method(i).get()->name();
+ if (service->method(i)->NoStreaming()) {
+ printer->Print(*vars, "WithStreamedUnaryMethod_$method_name$<");
+ }
+ }
+ printer->Print("Service");
+ for (int i = 0; i < service->method_count(); ++i) {
+ if (service->method(i)->NoStreaming()) {
+ printer->Print(" >");
+ }
+ }
+ printer->Print(" StreamedUnaryService;\n");
+
printer->Outdent();
printer->Print("};\n");
printer->Print(service->GetTrailingComments().c_str());
@@ -1080,6 +1154,9 @@
(*vars)["Idx"] = as_string(i);
if (method->NoStreaming()) {
(*vars)["StreamingType"] = "NORMAL_RPC";
+ // NOTE: There is no reason to consider streamed-unary as a separate
+ // category here since this part is setting up the client-side stub
+ // and this appears as a NORMAL_RPC from the client-side.
} else if (method->ClientOnlyStreaming()) {
(*vars)["StreamingType"] = "CLIENT_STREAMING";
} else if (method->ServerOnlyStreaming()) {
diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden
index ef3d1aa..7b0fd6c 100644
--- a/test/cpp/codegen/compiler_test_golden
+++ b/test/cpp/codegen/compiler_test_golden
@@ -43,6 +43,7 @@
#include <grpc++/impl/codegen/async_stream.h>
#include <grpc++/impl/codegen/async_unary_call.h>
+#include <grpc++/impl/codegen/method_handler_impl.h>
#include <grpc++/impl/codegen/proto_utils.h>
#include <grpc++/impl/codegen/rpc_method.h>
#include <grpc++/impl/codegen/service_type.h>
@@ -206,6 +207,27 @@
return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
}
};
+ template <class BaseClass>
+ class WithStreamedUnaryMethod_MethodA1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithStreamedUnaryMethod_MethodA1() {
+ ::grpc::Service::MarkMethodStreamedUnary(0,
+ new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodA1<BaseClass>::StreamedMethodA1, this, std::placeholders::_1, std::placeholders::_2)));
+ }
+ ~WithStreamedUnaryMethod_MethodA1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable regular version of this method
+ ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ // replace default version of method with streamed unary
+ virtual ::grpc::Status StreamedMethodA1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0;
+ };
+ typedef WithStreamedUnaryMethod_MethodA1<Service > StreamedUnaryService;
};
// ServiceB leading comment 1
@@ -284,6 +306,27 @@
return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
}
};
+ template <class BaseClass>
+ class WithStreamedUnaryMethod_MethodB1 : public BaseClass {
+ private:
+ void BaseClassMustBeDerivedFromService(const Service *service) {}
+ public:
+ WithStreamedUnaryMethod_MethodB1() {
+ ::grpc::Service::MarkMethodStreamedUnary(0,
+ new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodB1<BaseClass>::StreamedMethodB1, this, std::placeholders::_1, std::placeholders::_2)));
+ }
+ ~WithStreamedUnaryMethod_MethodB1() GRPC_OVERRIDE {
+ BaseClassMustBeDerivedFromService(this);
+ }
+ // disable regular version of this method
+ ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) GRPC_FINAL GRPC_OVERRIDE {
+ abort();
+ return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+ }
+ // replace default version of method with streamed unary
+ virtual ::grpc::Status StreamedMethodB1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0;
+ };
+ typedef WithStreamedUnaryMethod_MethodB1<Service > StreamedUnaryService;
};
// ServiceB trailing comment 1
diff --git a/test/cpp/end2end/hybrid_end2end_test.cc b/test/cpp/end2end/hybrid_end2end_test.cc
index 7e0c0e8..a6ea13a 100644
--- a/test/cpp/end2end/hybrid_end2end_test.cc
+++ b/test/cpp/end2end/hybrid_end2end_test.cc
@@ -199,7 +199,8 @@
HybridEnd2endTest() {}
void SetUpServer(::grpc::Service* service1, ::grpc::Service* service2,
- AsyncGenericService* generic_service) {
+ AsyncGenericService* generic_service,
+ int max_message_size = 0) {
int port = grpc_pick_unused_port_or_die();
server_address_ << "localhost:" << port;
@@ -217,6 +218,11 @@
if (generic_service) {
builder.RegisterAsyncGenericService(generic_service);
}
+
+ if (max_message_size != 0) {
+ builder.SetMaxMessageSize(max_message_size);
+ }
+
// Create a separate cq for each potential handler.
for (int i = 0; i < 5; i++) {
cqs_.push_back(builder.AddCompletionQueue(false));
@@ -415,6 +421,83 @@
request_stream_handler_thread.join();
}
+// Add a second service with one sync streamed unary method.
+class StreamedUnaryDupPkg
+ : public duplicate::EchoTestService::WithStreamedUnaryMethod_Echo<
+ TestServiceImplDupPkg> {
+ public:
+ Status StreamedEcho(ServerContext* context,
+ ServerUnaryStreamer<EchoRequest, EchoResponse>* stream)
+ GRPC_OVERRIDE {
+ EchoRequest req;
+ EchoResponse resp;
+ uint32_t next_msg_sz;
+ stream->NextMessageSize(&next_msg_sz);
+ gpr_log(GPR_INFO, "Streamed Unary Next Message Size is %u", next_msg_sz);
+ GPR_ASSERT(stream->Read(&req));
+ resp.set_message(req.message() + "_dup");
+ GPR_ASSERT(stream->Write(resp));
+ return Status::OK;
+ }
+};
+
+TEST_F(HybridEnd2endTest,
+ AsyncRequestStreamResponseStream_SyncStreamedUnaryDupService) {
+ typedef EchoTestService::WithAsyncMethod_RequestStream<
+ EchoTestService::WithAsyncMethod_ResponseStream<TestServiceImpl>>
+ SType;
+ SType service;
+ StreamedUnaryDupPkg dup_service;
+ SetUpServer(&service, &dup_service, nullptr, 8192);
+ ResetStub();
+ std::thread response_stream_handler_thread(HandleServerStreaming<SType>,
+ &service, cqs_[0].get());
+ std::thread request_stream_handler_thread(HandleClientStreaming<SType>,
+ &service, cqs_[1].get());
+ TestAllMethods();
+ SendEchoToDupService();
+ response_stream_handler_thread.join();
+ request_stream_handler_thread.join();
+}
+
+// Add a second service that is fully Streamed Unary
+class FullyStreamedUnaryDupPkg
+ : public duplicate::EchoTestService::StreamedUnaryService {
+ public:
+ Status StreamedEcho(ServerContext* context,
+ ServerUnaryStreamer<EchoRequest, EchoResponse>* stream)
+ GRPC_OVERRIDE {
+ EchoRequest req;
+ EchoResponse resp;
+ uint32_t next_msg_sz;
+ stream->NextMessageSize(&next_msg_sz);
+ gpr_log(GPR_INFO, "Streamed Unary Next Message Size is %u", next_msg_sz);
+ GPR_ASSERT(stream->Read(&req));
+ resp.set_message(req.message() + "_dup");
+ GPR_ASSERT(stream->Write(resp));
+ return Status::OK;
+ }
+};
+
+TEST_F(HybridEnd2endTest,
+ AsyncRequestStreamResponseStream_SyncFullyStreamedUnaryDupService) {
+ typedef EchoTestService::WithAsyncMethod_RequestStream<
+ EchoTestService::WithAsyncMethod_ResponseStream<TestServiceImpl>>
+ SType;
+ SType service;
+ FullyStreamedUnaryDupPkg dup_service;
+ SetUpServer(&service, &dup_service, nullptr, 8192);
+ ResetStub();
+ std::thread response_stream_handler_thread(HandleServerStreaming<SType>,
+ &service, cqs_[0].get());
+ std::thread request_stream_handler_thread(HandleClientStreaming<SType>,
+ &service, cqs_[1].get());
+ TestAllMethods();
+ SendEchoToDupService();
+ response_stream_handler_thread.join();
+ request_stream_handler_thread.join();
+}
+
// Add a second service with one async method.
TEST_F(HybridEnd2endTest, AsyncRequestStreamResponseStream_AsyncDupService) {
typedef EchoTestService::WithAsyncMethod_RequestStream<
diff --git a/test/cpp/end2end/mock_test.cc b/test/cpp/end2end/mock_test.cc
index 0ace5d9..4052627 100644
--- a/test/cpp/end2end/mock_test.cc
+++ b/test/cpp/end2end/mock_test.cc
@@ -31,6 +31,7 @@
*
*/
+#include <climits>
#include <thread>
#include <grpc++/channel.h>
@@ -63,6 +64,10 @@
: public ClientReaderWriterInterface<W, R> {
public:
void WaitForInitialMetadata() GRPC_OVERRIDE {}
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = UINT_MAX;
+ return true;
+ }
bool Read(R* msg) GRPC_OVERRIDE { return true; }
bool Write(const W& msg) GRPC_OVERRIDE { return true; }
bool WritesDone() GRPC_OVERRIDE { return true; }
@@ -74,6 +79,10 @@
public:
MockClientReaderWriter() : writes_done_(false) {}
void WaitForInitialMetadata() GRPC_OVERRIDE {}
+ bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = UINT_MAX;
+ return true;
+ }
bool Read(EchoResponse* msg) GRPC_OVERRIDE {
if (writes_done_) return false;
msg->set_message(last_message_);