mirror of https://github.com/grpc/grpc.git
Vijay Pai
6 years ago
parent
d766b9a2a4
commit
8bf52535d1
27 changed files with 772 additions and 7 deletions
@ -0,0 +1,79 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
// Forward declarations
|
||||
namespace grpc_core { |
||||
class CQCallbackInterface; |
||||
}; |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
class CallbackWithStatusTag { |
||||
public: |
||||
// TODO(vjpai): make impl and ops part of this structure to avoid allocation,
|
||||
// ownership transfer, and delete
|
||||
CallbackWithStatusTag(std::function<void(Status)> f, bool self_delete, |
||||
CompletionQueueTag* ops); |
||||
~CallbackWithStatusTag() { delete ops_; } |
||||
void* tag() { return static_cast<void*>(impl_); } |
||||
Status* status_ptr() { return status_; } |
||||
CompletionQueueTag* ops() { return ops_; } |
||||
|
||||
// force_run can only be performed on a tag before it can ever be active
|
||||
void force_run(Status s); |
||||
|
||||
private: |
||||
grpc_core::CQCallbackInterface* impl_; |
||||
Status* status_; |
||||
CompletionQueueTag* ops_; |
||||
}; |
||||
|
||||
class CallbackWithSuccessTag { |
||||
public: |
||||
// TODO(vjpai): make impl and ops part of this structure to avoid allocation,
|
||||
// ownership transfer, and delete
|
||||
CallbackWithSuccessTag(std::function<void(bool)> f, bool self_delete, |
||||
CompletionQueueTag* ops); |
||||
~CallbackWithSuccessTag() { delete ops_; } |
||||
void* tag() { return static_cast<void*>(impl_); } |
||||
CompletionQueueTag* ops() { return ops_; } |
||||
|
||||
// force_run can only be performed on a tag before it can ever be active
|
||||
void force_run(bool ok); |
||||
|
||||
private: |
||||
grpc_core::CQCallbackInterface* impl_; |
||||
CompletionQueueTag* ops_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,92 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
#define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc { |
||||
|
||||
class Channel; |
||||
class ClientContext; |
||||
class CompletionQueue; |
||||
|
||||
namespace internal { |
||||
class RpcMethod; |
||||
|
||||
/// Perform a callback-based unary call
|
||||
/// TODO(vjpai): Combine as much as possible with the blocking unary call code
|
||||
template <class InputMessage, class OutputMessage> |
||||
void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CallbackUnaryCallImpl<InputMessage, OutputMessage> x( |
||||
channel, method, context, request, result, on_completion); |
||||
} |
||||
|
||||
template <class InputMessage, class OutputMessage> |
||||
class CallbackUnaryCallImpl { |
||||
public: |
||||
CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CompletionQueue* cq = channel->CallbackCQ(); |
||||
GPR_CODEGEN_ASSERT(cq != nullptr); |
||||
|
||||
// TODO(vjpai): Allocate this as part of the tag's arena
|
||||
auto* ops = new CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, |
||||
CallOpRecvInitialMetadata, |
||||
CallOpRecvMessage<OutputMessage>, |
||||
CallOpClientSendClose, CallOpClientRecvStatus>; |
||||
|
||||
// TODO(vjpai): Move to using pre-allocated tags rather than new/self-delete
|
||||
auto* tag = new CallbackWithStatusTag(on_completion, true, ops); |
||||
|
||||
// TODO(vjpai): Unify code with sync API as much as possible
|
||||
Call call(channel->CreateCall(method, context, cq)); |
||||
Status s = ops->SendMessage(*request); |
||||
if (!s.ok()) { |
||||
tag->force_run(s); |
||||
return; |
||||
} |
||||
ops->SendInitialMetadata(context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
ops->RecvInitialMetadata(context); |
||||
ops->RecvMessage(result); |
||||
ops->AllowNoMessage(); |
||||
ops->ClientSendClose(); |
||||
ops->ClientRecvStatus(context, tag->status_ptr()); |
||||
ops->set_cq_tag(tag->tag()); |
||||
call.PerformOps(ops); |
||||
} |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,24 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_SUPPORT_CLIENT_CALLBACK_H |
||||
#define GRPCPP_SUPPORT_CLIENT_CALLBACK_H |
||||
|
||||
#include <grpcpp/impl/codegen/client_callback.h> |
||||
|
||||
#endif // GRPCPP_SUPPORT_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,109 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
#include "src/core/lib/gprpp/memory.h" |
||||
#include "src/core/lib/surface/completion_queue.h" |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface { |
||||
public: |
||||
CallbackWithSuccessImpl(CallbackWithSuccessTag* parent, |
||||
std::function<void(bool)> f, bool self_delete) |
||||
: parent_(parent), func_(f), self_delete_(self_delete) {} |
||||
|
||||
void Run(bool ok) override { |
||||
void* ignored = parent_->ops(); |
||||
bool new_ok = ok; |
||||
GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok)); |
||||
GPR_ASSERT(ignored == parent_->ops()); |
||||
func_(ok); |
||||
if (self_delete_) { |
||||
delete parent_; |
||||
// Must use grpc_core::Delete since base is GRPC_ABSTRACT
|
||||
grpc_core::Delete(this); |
||||
} |
||||
} |
||||
|
||||
private: |
||||
CallbackWithSuccessTag* parent_; |
||||
std::function<void(bool)> func_; |
||||
bool self_delete_; |
||||
}; |
||||
|
||||
class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface { |
||||
public: |
||||
CallbackWithStatusImpl(CallbackWithStatusTag* parent, |
||||
std::function<void(Status)> f, bool self_delete) |
||||
: parent_(parent), func_(f), status_(), self_delete_(self_delete) {} |
||||
|
||||
void Run(bool ok) override { |
||||
void* ignored = parent_->ops(); |
||||
|
||||
GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok)); |
||||
GPR_ASSERT(ignored == parent_->ops()); |
||||
|
||||
func_(status_); |
||||
if (self_delete_) { |
||||
delete parent_; |
||||
// Must use grpc_core::Delete since base is GRPC_ABSTRACT
|
||||
grpc_core::Delete(this); |
||||
} |
||||
} |
||||
Status* status_ptr() { return &status_; } |
||||
|
||||
private: |
||||
CallbackWithStatusTag* parent_; |
||||
std::function<void(Status)> func_; |
||||
Status status_; |
||||
bool self_delete_; |
||||
}; |
||||
|
||||
} // namespace
|
||||
|
||||
CallbackWithSuccessTag::CallbackWithSuccessTag(std::function<void(bool)> f, |
||||
bool self_delete, |
||||
CompletionQueueTag* ops) |
||||
: impl_(grpc_core::New<CallbackWithSuccessImpl>(this, f, self_delete)), |
||||
ops_(ops) {} |
||||
|
||||
void CallbackWithSuccessTag::force_run(bool ok) { impl_->Run(ok); } |
||||
|
||||
CallbackWithStatusTag::CallbackWithStatusTag(std::function<void(Status)> f, |
||||
bool self_delete, |
||||
CompletionQueueTag* ops) |
||||
: ops_(ops) { |
||||
auto* impl = grpc_core::New<CallbackWithStatusImpl>(this, f, self_delete); |
||||
impl_ = impl; |
||||
status_ = impl->status_ptr(); |
||||
} |
||||
|
||||
void CallbackWithStatusTag::force_run(Status s) { |
||||
*status_ = s; |
||||
impl_->Run(true); |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,126 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <functional> |
||||
#include <mutex> |
||||
|
||||
#include <grpcpp/channel.h> |
||||
#include <grpcpp/client_context.h> |
||||
#include <grpcpp/create_channel.h> |
||||
#include <grpcpp/generic/generic_stub.h> |
||||
#include <grpcpp/impl/codegen/proto_utils.h> |
||||
#include <grpcpp/server.h> |
||||
#include <grpcpp/server_builder.h> |
||||
#include <grpcpp/server_context.h> |
||||
#include <grpcpp/support/client_callback.h> |
||||
|
||||
#include "src/proto/grpc/testing/echo.grpc.pb.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
#include "test/cpp/util/byte_buffer_proto_helper.h" |
||||
|
||||
#include <gtest/gtest.h> |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
|
||||
class ClientCallbackEnd2endTest : public ::testing::Test { |
||||
protected: |
||||
ClientCallbackEnd2endTest() {} |
||||
|
||||
void SetUp() override { |
||||
ServerBuilder builder; |
||||
builder.RegisterService(&service_); |
||||
|
||||
server_ = builder.BuildAndStart(); |
||||
is_server_started_ = true; |
||||
} |
||||
|
||||
void ResetStub() { |
||||
ChannelArguments args; |
||||
channel_ = server_->InProcessChannel(args); |
||||
stub_.reset(new GenericStub(channel_)); |
||||
} |
||||
|
||||
void TearDown() override { |
||||
if (is_server_started_) { |
||||
server_->Shutdown(); |
||||
} |
||||
} |
||||
|
||||
void SendRpcs(int num_rpcs) { |
||||
const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo"); |
||||
grpc::string test_string(""); |
||||
for (int i = 0; i < num_rpcs; i++) { |
||||
EchoRequest request; |
||||
std::unique_ptr<ByteBuffer> send_buf; |
||||
ByteBuffer recv_buf; |
||||
ClientContext cli_ctx; |
||||
|
||||
test_string += "Hello world. "; |
||||
request.set_message(test_string); |
||||
send_buf = SerializeToByteBuffer(&request); |
||||
|
||||
std::mutex mu; |
||||
std::condition_variable cv; |
||||
bool done; |
||||
stub_->experimental().UnaryCall( |
||||
&cli_ctx, kMethodName, send_buf.get(), &recv_buf, |
||||
[&request, &recv_buf, &done, &mu, &cv](Status s) { |
||||
GPR_ASSERT(s.ok()); |
||||
|
||||
EchoResponse response; |
||||
EXPECT_TRUE(ParseFromByteBuffer(&recv_buf, &response)); |
||||
EXPECT_EQ(request.message(), response.message()); |
||||
std::lock_guard<std::mutex> l(mu); |
||||
done = true; |
||||
cv.notify_one(); |
||||
}); |
||||
std::unique_lock<std::mutex> l(mu); |
||||
while (!done) { |
||||
cv.wait(l); |
||||
} |
||||
} |
||||
} |
||||
bool is_server_started_; |
||||
std::shared_ptr<Channel> channel_; |
||||
std::unique_ptr<grpc::GenericStub> stub_; |
||||
TestServiceImpl service_; |
||||
std::unique_ptr<Server> server_; |
||||
}; |
||||
|
||||
TEST_F(ClientCallbackEnd2endTest, SimpleRpc) { |
||||
ResetStub(); |
||||
SendRpcs(1); |
||||
} |
||||
|
||||
TEST_F(ClientCallbackEnd2endTest, SequentialRpcs) { |
||||
ResetStub(); |
||||
SendRpcs(10); |
||||
} |
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc_test_init(argc, argv); |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue