Revision control
Copy as Markdown
// Copyright (C) 2022 The Qt Company Ltd.
// Copyright (C) 2019 Alexey Edelev <semlanik@gmail.com>
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include <QtGrpc/private/qgrpcoperation_p.h>
#include <QtGrpc/private/qgrpcoperationcontext_p.h>
#include <QtGrpc/private/qtgrpclogging_p.h>
#include <QtGrpc/qabstractgrpcchannel.h>
#include <QtGrpc/qgrpcoperation.h>
#include <QtGrpc/qgrpcoperationcontext.h>
#include <QtProtobuf/private/qprotobufmessage_p.h>
#include <QtCore/qbytearray.h>
#include <QtCore/qeventloop.h>
#include <QtCore/qpointer.h>
QT_BEGIN_NAMESPACE
using namespace Qt::StringLiterals;
QGrpcOperationPrivate::~QGrpcOperationPrivate()
= default;
void QGrpcOperationPrivate::asyncFinishInvalid(QGrpcStatus &&status)
{
Q_ASSERT(!status.isOk());
Q_Q(QGrpcOperation);
state = State::Invalid;
QMetaObject::invokeMethod(
q,
[qPtr = QPointer(q), s = std::move(status)] {
if (!qPtr)
return;
emit qPtr->context().finished(s);
},
Qt::QueuedConnection);
}
std::optional<QByteArray>
QGrpcOperationPrivate::serializeInitialMessage(const QProtobufMessage &message)
{
const auto serializer = operationContext->serializer();
auto ch = channel.lock();
Q_ASSERT(state == State::Valid);
Q_ASSERT(serializer);
Q_ASSERT(ch.get()); // This function should only be called if the channel is valid
auto &interceptorEngine = QAbstractGrpcChannelPrivate::get(ch.get())->interceptorEngine;
if (interceptorEngine.hasHandlerFor(QtGrpc::InterceptorCapability::Start)) {
auto mt = QProtobufMessagePrivate::get(&message)->metaObject->metaType();
auto *interceptedMessage = static_cast<QProtobufMessage *>(mt.create(&message));
Q_ASSERT(interceptedMessage);
const QScopeGuard mtGuard([mt = std::move(mt), interceptedMessage] {
mt.destroy(interceptedMessage);
});
const auto continuation = interceptorEngine.onStart(*operationContext,
*interceptedMessage,
operationContext->d_func()->options);
if (continuation == QGrpcStartInterceptor::Continuation::Drop) {
Q_Q(QGrpcOperation);
QMetaObject::invokeMethod(
q,
[qPtr = QPointer(q)] {
if (!qPtr)
return;
emit qPtr->context().finished({ QtGrpc::StatusCode::Aborted,
"Interceptors dropped the call"_L1 });
},
Qt::QueuedConnection);
return {};
}
return serializer->serialize(interceptedMessage);
}
return serializer->serialize(&message);
}
/*!
\class QGrpcOperation
\inmodule QtGrpc
\brief The QGrpcOperation class provides common operations to handle the
\gRPC communication from the client side.
QGrpcOperation serves as the base class for the four \gRPC method types:
QGrpcCallReply (unary calls), QGrpcServerStream (server streaming),
QGrpcClientStream (client streaming), and QGrpcBidiStream (bidirectional
streaming). It provides a common interface for interacting with these
remote procedure calls (RPCs).
Each QGrpcOperation corresponds to a specific RPC requested through the
generated client interface.
For a high-level overview, refer to the \l {clientguide} {Qt GRPC
Client Guide}.
*/
/*!
\fn void QGrpcOperation::finished(const QGrpcStatus &status)
This signal is emitted when a previously started RPC has finished. The \a
status provides additional information about the outcome of the RPC.
After this signal is received, no further write or read operations should
be performed on the operation object. At this point, it is safe to reuse or
destroy the RPC object.
\note This signal is emitted only once, and in most cases, you will want to
disconnect right after receiving it to avoid issues, such as lambda
captures not being destroyed after receiving the signal. An easy way to
achieve this is by using the \l {Qt::} {SingleShotConnection} connection
type. See \l {Single Shot RPCs} for further details.
*/
/*!
\fn void QGrpcOperation::serverInitialMetadataReceived(const QMultiHash<QByteArray, QByteArray> &metadata)
\since 6.11
//![serverInitialMetadataReceived]
This signal is emitted when the server responds with its initial metadata.
Initial metadata from the server is sent to the client before the initial
response of an RPC.
The received metadata is provided in \a metadata and is also available via
\l{serverInitialMetadata()}.
//![serverInitialMetadataReceived]
\include qtgrpc-shared.qdocinc metadata-note
\sa serverInitialMetadata()
*/
/*!
\internal
Constructs a QGrpcOperation from its subclass.
This is indirectly called by the generated client interface.
*/
QGrpcOperation::QGrpcOperation(QtGrpc::RpcDescriptor descriptor, const QGrpcCallOptions &options,
std::weak_ptr<QAbstractGrpcChannel> &&channel)
: QObject(*new QGrpcOperationPrivate())
{
Q_D(QGrpcOperation);
const auto lockedChannel = channel.lock();
d->operationContext = new QGrpcOperationContext(std::move(descriptor), options,
lockedChannel ? lockedChannel->serializer()
: nullptr,
this,
QGrpcOperationContext::PrivateConstructor{});
[[maybe_unused]] bool valid = false;
valid = connect(d->operationContext, &QGrpcOperationContext::messageReceived, this,
[this](const QByteArray &data) {
Q_D(QGrpcOperation);
d->data = data;
auto channel = d->channel.lock();
if (!channel) {
qGrpcCritical("message received without active channel");
return;
}
auto &engine = QAbstractGrpcChannelPrivate::get(channel.get())
->interceptorEngine;
if (engine.hasHandlerFor(QtGrpc::InterceptorCapability::MessageReceived))
engine.onMessageReceived(*d->operationContext, d->data);
});
Q_ASSERT_X(valid, "QGrpcOperation::QGrpcOperation",
"Unable to make connection to the 'messageReceived' signal");
valid = connect(d->operationContext, &QGrpcOperationContext::finished, this,
[this](const QGrpcStatus &status) {
Q_D(QGrpcOperation);
if (isFinished())
return;
d->state = QGrpcOperationPrivate::State::Finished;
if (auto ch = d->channel.lock()) {
auto &engine = QAbstractGrpcChannelPrivate::get(ch.get())
->interceptorEngine;
if (engine.hasHandlerFor(QtGrpc::InterceptorCapability::Finished)) {
auto interceptedStatus = status;
engine.onFinished(*d->operationContext, interceptedStatus);
emit finished(interceptedStatus);
return;
}
}
emit finished(status);
});
Q_ASSERT_X(valid, "QGrpcOperation::QGrpcOperation",
"Unable to make connection to the 'finished' signal");
valid = connect(d->operationContext, &QGrpcOperationContext::serverInitialMetadataReceived,
this, &QGrpcOperation::serverInitialMetadataReceived);
Q_ASSERT_X(valid, "QGrpcOperation::QGrpcOperation",
"Unable to make connection to the 'serverInitialMetadataReceived' signal");
if (!lockedChannel) {
d->asyncFinishInvalid({ QtGrpc::StatusCode::Aborted, tr("Channel is not available") });
return;
}
d->channel = std::move(channel);
if (!d->operationContext->serializer()) {
d->asyncFinishInvalid({ QtGrpc::StatusCode::Aborted, tr("Serializer is not available") });
return;
}
auto *chPriv = QAbstractGrpcChannelPrivate::get(lockedChannel.get());
Q_ASSERT(chPriv);
if (chPriv->threadId != QThread::currentThreadId()) {
d->asyncFinishInvalid({ QtGrpc::StatusCode::Aborted, tr("RPC started from wrong thread") });
return;
}
d->state = QGrpcOperationPrivate::State::Valid;
}
/*!
Destroys the QGrpcOperation.
*/
QGrpcOperation::~QGrpcOperation() = default;
/*!
\fn template <typename T, QtProtobuf::if_protobuf_message<T> = true> std::optional<T> QGrpcOperation::read() const
Reads a message from a raw byte array stored within this operation object.
Returns an optional deserialized message. On failure, \c {std::nullopt} is
returned.
\note This function only participates in overload resolution if \c T is a
subclass of QProtobufMessage.
\sa read(QProtobufMessage *)
*/
/*!
\since 6.8
Reads a message from a raw byte array stored within this operation object.
The function writes the deserialized value to the \a message pointer.
If the deserialization is successful, this function returns \c true.
Otherwise, it returns \c false.
\sa read()
*/
bool QGrpcOperation::read(QProtobufMessage *message) const
{
if (!message) {
qGrpcWarning("Read called on nullptr message");
return false;
}
Q_D(const QGrpcOperation);
const auto ser = d->operationContext->serializer();
if (!ser) {
qGrpcWarning("Serializer unavailable");
return false;
}
if (auto responseMetaType = d->operationContext->responseMetaType(); responseMetaType.isValid()
&& QProtobufMessagePrivate::get(message)->metaObject != responseMetaType.metaObject()) {
qGrpcWarning("Operation result meta type doesn't match the message meta type.");
}
if (!ser->deserialize(message, d->data)) {
qGrpcWarning() << "Unable to deserialize message(" << qToUnderlying(ser->lastError()) <<"): "
<< ser->lastErrorString();
return false;
}
return true;
}
/*!
Tries to cancel the RPC immediately. Successful cancellation cannot be
guaranteed. Emits the \l finished signal with a \l {QtGrpc::StatusCode::}
{Cancelled} status code.
\sa QGrpcOperationContext::cancelRequested
*/
void QGrpcOperation::cancel()
{
if (isFinished())
return;
Q_D(QGrpcOperation);
auto channel = d->channel.lock();
if (!channel) {
qGrpcCritical("cancel received without active channel");
return;
}
auto &engine = QAbstractGrpcChannelPrivate::get(channel.get())->interceptorEngine;
if (engine.hasHandlerFor(QtGrpc::InterceptorCapability::Cancel))
engine.onCancel(*d->operationContext);
emit d->operationContext->cancelRequested();
}
#if QT_DEPRECATED_SINCE(6, 13)
/*!
\deprecated [6.13] Use serverInitialMetadata() and serverTrailingMetadata() instead.
\include qgrpcoperation.cpp serverInitialMetadata-shared
\sa serverInitialMetadata() serverTrailingMetadata()
*/
const QHash<QByteArray, QByteArray> &QGrpcOperation::metadata() const & noexcept
{
Q_D(const QGrpcOperation);
QT_IGNORE_DEPRECATIONS(return d->operationContext->serverMetadata();)
}
#endif // QT_DEPRECATED_SINCE(6, 13)
/*!
\since 6.10
//! [serverInitialMetadata-shared]
Returns the initial metadata received from the server before any response
messages.
Initial metadata is sent by the server immediately after the call is
established. It may include key-value pairs that provide context for the
call.
\include qtgrpc-shared.qdocinc metadata-note
//! [serverInitialMetadata-shared]
The metadata may contain multiple entries under the same key.
\sa serverInitialMetadataReceived() serverTrailingMetadata()
*/
const QMultiHash<QByteArray, QByteArray> &QGrpcOperation::serverInitialMetadata() const & noexcept
{
Q_D(const QGrpcOperation);
return d->operationContext->serverInitialMetadata();
}
/*!
\since 6.10
//! [serverTrailingMetadata]
Returns the trailing metadata received from the server after all response
messages.
Trailing metadata is sent only by the server once all response messages
have been sent and just before the RPC completes. It may include key-value
pairs providing additional context about the completed call.
\include qtgrpc-shared.qdocinc metadata-note
//! [serverTrailingMetadata]
The metadata may contain multiple entries under the same key.
\sa serverInitialMetadata()
*/
const QMultiHash<QByteArray, QByteArray> &QGrpcOperation::serverTrailingMetadata() const & noexcept
{
Q_D(const QGrpcOperation);
return d->operationContext->serverTrailingMetadata();
}
/*!
Returns the method name associated with this RPC operation.
*/
QLatin1StringView QGrpcOperation::method() const noexcept
{
Q_D(const QGrpcOperation);
return d->operationContext->method();
}
/*!
Returns true if this operation has finished, meaning that no more
operations can happen on the corresponding RPC, otherwise returns false.
*/
bool QGrpcOperation::isFinished() const noexcept
{
Q_D(const QGrpcOperation);
return d->state == QGrpcOperationPrivate::State::Finished;
}
/*!
\internal
\fn const QGrpcOperationContext &QGrpcOperation::context() const &
\fn QGrpcOperationContext &QGrpcOperation::context() &
Returns a reference to the internal operation context.
*/
const QGrpcOperationContext &QGrpcOperation::context() const & noexcept
{
Q_D(const QGrpcOperation);
return *d->operationContext;
}
void QGrpcOperation::writeMessage(const QProtobufMessage &message)
{
Q_D(QGrpcOperation);
auto channel = d->channel.lock();
if (!channel) {
qGrpcCritical("writeMessage received without active channel");
return;
}
QByteArray messageData;
auto serializer = d->operationContext->serializer();
auto &engine = QAbstractGrpcChannelPrivate::get(channel.get())->interceptorEngine;
if (engine.hasHandlerFor(QtGrpc::InterceptorCapability::WriteMessage)) {
auto mt = QProtobufMessagePrivate::get(&message)->metaObject->metaType();
auto *interceptedMessage = static_cast<QProtobufMessage *>(mt.create(&message));
Q_ASSERT(interceptedMessage);
engine.onWriteMessage(*d->operationContext, *interceptedMessage);
messageData = serializer->serialize(interceptedMessage);
mt.destroy(interceptedMessage);
} else {
messageData = serializer->serialize(&message);
}
emit d->operationContext->writeMessageRequested(messageData);
}
void QGrpcOperation::writesDone()
{
Q_D(QGrpcOperation);
auto channel = d->channel.lock();
if (!channel) {
qGrpcCritical("writesDone received without active channel");
return;
}
auto &engine = QAbstractGrpcChannelPrivate::get(channel.get())->interceptorEngine;
if (engine.hasHandlerFor(QtGrpc::InterceptorCapability::WritesDone))
engine.onWritesDone(*d->operationContext);
emit d->operationContext->writesDoneRequested();
}
bool QGrpcOperation::event(QEvent *event)
{
return QObject::event(event);
}
/*!
Returns the meta type of the RPC response message.
*/
QMetaType QGrpcOperation::responseMetaType() const
{
Q_D(const QGrpcOperation);
return d->operationContext->responseMetaType();
}
QT_END_NAMESPACE
#include "moc_qgrpcoperation.cpp"