first sets of changes, it builds

pull/1003/head
Yang Gao 10 years ago
parent c79a57c3a9
commit 1c40233814
  1. 8
      Makefile
  2. 4
      build.json
  3. 32
      include/grpc++/anonymous_service.h
  4. 11
      include/grpc++/anonymous_stub.h
  5. 8
      include/grpc++/server.h
  6. 1
      include/grpc++/server_builder.h
  7. 2
      include/grpc++/server_context.h
  8. 47
      src/cpp/server/anonymous_service.cc
  9. 50
      src/cpp/server/server.cc
  10. 13
      src/cpp/server/server_builder.cc
  11. 2
      src/cpp/util/byte_buffer.cc

@ -3085,16 +3085,20 @@ LIBGRPC++_SRC = \
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/call.cc: $(OPENSSL_DEP)
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/call.o:
$(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:

@ -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"
],

@ -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 @@ class AnonymousServerContext : public ServerContext {
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) {}
struct CallDetails {
grpc::string method;
grpc::string host;
};
void RequestCall(AnonymousServerContext* ctx, GenericServerReaderWriter* reader_writer, CompletionQueue* cq, void* tag);
// 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) {}
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

@ -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 @@ class AnonymousStub {
std::shared_ptr<ChannelInterface> channel_;
};
} // namespace
} // namespace grpc
#endif
#endif // GRPCXX_ANONYMOUS_STUB_H

@ -48,6 +48,8 @@
struct grpc_server;
namespace grpc {
class AnonymousServerContext;
class AnonymousService;
class AsynchronousService;
class RpcService;
class RpcServiceMethod;
@ -69,6 +71,7 @@ class Server GRPC_FINAL : private CallHook,
void Wait();
private:
friend class AnonymousService;
friend class ServerBuilder;
class SyncRequest;
@ -82,6 +85,7 @@ class Server GRPC_FINAL : private CallHook,
// 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 @@ class Server GRPC_FINAL : private CallHook,
ServerAsyncStreamingInterface* stream,
CompletionQueue* cq, void* tag);
void RequestAsyncAnonymousCall(AnonymousServerContext* context,
ServerAsyncStreamingInterface* stream,
CompletionQueue* cq, void* tag);
// Completion queue.
CompletionQueue cq_;

@ -87,6 +87,7 @@ class ServerBuilder {
std::vector<AsynchronousService*> async_services_;
std::vector<grpc::string> ports_;
std::shared_ptr<ServerCredentials> creds_;
AnonymousService* anonymous_service_;
ThreadPoolInterface* thread_pool_;
};

@ -66,7 +66,7 @@ class CompletionQueue;
class Server;
// Interface of server side rpc context.
class ServerContext GRPC_FINAL {
class ServerContext {
public:
ServerContext(); // for async calls
~ServerContext();

@ -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

@ -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 @@ bool Server::RegisterAsyncService(AsynchronousService* service) {
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 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
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 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
*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 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
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 @@ void Server::RequestAsyncCall(void* registered_method, ServerContext* context,
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() {
{

@ -51,6 +51,16 @@ void ServerBuilder::RegisterAsyncService(AsynchronousService* service) {
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 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
return nullptr;
}
}
if (anonymous_service_) {
server->RegisterAnonymousService(anonymous_service_);
}
for (auto& port : ports_) {
if (!server->AddPort(port)) {
return nullptr;

@ -0,0 +1,2 @@
#include <grpc++/byte_buffer.h>
Loading…
Cancel
Save