first sets of changes, it builds
diff --git a/Makefile b/Makefile
index 186aec6..f68228f 100644
--- a/Makefile
+++ b/Makefile
@@ -3085,16 +3085,20 @@
src/cpp/common/completion_queue.cc \
src/cpp/common/rpc_method.cc \
src/cpp/proto/proto_utils.cc \
+ src/cpp/server/anonymous_service.cc \
src/cpp/server/server.cc \
src/cpp/server/server_builder.cc \
src/cpp/server/server_context.cc \
src/cpp/server/server_credentials.cc \
src/cpp/server/thread_pool.cc \
+ src/cpp/util/byte_buffer.cc \
src/cpp/util/status.cc \
src/cpp/util/time.cc \
PUBLIC_HEADERS_CXX += \
+ include/grpc++/anonymous_service.h \
include/grpc++/async_unary_call.h \
+ include/grpc++/byte_buffer.h \
include/grpc++/channel_arguments.h \
include/grpc++/channel_interface.h \
include/grpc++/client_context.h \
@@ -3162,11 +3166,13 @@
src/cpp/common/completion_queue.cc: $(OPENSSL_DEP)
src/cpp/common/rpc_method.cc: $(OPENSSL_DEP)
src/cpp/proto/proto_utils.cc: $(OPENSSL_DEP)
+src/cpp/server/anonymous_service.cc: $(OPENSSL_DEP)
src/cpp/server/server.cc: $(OPENSSL_DEP)
src/cpp/server/server_builder.cc: $(OPENSSL_DEP)
src/cpp/server/server_context.cc: $(OPENSSL_DEP)
src/cpp/server/server_credentials.cc: $(OPENSSL_DEP)
src/cpp/server/thread_pool.cc: $(OPENSSL_DEP)
+src/cpp/util/byte_buffer.cc: $(OPENSSL_DEP)
src/cpp/util/status.cc: $(OPENSSL_DEP)
src/cpp/util/time.cc: $(OPENSSL_DEP)
endif
@@ -3221,11 +3227,13 @@
$(OBJDIR)/$(CONFIG)/src/cpp/common/completion_queue.o:
$(OBJDIR)/$(CONFIG)/src/cpp/common/rpc_method.o:
$(OBJDIR)/$(CONFIG)/src/cpp/proto/proto_utils.o:
+$(OBJDIR)/$(CONFIG)/src/cpp/server/anonymous_service.o:
$(OBJDIR)/$(CONFIG)/src/cpp/server/server.o:
$(OBJDIR)/$(CONFIG)/src/cpp/server/server_builder.o:
$(OBJDIR)/$(CONFIG)/src/cpp/server/server_context.o:
$(OBJDIR)/$(CONFIG)/src/cpp/server/server_credentials.o:
$(OBJDIR)/$(CONFIG)/src/cpp/server/thread_pool.o:
+$(OBJDIR)/$(CONFIG)/src/cpp/util/byte_buffer.o:
$(OBJDIR)/$(CONFIG)/src/cpp/util/status.o:
$(OBJDIR)/$(CONFIG)/src/cpp/util/time.o:
diff --git a/build.json b/build.json
index fdf87f3..7dc792b 100644
--- a/build.json
+++ b/build.json
@@ -393,7 +393,9 @@
"build": "all",
"language": "c++",
"public_headers": [
+ "include/grpc++/anonymous_service.h",
"include/grpc++/async_unary_call.h",
+ "include/grpc++/byte_buffer.h",
"include/grpc++/channel_arguments.h",
"include/grpc++/channel_interface.h",
"include/grpc++/client_context.h",
@@ -434,11 +436,13 @@
"src/cpp/common/completion_queue.cc",
"src/cpp/common/rpc_method.cc",
"src/cpp/proto/proto_utils.cc",
+ "src/cpp/server/anonymous_service.cc",
"src/cpp/server/server.cc",
"src/cpp/server/server_builder.cc",
"src/cpp/server/server_context.cc",
"src/cpp/server/server_credentials.cc",
"src/cpp/server/thread_pool.cc",
+ "src/cpp/util/byte_buffer.cc",
"src/cpp/util/status.cc",
"src/cpp/util/time.cc"
],
diff --git a/include/grpc++/anonymous_service.h b/include/grpc++/anonymous_service.h
index 81b18ea..456fb8b 100644
--- a/include/grpc++/anonymous_service.h
+++ b/include/grpc++/anonymous_service.h
@@ -31,14 +31,17 @@
*
*/
-#ifndef __GRPCPP_ANONYMOUS_SERVICE_H_
-#define __GRPCPP_ANONYMOUS_SERVICE_H_
+#ifndef GRPCXX_ANONYMOUS_SERVICE_H
+#define GRPCXX_ANONYMOUS_SERVICE_H
+#include <grpc++/byte_buffer.h>
#include <grpc++/stream.h>
+struct grpc_server;
+
namespace grpc {
-typedef ServerReaderWriter<ByteBuffer, ByteBuffer> GenericServerReaderWriter;
+typedef ServerAsyncReaderWriter<ByteBuffer, ByteBuffer> GenericServerReaderWriter;
class AnonymousServerContext : public ServerContext {
public:
@@ -50,23 +53,20 @@
grpc::string host_;
};
-// Anonymous stubs provide a type-unsafe interface to call gRPC methods
-// by name.
class AnonymousService {
public:
- explicit AnonymousService(CompletionQueue* cq) : cq_(cq) {}
+ // TODO(yangg) Once we can add multiple completion queues to the server
+ // in c core, add a CompletionQueue* argument to the ctor here.
+ AnonymousService() : server_(nullptr) {}
- struct CallDetails {
- grpc::string method;
- grpc::string host;
- };
-
- void RequestCall(AnonymousServerContext* ctx, GenericServerReaderWriter* reader_writer, CompletionQueue* cq, void* tag);
-
+ void RequestCall(AnonymousServerContext* ctx,
+ GenericServerReaderWriter* reader_writer,
+ CompletionQueue* cq, void* tag);
private:
- CompletionQueue* const cq_;
+ friend class Server;
+ Server* server_;
};
-} // namespace
+} // namespace grpc
-#endif
+#endif // GRPCXX_ANONYMOUS_SERVICE_H
diff --git a/include/grpc++/anonymous_stub.h b/include/grpc++/anonymous_stub.h
index 4a7543c..ccb474e 100644
--- a/include/grpc++/anonymous_stub.h
+++ b/include/grpc++/anonymous_stub.h
@@ -31,14 +31,15 @@
*
*/
-#ifndef __GRPCPP_ANONYMOUS_STUB_H_
-#define __GRPCPP_ANONYMOUS_STUB_H_
+#ifndef GRPCXX_ANONYMOUS_STUB_H
+#define GRPCXX_ANONYMOUS_STUB_H
+#include <grpc++/byte_buffer.h>
#include <grpc++/stream.h>
namespace grpc {
-typedef ClientReaderWriter<ByteBuffer, ByteBuffer> GenericClientReaderWriter;
+typedef ClientAsyncReaderWriter<ByteBuffer, ByteBuffer> GenericClientReaderWriter;
// Anonymous stubs provide a type-unsafe interface to call gRPC methods
// by name.
@@ -53,6 +54,6 @@
std::shared_ptr<ChannelInterface> channel_;
};
-} // namespace
+} // namespace grpc
-#endif
+#endif // GRPCXX_ANONYMOUS_STUB_H
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index e3ba93e..7e8f6d4 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -48,6 +48,8 @@
struct grpc_server;
namespace grpc {
+class AnonymousServerContext;
+class AnonymousService;
class AsynchronousService;
class RpcService;
class RpcServiceMethod;
@@ -69,6 +71,7 @@
void Wait();
private:
+ friend class AnonymousService;
friend class ServerBuilder;
class SyncRequest;
@@ -82,6 +85,7 @@
// The service must exist for the lifetime of the Server instance.
bool RegisterService(RpcService* service);
bool RegisterAsyncService(AsynchronousService* service);
+ void RegisterAnonymousService(AnonymousService* service);
// Add a listening port. Can be called multiple times.
int AddPort(const grpc::string& addr);
// Start the server.
@@ -99,6 +103,10 @@
ServerAsyncStreamingInterface* stream,
CompletionQueue* cq, void* tag);
+ void RequestAsyncAnonymousCall(AnonymousServerContext* context,
+ ServerAsyncStreamingInterface* stream,
+ CompletionQueue* cq, void* tag);
+
// Completion queue.
CompletionQueue cq_;
diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h
index 60c891c..514a668 100644
--- a/include/grpc++/server_builder.h
+++ b/include/grpc++/server_builder.h
@@ -87,6 +87,7 @@
std::vector<AsynchronousService*> async_services_;
std::vector<grpc::string> ports_;
std::shared_ptr<ServerCredentials> creds_;
+ AnonymousService* anonymous_service_;
ThreadPoolInterface* thread_pool_;
};
diff --git a/include/grpc++/server_context.h b/include/grpc++/server_context.h
index a986fff..9e3b80c 100644
--- a/include/grpc++/server_context.h
+++ b/include/grpc++/server_context.h
@@ -66,7 +66,7 @@
class Server;
// Interface of server side rpc context.
-class ServerContext GRPC_FINAL {
+class ServerContext {
public:
ServerContext(); // for async calls
~ServerContext();
diff --git a/src/cpp/server/anonymous_service.cc b/src/cpp/server/anonymous_service.cc
new file mode 100644
index 0000000..ef20cad
--- /dev/null
+++ b/src/cpp/server/anonymous_service.cc
@@ -0,0 +1,47 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc++/anonymous_service.h>
+
+#include <grpc++/server.h>
+
+namespace grpc {
+
+void AnonymousService::RequestCall(AnonymousServerContext* ctx,
+ GenericServerReaderWriter* reader_writer,
+ CompletionQueue* cq, void* tag) {
+ server_->RequestAsyncAnonymousCall(ctx, reader_writer, cq, tag);
+}
+
+} // namespace grpc
+
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 2a5a7fe..4de1ad2 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -37,6 +37,7 @@
#include <grpc/grpc.h>
#include <grpc/grpc_security.h>
#include <grpc/support/log.h>
+#include <grpc++/anonymous_service.h>
#include <grpc++/completion_queue.h>
#include <grpc++/impl/rpc_service_method.h>
#include <grpc++/impl/service_type.h>
@@ -239,6 +240,12 @@
return true;
}
+void Server::RegisterAnonymousService(AnonymousService* service) {
+ GPR_ASSERT(service->server_ == nullptr &&
+ "Can only register an anonymous service against one server.");
+ service->server_ = this;
+}
+
int Server::AddPort(const grpc::string& addr) {
GPR_ASSERT(!started_);
if (secure_) {
@@ -306,15 +313,36 @@
stream_(stream),
cq_(cq),
ctx_(ctx),
+ anonymous_ctx_(nullptr),
server_(server),
call_(nullptr),
payload_(nullptr) {
memset(&array_, 0, sizeof(array_));
+ grpc_call_details_init(&call_details_);
grpc_server_request_registered_call(
- server->server_, registered_method, &call_, &deadline_, &array_,
- request ? &payload_ : nullptr, cq->cq(), this);
+ server->server_, registered_method, &call_, &call_details_.deadline,
+ &array_, request ? &payload_ : nullptr, cq->cq(), this);
}
+ AsyncRequest(Server* server, AnonymousServerContext* ctx,
+ ServerAsyncStreamingInterface* stream, CompletionQueue* cq,
+ void* tag)
+ : tag_(tag),
+ request_(nullptr),
+ stream_(stream),
+ cq_(cq),
+ ctx_(nullptr),
+ anonymous_ctx_(ctx),
+ server_(server),
+ call_(nullptr),
+ payload_(nullptr) {
+ memset(&array_, 0, sizeof(array_));
+ grpc_call_details_init(&call_details_);
+ grpc_server_request_call(
+ server->server_, &call_, &call_details_, &array_, cq->cq(), this);
+ }
+
+
~AsyncRequest() {
if (payload_) {
grpc_byte_buffer_destroy(payload_);
@@ -332,20 +360,22 @@
*status = false;
}
}
+ ServerContext* ctx = ctx_ ? ctx_ : anonymous_ctx_;
+ GPR_ASSERT(ctx);
if (*status) {
- ctx_->deadline_ = Timespec2Timepoint(deadline_);
+ ctx->deadline_ = Timespec2Timepoint(call_details_.deadline);
for (size_t i = 0; i < array_.count; i++) {
- ctx_->client_metadata_.insert(std::make_pair(
+ ctx->client_metadata_.insert(std::make_pair(
grpc::string(array_.metadata[i].key),
grpc::string(
array_.metadata[i].value,
array_.metadata[i].value + array_.metadata[i].value_length)));
}
}
- ctx_->call_ = call_;
+ ctx->call_ = call_;
Call call(call_, server_, cq_);
if (orig_status && call_) {
- ctx_->BeginCompletionOp(&call);
+ ctx->BeginCompletionOp(&call);
}
// just the pointers inside call are copied here
stream_->BindCall(&call);
@@ -359,9 +389,10 @@
ServerAsyncStreamingInterface* const stream_;
CompletionQueue* const cq_;
ServerContext* const ctx_;
+ AnonymousServerContext* const anonymous_ctx_;
Server* const server_;
grpc_call* call_;
- gpr_timespec deadline_;
+ grpc_call_details call_details_;
grpc_metadata_array array_;
grpc_byte_buffer* payload_;
};
@@ -372,6 +403,11 @@
CompletionQueue* cq, void* tag) {
new AsyncRequest(this, registered_method, context, request, stream, cq, tag);
}
+void Server::RequestAsyncAnonymousCall(AnonymousServerContext* context,
+ ServerAsyncStreamingInterface* stream,
+ CompletionQueue* cq, void* tag) {
+ new AsyncRequest(this, context, stream, cq, tag);
+}
void Server::ScheduleCallback() {
{
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index ae60f3d..cc6f8ca 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -51,6 +51,16 @@
async_services_.push_back(service);
}
+void ServerBuilder::RegisterAnonymousService(AnonymousService* service) {
+ if (anonymous_service_) {
+ gpr_log(GPR_ERROR,
+ "Adding multiple AnonymousService is unsupported for now. "
+ "Dropping the service %p", service);
+ return;
+ }
+ anonymous_service_ = service;
+}
+
void ServerBuilder::AddPort(const grpc::string& addr) {
ports_.push_back(addr);
}
@@ -89,6 +99,9 @@
return nullptr;
}
}
+ if (anonymous_service_) {
+ server->RegisterAnonymousService(anonymous_service_);
+ }
for (auto& port : ports_) {
if (!server->AddPort(port)) {
return nullptr;
diff --git a/src/cpp/util/byte_buffer.cc b/src/cpp/util/byte_buffer.cc
new file mode 100644
index 0000000..3846156
--- /dev/null
+++ b/src/cpp/util/byte_buffer.cc
@@ -0,0 +1,2 @@
+
+#include <grpc++/byte_buffer.h>