mesytec-mnode/src/mnode_proto_ping_client.cc

#include <memory>

#include <mesytec-mvlc/mesytec-mvlc.h>
#include <mesytec-mvlc/util/signal_handling.h>
#include "proto/service.pb.h"
#include <mesytec-mnode/mnode_nng.h>

using namespace mesytec;
using namespace mesytec::mnode;
using namespace std::literals;

void client_cb(void *arg);

class Work
{
  public:
    explicit Work(nng_socket socket)
        : socket(socket)
        , request_(nng::make_unique_msg())
    {
    }

    ~Work()
    {
        nng_ctx_close(ctx);
        nng_aio_free(aio);
    }

    void work()
    {
        switch (state)
        {
        case INIT:
            nng_aio_alloc(&aio, client_cb, this);
            nng_ctx_open(&ctx, socket);

            state = SEND;
            request_ = make_request();
            nng_aio_set_msg(aio, nng::clone_message(request_.get()).release());
            nng_ctx_send(ctx, aio);
            break;

        case SEND:
            if (auto rv = nng_aio_result(aio))
            {
                nng::mesy_nng_error("nng_ctx_send", rv);
                nng::make_unique_msg(nng_aio_get_msg(aio)).reset();
                state = SEND;
                request_ = make_request();
                nng_aio_set_msg(aio, nng::clone_message(request_.get()).release());
                nng_ctx_send(ctx, aio);
            }
            else
            {
                state = RECEIVE;
                nng_ctx_recv(ctx, aio);
            }
            break;

        case RECEIVE:
            if (auto rv = nng_aio_result(aio))
            {
                nng::mesy_nng_error("nng_ctx_recv", rv);
                state = SEND;
                request_ = make_request();
                nng_aio_set_msg(aio, nng::clone_message(request_.get()).release());
                nng_ctx_send(ctx, aio);
            }
            else
            {
                auto reply = nng::make_unique_msg(nng_aio_get_msg(aio));
                handle_reply(std::move(request_), std::move(reply));
                state = SEND;
                request_ = make_request();
                nng_aio_set_msg(aio, nng::clone_message(request_.get()).release());
                nng_ctx_send(ctx, aio);
            }

            break;
        }

    }

    nng::unique_msg make_request()
    {
        Ping ping;
        ping.set_peer_id(42);
        ping.set_sequence_number(++sequence_number);
        return nng::make_message(ping.SerializeAsString());
    }

    void handle_reply(nng::unique_msg &&request, nng::unique_msg &&reply)
    {
        Pong pong;
        pong.ParseFromArray(nng_msg_body(reply.get()), nng_msg_len(reply.get()));
        if (pong.peer_id() != 42 || pong.sequence_number() != sequence_number)
        {
            spdlog::error("received pong with unexpected values: {}", pong.ShortDebugString());
        }
    }

    void report()
    {
        spdlog::info("Work: state={}, sequence_number={}", static_cast<int>(state), sequence_number);
    }

  private:
    enum State
    {
        INIT,    // allocate aio, open ctx, send first request
        SEND,    // send request
        RECEIVE, // receive response
    };

    size_t sequence_number = 0;
    State state = INIT;
    nng_socket socket;
    nng_aio *aio = nullptr;
    nng_ctx ctx;

    nng::unique_msg request_;
    Ping ping;
    Pong pong;
};

void client_cb(void *arg)
{
    auto work = static_cast<Work *>(arg);
    work->work();
}

int main()
{
    spdlog::set_level(spdlog::level::info);

    auto socket = nng::make_req_socket();

    if (int res = nng_dial(socket, "tcp://localhost:5555", nullptr, 0))
    {
        nng::mesy_nng_error("nng_dial", res);
        return res;
    }

    mvlc::util::Stopwatch sw;
    Work work(socket);
    work.work();

    for (;;)
    {
        nng_msleep(100);
        if (sw.get_interval() >= 1s)
        {
            work.report();
            sw.interval();
        }
    }

    return 0;
}