qiurui
/
qtgrpc
mirror of https://github.com/qt/qtgrpc.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement QGrpcChannelOptions::deadline() 

Use QGrpcChannelOptions::deadline() as a default value of a deadline
for each call and stream executed on channel. If QGrpcCallOptions
deadline is specified, use it instead.

Task-number: QTBUG-113571
Pick-to: 6.6
Change-Id: I678d97e1c06f24bb517bbd43550b1c6e13188218
Reviewed-by: Alexey Edelev <alexey.edelev@qt.io>
This commit is contained in:
Konrad Kujawa 2023-08-03 14:57:46 +02:00
parent bcdb701d50
commit b33bfe4397
8 changed files with 250 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/grpc/qgrpccalloptions.cpp
View File

@ -91,6 +91,8 @@ QGrpcCallOptions &QGrpcCallOptions::withMetadata(const QGrpcMetadata &metadata)
Returns deadline value for a call.
Deadline value controls the maximum execution time of an call or a stream.
This value overrides value set by QGrpcChannelOptions::deadline()
for a specific call or stream.
If value was not set returns empty std::optional.
*/

35
src/grpc/qgrpcchannel.cpp
View File

@ -79,6 +79,16 @@ static QByteArray buildRpcName(QLatin1StringView service, QLatin1StringView meth
return '/' % QByteArrayView(service) % '/' % QByteArrayView(method);
}
static std::optional<std::chrono::milliseconds> deadlineForCall(
const QGrpcChannelOptions &channelOptions, const QGrpcCallOptions &callOptions)
{
if (callOptions.deadline())
return *callOptions.deadline();
if (channelOptions.deadline())
return *channelOptions.deadline();
return std::nullopt;
}
QGrpcChannelStream::QGrpcChannelStream(grpc::Channel *channel, QLatin1StringView method,
QByteArrayView data)
{
@ -182,21 +192,24 @@ void QGrpcChannelCall::waitForFinished(const QDeadlineTimer &deadline)
thread->wait(deadline);
}
QGrpcChannelPrivate::QGrpcChannelPrivate(const QGrpcChannelOptions &options,
QGrpcChannelPrivate::QGrpcChannelPrivate(const QGrpcChannelOptions &channelOptions,
QGrpcChannel::NativeGrpcChannelCredentials credentialsType)
: m_channelOptions(channelOptions)
{
switch (credentialsType) {
case QGrpcChannel::InsecureChannelCredentials:
m_credentials = grpc::InsecureChannelCredentials();
m_channel = grpc::CreateChannel(options.host().toString().toStdString(), m_credentials);
m_channel = grpc::CreateChannel(m_channelOptions.host().toString().toStdString(),
m_credentials);
break;
case QGrpcChannel::GoogleDefaultCredentials:
m_credentials = grpc::GoogleDefaultCredentials();
m_channel = grpc::CreateChannel(options.host().toString().toStdString(), m_credentials);
m_channel = grpc::CreateChannel(m_channelOptions.host().toString().toStdString(),
m_credentials);
break;
case QGrpcChannel::SslDefaultCredentials:
#if QT_CONFIG(ssl)
if (auto maybeSslConfig = options.sslConfiguration()) {
if (auto maybeSslConfig = m_channelOptions.sslConfiguration()) {
grpc::SslCredentialsOptions options;
auto accumulateSslCert = [](const std::string &lhs, const QSslCertificate &rhs) {
return lhs + rhs.toPem().toStdString();
@ -215,7 +228,8 @@ QGrpcChannelPrivate::QGrpcChannelPrivate(const QGrpcChannelOptions &options,
#else
m_credentials = grpc::SslCredentials(grpc::SslCredentialsOptions());
#endif
m_channel = grpc::CreateChannel(options.host().toString().toStdString(), m_credentials);
m_channel = grpc::CreateChannel(m_channelOptions.host().toString().toStdString(),
m_credentials);
break;
}
}
@ -258,8 +272,8 @@ std::shared_ptr<QGrpcCallReply> QGrpcChannelPrivate::call(QLatin1StringView meth
});
call->start();
if (options.deadline())
QTimer::singleShot(*options.deadline(), call.get(), [call] { call->cancel(); });
if (auto deadline = deadlineForCall(m_channelOptions, options))
QTimer::singleShot(*deadline, call.get(), [call] { call->cancel(); });
return reply;
}
@ -271,7 +285,8 @@ QGrpcStatus QGrpcChannelPrivate::call(QLatin1StringView method, QLatin1StringVie
QGrpcChannelCall call(m_channel.get(), QLatin1StringView(rpcName), args);
call.start();
options.deadline() ? call.waitForFinished(*options.deadline()) : call.waitForFinished();
auto deadline = deadlineForCall(m_channelOptions, options);
deadline ? call.waitForFinished(*deadline) : call.waitForFinished();
ret = call.response;
return call.status;
@ -323,8 +338,8 @@ std::shared_ptr<QGrpcStream> QGrpcChannelPrivate::startStream(QLatin1StringView
});
sub->start();
if (options.deadline())
QTimer::singleShot(*options.deadline(), sub.get(), [sub] { sub->cancel(); });
if (auto deadline = deadlineForCall(m_channelOptions, options))
QTimer::singleShot(*deadline, sub.get(), [sub] { sub->cancel(); });
return stream;
}

3
src/grpc/qgrpcchannel_p.h
View File

@ -89,8 +89,9 @@ struct QGrpcChannelPrivate
{
std::shared_ptr<grpc::Channel> m_channel;
std::shared_ptr<grpc::ChannelCredentials> m_credentials;
QGrpcChannelOptions m_channelOptions;
explicit QGrpcChannelPrivate(const QGrpcChannelOptions &options,
explicit QGrpcChannelPrivate(const QGrpcChannelOptions &channelOptions,
QGrpcChannel::NativeGrpcChannelCredentials credentialsType);
~QGrpcChannelPrivate();

3
src/grpc/qgrpcchanneloptions.cpp
View File

@ -96,6 +96,9 @@ QGrpcChannelOptions &QGrpcChannelOptions::withMetadata(const QGrpcMetadata &meta
/*!
Returns deadline value for setting up the channel.
Deadline value controls the maximum execution time of any call or stream
executed on the channel.
If value was not set returns empty std::optional.
*/
std::optional<std::chrono::milliseconds> QGrpcChannelOptions::deadline() const

14
src/grpc/qgrpchttp2channel.cpp
View File

@ -104,6 +104,16 @@ static QGrpcMetadata collectMetadata(QNetworkReply *networkReply)
networkReply->rawHeaderPairs().end());
}
static std::optional<std::chrono::milliseconds> deadlineForCall(
const QGrpcChannelOptions &channelOptions, const QGrpcCallOptions &callOptions)
{
if (callOptions.deadline())
return *callOptions.deadline();
if (channelOptions.deadline())
return *channelOptions.deadline();
return std::nullopt;
}
struct QGrpcHttp2ChannelPrivate
{
struct ExpectedData
@ -157,8 +167,8 @@ struct QGrpcHttp2ChannelPrivate
QGrpcHttp2ChannelPrivate::abortNetworkReply(networkReply);
});
#endif
if (callOptions.deadline()) {
QTimer::singleShot(*callOptions.deadline(), networkReply, [networkReply] {
if (auto deadline = deadlineForCall(channelOptions, callOptions)) {
QTimer::singleShot(*deadline, networkReply, [networkReply] {
QGrpcHttp2ChannelPrivate::abortNetworkReply(networkReply);
});
}

1
tests/auto/grpc/CMakeLists.txt
View File

@ -11,6 +11,7 @@ add_compile_definitions(QT_GRPC_TEST_MESSAGE_LATENCY=${QT_INTERNAL_GRPC_TEST_MES
if(TARGET WrapgRPC::WrapLibgRPC)
add_subdirectory(shared)
add_subdirectory(client)
add_subdirectory(client_no_channel)
add_subdirectory(server)
if(QT_FEATURE_ssl)
add_subdirectory(ssl_client)

43
tests/auto/grpc/client_no_channel/CMakeLists.txt Normal file
View File

@ -0,0 +1,43 @@
# Copyright (C) 2023 The Qt Company Ltd.
# SPDX-License-Identifier: BSD-3-Clause
if(NOT QT_BUILD_STANDALONE_TESTS AND NOT QT_BUILDING_QT)
cmake_minimum_required(VERSION 3.16)
project(tst_grpc_client_no_channel LANGUAGES CXX)
find_package(Qt6BuildInternals REQUIRED COMPONENTS STANDALONE_TEST)
include("../shared/test_server/CMakeLists.txt")
endif()
if(NOT TARGET testserver)
return()
endif()
qt6_add_protobuf(tst_grpc_client_no_channel_qtgrpc_gen
OUTPUT_DIRECTORY
"${CMAKE_CURRENT_BINARY_DIR}/qt_grpc_generated"
PROTO_FILES
${CMAKE_CURRENT_SOURCE_DIR}/../shared/proto/testservice.proto
)
qt6_add_grpc(tst_grpc_client_no_channel_qtgrpc_gen CLIENT
PROTO_FILES
${CMAKE_CURRENT_SOURCE_DIR}/../shared/proto/testservice.proto
OUTPUT_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/qt_grpc_generated"
)
qt_autogen_tools_initial_setup(tst_grpc_client_no_channel_qtgrpc_gen)
qt_internal_add_test(tst_grpc_client_no_channel
SOURCES
tst_grpc_client_no_channel.cpp
INCLUDE_DIRECTORIES
${CMAKE_CURRENT_SOURCE_DIR}/../shared
DEFINES
TEST_GRPC_SERVER_PATH="$<TARGET_FILE:testserver>"
LIBRARIES
Qt::Test
Qt::Core
Qt::Grpc
tst_grpc_client_no_channel_qtgrpc_gen
)
add_dependencies(tst_grpc_client_no_channel testserver)

162
tests/auto/grpc/client_no_channel/tst_grpc_client_no_channel.cpp Normal file
View File

@ -0,0 +1,162 @@
// Copyright (C) 2023 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include <QGrpcCallReply>
#if QT_CONFIG(native_grpc)
# include <QGrpcChannel>
#endif
#include <QCoreApplication>
#include <QCryptographicHash>
#include <QGrpcHttp2Channel>
#include <QSignalSpy>
#include <QTest>
#include <QtGrpc/QGrpcCallOptions>
#include <QtGrpc/QGrpcChannelOptions>
#include <memory>
#if QT_CONFIG(native_grpc)
# include <grpcpp/security/credentials.h>
#endif
#include "testservice_client.grpc.qpb.h"
#include <message_latency_defs.h>
#include <server_proc_runner.h>
using namespace Qt::Literals::StringLiterals;
using namespace qtgrpc::tests;
namespace {
class ChannelFactory
{
public:
enum ChannelType {
HTTP2,
GRPC_SOCKET,
GRPC_HTTP,
};
ChannelFactory(ChannelType type) : type(type) { }
std::shared_ptr<QAbstractGrpcChannel> create(const QGrpcChannelOptions &options) const
{
switch (type) {
case HTTP2:
return std::make_shared<QGrpcHttp2Channel>(options);
#if QT_CONFIG(native_grpc)
case GRPC_SOCKET:
case GRPC_HTTP:
return std::make_shared<QGrpcChannel>(options,
QGrpcChannel::InsecureChannelCredentials);
#endif
}
return nullptr;
}
QUrl getUrl() const
{
switch (type) {
case HTTP2:
return QUrl("http://localhost:50051", QUrl::StrictMode);
#if QT_CONFIG(native_grpc)
case GRPC_SOCKET:
return QUrl("unix:///tmp/test.sock");
case GRPC_HTTP:
return QUrl("localhost:50051");
#endif
}
return QUrl();
}
private:
ChannelType type;
};
} // namespace
class QtGrpcClientNoChannelTest : public QObject
{
Q_OBJECT
private slots:
void createClient(const QGrpcChannelOptions &channelOptions)
{
client = std::make_shared<TestService::Client>();
auto channel = channelFactory->create(channelOptions);
QVERIFY2(channel, "Channel could not been created.");
client->attachChannel(std::move(channel));
}
void initTestCase_data()
{
QTest::addColumn<std::shared_ptr<ChannelFactory>>("factory");
QTest::newRow("Http2Client") << std::make_shared<ChannelFactory>(ChannelFactory::HTTP2);
#if QT_CONFIG(native_grpc)
# ifndef Q_OS_WINDOWS
QTest::newRow("GrpcSocket")
<< std::make_shared<ChannelFactory>(ChannelFactory::GRPC_SOCKET);
# endif
QTest::newRow("GrpcHttp") << std::make_shared<ChannelFactory>(ChannelFactory::GRPC_HTTP);
#endif
}
void initTestCase() { qRegisterProtobufTypes(); }
void init()
{
if (serverProc.state() != QProcess::ProcessState::Running) {
qInfo() << "Restarting server";
serverProc.restart();
QVERIFY2(serverProc.state() == QProcess::ProcessState::Running,
"Precondition failed - Server cannot be started.");
}
QFETCH_GLOBAL(std::shared_ptr<ChannelFactory>, factory);
channelFactory = std::move(factory);
}
void ChannelAndCallDeadlineTest();
private:
ServerProcRunner serverProc{ QFINDTESTDATA(TEST_GRPC_SERVER_PATH) };
std::shared_ptr<ChannelFactory> channelFactory;
std::shared_ptr<TestService::Client> client;
};
void QtGrpcClientNoChannelTest::ChannelAndCallDeadlineTest()
{
constexpr auto channelTimeout = std::chrono::milliseconds(static_cast<int64_t>(MessageLatency
* 0.25));
constexpr auto callTimeout = std::chrono::milliseconds(static_cast<int64_t>(MessageLatency
* 0.6));
QGrpcChannelOptions channelOpts(channelFactory->getUrl());
channelOpts.withDeadline(channelTimeout);
createClient(channelOpts);
QGrpcCallOptions callOpts;
callOpts.withDeadline(callTimeout);
auto executeTest = [this](uint64_t minTimeout, uint64_t maxTimeout,
std::optional<QGrpcCallOptions> callOptions = std::nullopt) {
SimpleStringMessage request;
request.setTestFieldString("sleep");
QSignalSpy clientErrorSpy(client.get(), &TestService::Client::errorOccurred);
QVERIFY(clientErrorSpy.isValid());
std::shared_ptr<QGrpcCallReply> reply;
if (callOptions)
reply = client->testMethod(request, *callOptions);
else
reply = client->testMethod(request);
QSignalSpy callFinishedSpy(reply.get(), &QGrpcCallReply::finished);
QVERIFY(callFinishedSpy.isValid());
// Still waiting for a timeout
QTRY_COMPARE_EQ_WITH_TIMEOUT(clientErrorSpy.count(), 0, minTimeout);
// Time window to receive the timout
QTRY_COMPARE_EQ_WITH_TIMEOUT(clientErrorSpy.count(), 1, maxTimeout);
const auto code = qvariant_cast<QGrpcStatus>(clientErrorSpy.at(0).first()).code();
// Really low timeout can trigger before service becomes available
QVERIFY(code == QGrpcStatus::StatusCode::Cancelled
|| code == QGrpcStatus::StatusCode::Unavailable);
};
executeTest(0, channelTimeout.count() + MessageLatencyThreshold);
executeTest(channelTimeout.count(), callTimeout.count() + MessageLatencyThreshold, callOpts);
}
QTEST_MAIN(QtGrpcClientNoChannelTest)
#include "tst_grpc_client_no_channel.moc"