Support registering services against specific hosts

pull/2320/head
Craig Tiller 10 years ago
parent 772187cdf0
commit 822d2c7beb
  1. 4
      include/grpc++/server.h
  2. 26
      include/grpc++/server_builder.h
  3. 2
      src/cpp/client/channel.cc
  4. 9
      src/cpp/server/server.cc
  5. 18
      src/cpp/server/server_builder.cc
  6. 27
      test/cpp/end2end/end2end_test.cc
  7. 1
      test/cpp/util/messages.proto

@ -84,8 +84,8 @@ class Server GRPC_FINAL : public GrpcLibrary, private CallHook {
int max_message_size);
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance.
bool RegisterService(RpcService* service);
bool RegisterAsyncService(AsynchronousService* service);
bool RegisterService(const grpc::string *host, RpcService* service);
bool RegisterAsyncService(const grpc::string *host, AsynchronousService* service);
void RegisterAsyncGenericService(AsyncGenericService* service);
// Add a listening port. Can be called multiple times.
int AddListeningPort(const grpc::string& addr, ServerCredentials* creds);

@ -69,6 +69,19 @@ class ServerBuilder {
// Register a generic service.
void RegisterAsyncGenericService(AsyncGenericService* service);
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance returned by
// BuildAndStart().
void RegisterService(const grpc::string& host,
SynchronousService* service);
// Register an asynchronous service. New calls will be delevered to cq.
// This call does not take ownership of the service or completion queue.
// The service and completion queuemust exist for the lifetime of the Server
// instance returned by BuildAndStart().
void RegisterAsyncService(const grpc::string& host,
AsynchronousService* service);
// Set max message size in bytes.
void SetMaxMessageSize(int max_message_size) {
max_message_size_ = max_message_size;
@ -98,9 +111,18 @@ class ServerBuilder {
int* selected_port;
};
typedef std::unique_ptr<grpc::string> HostString;
template <class T> struct NamedService {
explicit NamedService(T* s) : service(s) {}
explicit NamedService(const grpc::string& h, T *s)
: host(new grpc::string(h)), service(s) {}
HostString host;
T* service;
};
int max_message_size_;
std::vector<RpcService*> services_;
std::vector<AsynchronousService*> async_services_;
std::vector<NamedService<RpcService>> services_;
std::vector<NamedService<AsynchronousService>> async_services_;
std::vector<Port> ports_;
std::vector<ServerCompletionQueue*> cqs_;
std::shared_ptr<ServerCredentials> creds_;

@ -59,7 +59,7 @@ Channel::~Channel() { grpc_channel_destroy(c_channel_); }
Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) {
auto c_call =
method.channel_tag()
method.channel_tag() && context->authority().empty()
? grpc_channel_create_registered_call(c_channel_, cq->cq(),
method.channel_tag(),
context->raw_deadline())

@ -207,10 +207,11 @@ Server::~Server() {
delete sync_methods_;
}
bool Server::RegisterService(RpcService* service) {
bool Server::RegisterService(const grpc::string *host, RpcService* service) {
for (int i = 0; i < service->GetMethodCount(); ++i) {
RpcServiceMethod* method = service->GetMethod(i);
void* tag = grpc_server_register_method(server_, method->name(), nullptr);
void* tag = grpc_server_register_method(
server_, method->name(), host ? host->c_str() : nullptr);
if (!tag) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
method->name());
@ -222,14 +223,14 @@ bool Server::RegisterService(RpcService* service) {
return true;
}
bool Server::RegisterAsyncService(AsynchronousService* service) {
bool Server::RegisterAsyncService(const grpc::string *host, AsynchronousService* service) {
GPR_ASSERT(service->server_ == nullptr &&
"Can only register an asynchronous service against one server.");
service->server_ = this;
service->request_args_ = new void*[service->method_count_];
for (size_t i = 0; i < service->method_count_; ++i) {
void* tag = grpc_server_register_method(server_, service->method_names_[i],
nullptr);
host ? host->c_str() : nullptr);
if (!tag) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
service->method_names_[i]);

@ -51,11 +51,21 @@ std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue() {
}
void ServerBuilder::RegisterService(SynchronousService* service) {
services_.push_back(service->service());
services_.emplace_back(service->service());
}
void ServerBuilder::RegisterAsyncService(AsynchronousService* service) {
async_services_.push_back(service);
async_services_.emplace_back(service);
}
void ServerBuilder::RegisterService(
const grpc::string& addr, SynchronousService* service) {
services_.emplace_back(addr, service->service());
}
void ServerBuilder::RegisterAsyncService(
const grpc::string& addr, AsynchronousService* service) {
async_services_.emplace_back(addr, service);
}
void ServerBuilder::RegisterAsyncGenericService(AsyncGenericService* service) {
@ -97,13 +107,13 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
}
for (auto service = services_.begin(); service != services_.end();
service++) {
if (!server->RegisterService(*service)) {
if (!server->RegisterService(service->host.get(), service->service)) {
return nullptr;
}
}
for (auto service = async_services_.begin();
service != async_services_.end(); service++) {
if (!server->RegisterAsyncService(*service)) {
if (!server->RegisterAsyncService(service->host.get(), service->service)) {
return nullptr;
}
}

@ -87,12 +87,16 @@ void MaybeEchoDeadline(ServerContext* context, const EchoRequest* request,
class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service {
public:
TestServiceImpl() : signal_client_(false) {}
TestServiceImpl() : signal_client_(false), host_(nullptr) {}
explicit TestServiceImpl(const grpc::string& host) : signal_client_(false), host_(new grpc::string(host)) {}
Status Echo(ServerContext* context, const EchoRequest* request,
EchoResponse* response) GRPC_OVERRIDE {
response->set_message(request->message());
MaybeEchoDeadline(context, request, response);
if (host_) {
response->mutable_param()->set_host(*host_);
}
if (request->has_param() && request->param().client_cancel_after_us()) {
{
std::unique_lock<std::mutex> lock(mu_);
@ -191,6 +195,7 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service {
private:
bool signal_client_;
std::mutex mu_;
std::unique_ptr<grpc::string> host_;
};
class TestServiceImplDupPkg
@ -205,7 +210,7 @@ class TestServiceImplDupPkg
class End2endTest : public ::testing::Test {
protected:
End2endTest() : kMaxMessageSize_(8192), thread_pool_(2) {}
End2endTest() : kMaxMessageSize_(8192), special_service_("special"), thread_pool_(2) {}
void SetUp() GRPC_OVERRIDE {
int port = grpc_pick_unused_port_or_die();
@ -215,6 +220,7 @@ class End2endTest : public ::testing::Test {
builder.AddListeningPort(server_address_.str(),
FakeTransportSecurityServerCredentials());
builder.RegisterService(&service_);
builder.RegisterService("special", &special_service_);
builder.SetMaxMessageSize(
kMaxMessageSize_); // For testing max message size.
builder.RegisterService(&dup_pkg_service_);
@ -236,6 +242,7 @@ class End2endTest : public ::testing::Test {
std::ostringstream server_address_;
const int kMaxMessageSize_;
TestServiceImpl service_;
TestServiceImpl special_service_;
TestServiceImplDupPkg dup_pkg_service_;
ThreadPool thread_pool_;
};
@ -254,6 +261,22 @@ static void SendRpc(grpc::cpp::test::util::TestService::Stub* stub,
}
}
TEST_F(End2endTest, SimpleRpcWithHost) {
ResetStub();
EchoRequest request;
EchoResponse response;
request.set_message("Hello");
ClientContext context;
context.set_authority("special");
Status s = stub_->Echo(&context, request, &response);
EXPECT_EQ(response.message(), request.message());
EXPECT_TRUE(response.has_param());
EXPECT_EQ(response.param().host(), "special");
EXPECT_TRUE(s.ok());
}
TEST_F(End2endTest, SimpleRpc) {
ResetStub();
SendRpc(stub_.get(), 1);

@ -46,6 +46,7 @@ message EchoRequest {
message ResponseParams {
optional int64 request_deadline = 1;
optional string host = 2;
}
message EchoResponse {

Loading…
Cancel
Save