How to detect client closed connection about asio-grpc HOT 6 CLOSED

#pragma once

#include <functional>
#include <string>

#include "agrpc/asio_grpc.hpp"
#include "boost/asio.hpp"
#include "boost/asio/experimental/promise.hpp"
#include "boost/asio/experimental/use_promise.hpp"
#include "grpc/grpc.h"

namespace my_namespace {

using OnDonePromise = boost::asio::experimental::promise<
    void(),
    boost::asio::basic_system_executor<boost::asio::execution::detail::blocking::possibly_t<0>,
                                       boost::asio::execution::detail::relationship::fork_t<0>,
                                       std::allocator<void>>,
    std::allocator<void>>;

template <class Request, class Response>
using ResponseStreamingRequestHandlerFn = std::function<boost::asio::awaitable<void>(
    grpc::ServerContext&, Request&, grpc::ServerAsyncWriter<Response>&, OnDonePromise)>;

namespace detail {

template <class Request, class Response, class AsyncService, class AsyncRequestFn>
boost::asio::awaitable<void> ResponseStreamingRequestLoop(
    agrpc::GrpcContext& grpc_context, AsyncService& service, AsyncRequestFn async_request_fn,
    ResponseStreamingRequestHandlerFn<Request, Response> request_handler) {
  grpc::ServerContext server_context;
  OnDonePromise on_done =
      agrpc::notify_when_done(grpc_context, server_context, boost::asio::experimental::use_promise);
  Request request;
  grpc::ServerAsyncWriter<Response> response_writer{&server_context};
  const bool request_ok =
      co_await agrpc::request(async_request_fn, service, server_context, request, response_writer,
                              boost::asio::use_awaitable);
  if (!request_ok) {
    // At this point, `agrpc::notify_when_done` will never complete.
    grpc_context.work_finished();
    co_return;
  }
  boost::asio::co_spawn(grpc_context,
                        ResponseStreamingRequestLoop(grpc_context, service,
                                                     std::move(async_request_fn), request_handler),
                        boost::asio::detached);
  co_await request_handler(server_context, request, response_writer, std::move(on_done));
}

}  // namespace detail

// This helper function provides the same functionality as agrpc::repeatedly_request, with the
// addition of passing an on_done promise to the callback. This enables the callback to check if the
// client has closed the connection.
template <class Request, class Response, class AsyncService, class AsyncRequestFn>
void RepeatedRequestResponseStreaming(
    agrpc::GrpcContext& grpc_context, AsyncService& service, AsyncRequestFn async_request_fn,
    ResponseStreamingRequestHandlerFn<Request, Response> request_handler) {
  boost::asio::co_spawn(
      grpc_context,
      detail::ResponseStreamingRequestLoop(grpc_context, service, std::move(async_request_fn),
                                           std::move(request_handler)),
      boost::asio::detached);
}

}  // namespace my_namespace

  auto request_handler =
      [this](grpc::ServerContext& context, proto::MyServiceRequest& request,
             grpc::ServerAsyncWriter<proto::MyServiceResponse>& response_writer,
             OnDonePromise on_done) -> asio::awaitable<void> {
    while (true) {
      if (new_data_is_available) {
        if (!co_await agrpc::write(response_writer, proto::MyServiceResponse{})) {
          break;
        }
      }
      grpc::Alarm alarm;
      co_await agrpc::wait(alarm, std::chrono::system_clock::now() + std::chrono::seconds{1});
      if (on_done.completed()) {
        break;
      }
    }
    co_await agrpc::finish(response_writer, grpc::Status::OK);
  };
  RepeatedRequestResponseStreaming<proto::MyServiceRequest, proto::MyServiceResponse>(
      grpc_context, service, &proto::MyService::AsyncService::RequestMyEndpoint,
      std::move(request_handler));

#pragma once

#include <functional>
#include <string>

#include "agrpc/asio_grpc.hpp"
#include "boost/asio.hpp"
#include "boost/asio/experimental/promise.hpp"
#include "boost/asio/experimental/use_promise.hpp"
#include "grpc/grpc.h"

namespace my_namespace {

using OnDonePromise = boost::asio::experimental::promise<
    void(), asio::system_executor>;

namespace detail {

template <class AsyncService, class AsyncRequestFn>
boost::asio::awaitable<void> ResponseStreamingRequestLoop(
    agrpc::GrpcContext& grpc_context, AsyncService& service, AsyncRequestFn async_request_fn,
    RequestHandler request_handler) {
  grpc::ServerContext server_context;
  OnDonePromise on_done =
      agrpc::notify_when_done(grpc_context, server_context, boost::asio::experimental::use_promise);
  Request request;
  grpc::ServerAsyncWriter<Response> response_writer{&server_context};
  const bool request_ok =
      co_await agrpc::request(async_request_fn, service, server_context, request, response_writer,
                              boost::asio::use_awaitable);
  if (!request_ok) {
    // At this point, `agrpc::notify_when_done` will never complete.
    grpc_context.work_finished();
    co_return;
  }
  boost::asio::co_spawn(grpc_context,
                        ResponseStreamingRequestLoop(grpc_context, service,
                                                     async_request_fn, request_handler),
                        boost::asio::detached);
  co_await request_handler(server_context, request, response_writer, std::move(on_done));
}

}  // namespace detail

// This helper function provides the same functionality as agrpc::repeatedly_request, with the
// addition of passing an on_done promise to the callback. This enables the callback to check if the
// client has closed the connection.
template <class AsyncService, class AsyncRequestFn, class RequestHandler>
void RepeatedRequestResponseStreaming(
    agrpc::GrpcContext& grpc_context, AsyncService& service, AsyncRequestFn async_request_fn,
    RequestHandler request_handler) {
  boost::asio::co_spawn(
      grpc_context,
      detail::ResponseStreamingRequestLoop(grpc_context, service, async_request_fn,
                                           std::move(request_handler)),
      boost::asio::detached);
}

// ...
  auto request_handler =
      [this](grpc::ServerContext& context, proto::MyServiceRequest& request,
             grpc::ServerAsyncWriter<proto::MyServiceResponse>& response_writer,
             OnDonePromise on_done) -> asio::awaitable<void> {
    while (true) {
      if (new_data_is_available) {
        if (!co_await agrpc::write(response_writer, proto::MyServiceResponse{})) {
          break;
        }
      }
      grpc::Alarm alarm;
      co_await agrpc::wait(alarm, std::chrono::system_clock::now() + std::chrono::seconds{1});
      if (on_done.completed()) {
        break;
      }
    }
    co_await agrpc::finish(response_writer, grpc::Status::OK);
  };
  RepeatedRequestResponseStreaming(
      grpc_context, service, &proto::MyService::AsyncService::RequestMyEndpoint,
      std::move(request_handler));

}  // namespace my_namespace