Revision control
++ // Copyright (C) 2025 The Qt Company Ltd.
++ // SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
++
++ #include "mockserver.h"
++ #include "tags.h"
++
++ #include <QtTest/qtest.h>
++
++ #include <grpcpp/alarm.h>
++ #include <grpcpp/server_builder.h>
++
++ #include <cstring>
++
++ using namespace std::chrono_literals;
++
++ TagProcessor::TagProcessor(MockServer *server)
++ : mServer(server)
++ {
++ QVERIFY(mServer);
++ QVERIFY(mServer->cq());
++ mThread = std::thread(&TagProcessor::processLoop, this);
++ }
++
++ TagProcessor::~TagProcessor()
++ {
++ QVERIFY(waitForTagCompletion(5s));
++
++ mRunning.store(false);
++ wakeCQ();
++
++ if (mThread.joinable())
++ mThread.join();
++
++ drainTags();
++ }
++
++ void TagProcessor::processLoop()
++ {
++ while (mRunning.load(std::memory_order_acquire)) {
++ mServer->processTag();
++ }
++ }
++
++ void TagProcessor::registerTag(AbstractTag *tag)
++ {
++ std::scoped_lock lock(mMutex);
++ const auto it = mActiveTags.insert(tag);
++ QVERIFY(it.second);
++ }
++
++ void TagProcessor::unregisterTag(AbstractTag *tag)
++ {
++ std::scoped_lock lock(mMutex);
++ const auto erased = mActiveTags.erase(tag);
++ QCOMPARE_EQ(erased, 1u);
++ if (mActiveTags.empty())
++ mCv.notify_all();
++ }
++
++ size_t TagProcessor::activeTagCount() const noexcept
++ {
++ std::scoped_lock lock(mMutex);
++ return mActiveTags.size();
++ }
++
++ bool TagProcessor::waitForTagCompletion(std::chrono::milliseconds deadline) noexcept
++ {
++ std::unique_lock lock(mMutex);
++ return mCv.wait_for(lock, deadline, [this] { return mActiveTags.empty(); });
++ }
++
++ void TagProcessor::wakeCQ()
++ {
++ auto *alarm = new grpc::Alarm();
++ auto *tag = new DeleteTag<grpc::Alarm>(alarm, this);
++ alarm->Set(mServer->cq(), gpr_now(GPR_CLOCK_REALTIME), tag);
++ }
++
++ void TagProcessor::drainTags()
++ {
++ QVERIFY(!mRunning);
++ while (mServer->processTag(50))
++ ;
++ }
++
++ MockServer::MockServer() = default;
++ MockServer::~MockServer()
++ {
++ stop();
++ };
++
++ bool MockServer::start(std::vector<ListeningPort> ports, std::vector<grpc::Service *> services,
++ const grpc::ChannelArguments &channelArgs)
++ {
++ if (!transitionState(State::Stopped, State::Starting))
++ return false;
++
++ grpc::ServerBuilder builder;
++ for (auto &p : ports)
++ builder.AddListeningPort(p.addressUri, p.creds, &p.selectedPort);
++ for (auto *s : services)
++ builder.RegisterService(s);
++ const grpc_channel_args cArgs = channelArgs.c_channel_args();
++ for (size_t i = 0; i < cArgs.num_args; ++i) {
++ const grpc_arg &a = cArgs.args[i];
++ // Skip client-only args that are not applicable to a server builder.
++ if (std::strcmp(a.key, GRPC_ARG_PRIMARY_USER_AGENT_STRING) == 0)
++ continue;
++ if (a.type == GRPC_ARG_INTEGER)
++ builder.AddChannelArgument(a.key, a.value.integer);
++ else if (a.type == GRPC_ARG_STRING)
++ builder.AddChannelArgument(a.key, a.value.string);
++ }
++ mCQ = builder.AddCompletionQueue();
++ mServer = builder.BuildAndStart();
++ if (!mServer || !mCQ) {
++ mState = State::Stopped;
++ return false;
++ }
++
++ mState = State::Started;
++ return true;
++ }
++
++ bool MockServer::stop()
++ {
++ State currentState = mState.load();
++ if (currentState != State::Started && currentState != State::ShuttingDown)
++ return currentState == State::Stopped;
++
++ mState = State::ShuttingDown;
++
++ mServer->Shutdown(std::chrono::system_clock::now() + std::chrono::milliseconds(500));
++ mCQ->Shutdown();
++
++ void *drainTag;
++ bool drainOk;
++ while (mCQ->Next(&drainTag, &drainOk)) {
++ if (drainTag)
++ static_cast<AbstractTag*>(drainTag)->process(drainOk);
++ }
++
++ mServer.reset();
++ mCQ.reset();
++ mState = State::Stopped;
++
++ return true;
++ }
++
++ bool MockServer::processTag(int timeoutMs)
++ {
++ State currentState = mState.load();
++ if (currentState != State::Started)
++ return false;
++
++ void *rawTag = nullptr;
++ bool ok = false;
++ const auto status = timeoutMs < 0
++ ? mCQ->AsyncNext(&rawTag, &ok, gpr_inf_future(GPR_CLOCK_REALTIME))
++ : mCQ->AsyncNext(&rawTag, &ok,
++ std::chrono::system_clock::now() + std::chrono::milliseconds(timeoutMs));
++
++ if (rawTag && status == grpc::CompletionQueue::NextStatus::GOT_EVENT) {
++ static_cast<AbstractTag *>(rawTag)->process(ok);
++ return true;
++ }
++
++ return false;
++ }
++
++ MockServer &MockServer::step(int timeoutMs)
++ {
++ mFutures.emplace_back(std::async(std::launch::async,
++ [this, timeoutMs] { return processTag(timeoutMs); }));
++ return *this;
++ }
++
++ bool MockServer::waitForAllSteps()
++ {
++ for (auto &f : mFutures) {
++ if (!f.get())
++ return false;
++ }
++ mFutures.clear();
++ return true;
++ }
++
++ void MockServer::startRpcTag(AbstractRpcTag *tag)
++ {
++ QVERIFY(tag);
++ tag->start(mCQ.get());
++ }
++
++ bool MockServer::transitionState(State from, State to)
++ {
++ State expected = from;
++ return mState.compare_exchange_strong(expected, to);
++ }