Revision control
Copy as Markdown
// Copyright (C) 2023 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include <grpcclienttestbase.h>
#include <QtCore/QTimer>
#include <QtTest/QTest>
#include <QtTest/QSignalSpy>
#include <testservice_client.grpc.qpb.h>
#include <message_latency_defs.h>
using namespace Qt::Literals::StringLiterals;
using namespace qtgrpc::tests;
class QtGrpcClientClientStreamTest : public GrpcClientTestBase
{
Q_OBJECT
public:
QtGrpcClientClientStreamTest()
: GrpcClientTestBase(
Channels{ GrpcClientTestBase::Channel::Qt })
{
}
private Q_SLOTS:
void valid();
void sequentialSend();
void sequentialSendWithDone();
void sequentialSendWithDoneWhenChannelNotReady();
};
void QtGrpcClientClientStreamTest::valid()
{
const int ExpectedMessageCount = 4;
SimpleStringMessage request;
request.setTestFieldString("Stream1");
auto stream = client()->testMethodClientStream(request);
int i = 0;
QTimer sendTimer;
QObject::connect(&sendTimer, &QTimer::timeout, this, [&]() {
request.setTestFieldString("Stream" + QString::number(i + 2));
stream->writeMessage(request);
if (++i == ExpectedMessageCount)
sendTimer.stop();
});
sendTimer.start(MessageLatency);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.takeFirst();
QCOMPARE(args.count(), 1);
QVERIFY(args.takeFirst().value<QGrpcStatus>() == QtGrpc::StatusCode::Ok);
const auto result = stream->read<SimpleStringMessage>();
QCOMPARE_EQ(result->testFieldString(), "Stream11Stream22Stream33Stream44");
}
void QtGrpcClientClientStreamTest::sequentialSend()
{
const int ExpectedMessageCount = 4;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodClientStream(request);
// Ensure that messages are not lost during the sequential sending right after the stream is
// instanciated.
for (int i = 0; i < ExpectedMessageCount; ++i) {
stream->writeMessage(request);
}
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.takeFirst();
QCOMPARE(args.count(), 1);
QVERIFY(args.takeFirst().value<QGrpcStatus>() == QtGrpc::StatusCode::Ok);
const auto result = stream->read<SimpleStringMessage>();
QVERIFY(result.has_value());
QCOMPARE_EQ(result->testFieldString(), "Stream1Stream2Stream3Stream4");
}
void QtGrpcClientClientStreamTest::sequentialSendWithDone()
{
const int ExpectedMessageCount = 4;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodClientStreamWithDone(request);
// Ensure that messages are not lost during the sequential sending right after the stream is
// instanciated.
for (int i = 1; i < (ExpectedMessageCount - 1); ++i) {
stream->writeMessage(request);
}
stream->writesDone();
request.setTestFieldString("StreamWrong");
stream->writeMessage(request);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
std::optional<SimpleStringMessage> result = stream->read<SimpleStringMessage>();
QCOMPARE_EQ(result->testFieldString(), "Stream1Stream2Stream3");
}
void QtGrpcClientClientStreamTest::sequentialSendWithDoneWhenChannelNotReady()
{
auto channel = std::shared_ptr<
QUrl::StrictMode)));
auto client = std::make_shared<qtgrpc::tests::TestService::Client>();
client->attachChannel(std::move(channel));
const int ExpectedMessageCount = 4;
SimpleStringMessage request;
request.setTestFieldString("Stream1");
auto stream = client->testMethodClientStreamWithDone(request);
// Ensure that messages are not lost during the sequential sending right after the stream is
// instanciated.
for (int i = 1; i < (ExpectedMessageCount - 1); ++i) {
request.setTestFieldString(QString("Stream") + QString::number(i + 1));
stream->writeMessage(request);
}
stream->writesDone();
request.setTestFieldString("StreamWrong");
stream->writeMessage(request);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
std::optional<SimpleStringMessage> result = stream->read<SimpleStringMessage>();
QCOMPARE_EQ(result->testFieldString(), "Stream11Stream22Stream33");
}
QTEST_MAIN(QtGrpcClientClientStreamTest)
#include "tst_grpc_client_clientstream.moc"