qt-migration-legatus/qt6-after/qt6-migrated-legatus/include/rti/apputils/http/client/test/CivetWebSocketConnectionTester.h

/*
 * (c) Copyright, Real-Time Innovations, 2020.
 * All rights reserved.
 *
 * No duplications, whole or partial, manual or electronic, may be made
 * without express written permission.  Any such copies, or
 * revisions thereof, must display this notice unaltered.
 * This code contains trade secrets of Real-Time Innovations, Inc.
 */

#ifndef CIVETWEBSOCKET_TEST_CONNECTION_CLIENT_H
#define CIVETWEBSOCKET_TEST_CONNECTION_CLIENT_H

#include <memory>
#include <mutex>
#include "civetweb.h"
#include <queue>
#include <rti/apputils/http/server/CivetWebSocketConnection.hpp>
#include <rti/apputils/log/LogConfig.hpp>
#include "log/LogMsg.hpp"

// #include "utils/WebDDSUtils.h"

#define CIVETWEB_CLIENT_TIMEOUT_MS 1000
#define WEBSOCKET_ERROR_BUFFER_SIZE 10000
#define WEBSOCKETinput_buffer__SIZE 10000
#define WEBSERVER_DEFAULT_PORT 19020

namespace rti { namespace apputils { namespace http { namespace client {
namespace test {

class CivetWebSocketClient {
public:
    ~CivetWebSocketClient()
    {
        mg_close_connection(conn_);
    }

    static int websocket_client_data_handler(
            struct mg_connection * /*conn*/,
            int flags,
            char *data,
            size_t data_len,
            void *user_data)
    {
        CivetWebSocketClient *user = NULL;

        if (user_data == NULL) {
            SERVICELog_exception(
                    &RTI_LOG_ANY_FAILURE_s,
                    "No user data defined");
            return 1;
        }

        switch (((unsigned) flags & 0xFu)) {
        case MG_WEBSOCKET_OPCODE_TEXT:
            user = reinterpret_cast<CivetWebSocketClient *>(user_data);
            user->receive(std::string(data, data_len));
            break;
        default:
            SERVICELog_exception(
                    &RTI_LOG_ANY_FAILURE_s,
                    "The websocket message code received is not recognize");
            break;
        }

        return 1;
    }

    static void websocket_client_close_handler(
            const struct mg_connection *conn,
            void *user_data)
    {
        struct mg_context *context = mg_get_context(conn);

        if (user_data == NULL) {
            SERVICELog_exception(
                    &RTI_LOG_ANY_FAILURE_s,
                    "No user data defined");
            return;
        }

        mg_lock_context(context);

        mg_set_user_connection_data((struct mg_connection *) conn, NULL);

        mg_unlock_context(context);
    }

    static std::shared_ptr<CivetWebSocketClient> connect(
            const char *host = "localhost",
            int port = WEBSERVER_DEFAULT_PORT,
            bool use_ssl = false,
            const char *path = "/dds/websocket/test")
    {
        StringBuffer error_buffer(WEBSOCKET_ERROR_BUFFER_SIZE);
        auto result = std::make_shared<CivetWebSocketClient>();

        struct mg_connection *conn = mg_connect_websocket_client(
                host,
                port,
                use_ssl,
                (char *) error_buffer.c_str_ptr(),
                WEBSOCKET_ERROR_BUFFER_SIZE,
                path,
                NULL,
                websocket_client_data_handler,
                websocket_client_close_handler,
                result.get());

        if (conn == NULL) {
            SERVICELog_exception(&RTI_LOG_ANY_FAILURE_s, "No Connection");
            result.reset();
            return result;
        }

        result->connection(conn);
        return result;
    }

public:
    CivetWebSocketClient() = default;

    CivetWebSocketClient(mg_connection *conn)
            : conn_(conn)
    { }

    void send(const StringBuffer& msg)
    {
        mg_websocket_client_write(
                conn_,
                MG_WEBSOCKET_OPCODE_TEXT,
                msg.c_str(),
                msg.length());
    }

    void request_close()
    {
        mg_websocket_client_write(
                conn_,
                MG_WEBSOCKET_OPCODE_CONNECTION_CLOSE,
                NULL,
                0);
    }

    void receive(const StringBuffer& buffer)
    {
        std::lock_guard<std::recursive_mutex> guard(mutex_);
        received_buffer_.push(buffer);
    }

    bool buffer(StringBuffer& buffer)
    {
        std::lock_guard<std::recursive_mutex> guard(mutex_);
        bool received = false;
        if (!received_buffer_.empty()) {
            buffer = received_buffer_.front();
            received_buffer_.pop();
            received = true;
        }
        return received;
    }

    void connection(mg_connection *conn)
    {
        conn_ = conn;
    }

private:
    mg_connection *conn_ = nullptr;
    std::queue<StringBuffer> received_buffer_;
    std::recursive_mutex mutex_;
};

}}}}}  // namespace rti::apputils::http::client::test

#endif