The agrpc::GrpcContext hangs forever about asio-grpc HOT 13 CLOSED

    if (!m_is_shutdown.exchange(true)) {                                                              
        m_shutdown_thread = std::thread([this] {                                                                                  
            m_signals.cancel();                                                                       
            m_server->Shutdown();                                                                     
            std::cout << "I see this line" << std::endl;                                  
        });                                                                                           
    }

    m_grpc_context->run();                                                                            
    std::cout << "I see this as well" << std::endl;

detail::drain_completion_queue(*this);

inline void drain_completion_queue(agrpc::GrpcContext& grpc_context)
{
    while (detail::GrpcContextImplementation::do_one(grpc_context, detail::GrpcContextImplementation::INFINITE_FUTURE,
                                                     detail::InvokeHandler::NO, detail::AlwaysFalsePredicate{}))
    {
        std::cout << "------ in the loop" << std::endl;
    }
}

class MockServices {
public:
    MockServices(char const* address = mock::default_server_address)
        : m_grpc_context { m_builder.AddCompletionQueue() }
    {
        m_builder.AddListeningPort(address, grpc::InsecureServerCredentials());
        // Calls m_builder.RegisterService for many services.
        m_server = m_builder.BuildAndStart();
    }

    ~MockServices()
    {
        if (m_shutdown_thread.joinable()) {
            m_shutdown_thread.join();
        } else if (!m_is_shutdown.exchange(true)) {
            m_server->Shutdown();
        }

        m_server.reset();
    }

    void run()
    {
        m_grpc_context.run();
    }

    void stop()
    {
        if (!m_is_shutdown.exchange(true)) {
            m_shutdown_thread = std::thread([this] {
                m_server->Shutdown();
            });
        }
    }
    ...

private:
    std::unique_ptr<grpc::Server> m_server;
    grpc::ServerBuilder m_builder;
    agrpc::GrpcContext m_grpc_context;
    // Here are our services which we use in the constructor.
    std::atomic_bool m_is_shutdown { false };
    std::thread m_shutdown_thread;
};

// Copyright 2023 Dennis Hezel
//
// 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 "awaitable_client_rpc.hpp"
#include "awaitable_server_rpc.hpp"
#include "helloworld/helloworld.grpc.pb.h"
#include "helper.hpp"

#include <agrpc/asio_grpc.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/signal_set.hpp>
#include <grpcpp/client_context.h>
#include <grpcpp/create_channel.h>
#include <grpcpp/server.h>
#include <grpcpp/server_builder.h>

namespace asio = boost::asio;

class MockServices
{
  public:
    MockServices(char const* address) : m_grpc_context{m_builder.AddCompletionQueue()}
    {
        m_builder.AddListeningPort(address, grpc::InsecureServerCredentials());
        m_builder.RegisterService(&service);
        m_server = m_builder.BuildAndStart();
    }

    ~MockServices()
    {
        if (m_shutdown_thread.joinable())
        {
            m_shutdown_thread.join();
        }
        else if (!m_is_shutdown.exchange(true))
        {
            m_server->Shutdown();
        }

        m_server.reset();
    }

    void run() { m_grpc_context.run(); }

    void stop()
    {
        if (!m_is_shutdown.exchange(true))
        {
            m_shutdown_thread = std::thread(
                [this]
                {
                    m_server->Shutdown();
                });
        }
    }

    std::unique_ptr<grpc::Server> m_server;
    grpc::ServerBuilder m_builder;
    agrpc::GrpcContext m_grpc_context;
    helloworld::Greeter::AsyncService service;
    std::atomic_bool m_is_shutdown{false};
    std::thread m_shutdown_thread;
};

int main(int argc, const char** argv)
{
    const auto port = argc >= 2 ? argv[1] : "50051";
    const auto host = std::string("localhost:") + port;

    grpc::Status status;

    helloworld::Greeter::Stub stub{grpc::CreateChannel(host, grpc::InsecureChannelCredentials())};
    agrpc::GrpcContext grpc_context;

    MockServices s{"0.0.0.0:50051"};

    using RPC = example::AwaitableServerRPC<&helloworld::Greeter::AsyncService::RequestSayHello>;
    agrpc::register_awaitable_rpc_handler<RPC>(
        s.m_grpc_context, s.service,
        [&](RPC& rpc, RPC::Request& request) -> asio::awaitable<void>
        {
            helloworld::HelloReply response;
            response.set_message("Hello " + request.name());
            co_await rpc.finish(response, grpc::Status::OK);
        },
        asio::detached);

    std::thread t{[&]
                  {
                      s.run();
                  }};

    asio::co_spawn(
        grpc_context,
        [&]() -> asio::awaitable<void>
        {
            using CRPC = example::AwaitableClientRPC<&helloworld::Greeter::Stub::PrepareAsyncSayHello>;
            grpc::ClientContext client_context;
            helloworld::HelloRequest request;
            request.set_name("world");
            helloworld::HelloReply response;
            status = co_await CRPC::request(grpc_context, stub, client_context, request, response);
            std::cout << status.ok() << " response: " << response.message() << std::endl;
            s.stop();
        },
        asio::detached);

    grpc_context.run();

    abort_if_not(status.ok());
    t.join();
}