<commit_before>//
// Created by <NAME> on 03.02.2017.
//

#include "catch.hpp"

#include "server/nanomsg/publisher_server.h"
#include "server/nanomsg/subscriber_client.h"
#include "threads/thread.h"

#include <atomic>
#include <chrono>
#include <memory>
#include <vector>

using namespace CppCommon;
using namespace CppServer::Nanomsg;

class TestSubscriberClient : public SubscriberClient
{
public:
    std::atomic<bool> connected;
    std::atomic<bool> disconnected;
    std::atomic<bool> error;

    explicit TestSubscriberClient(const std::string& address, const std::string& topic = "")
        : SubscriberClient(address, topic),
          connected(false),
          disconnected(false),
          error(false)
    {
    }

protected:
    void onConnected() override { connected = true; }
    void onDisconnected() override { disconnected = true; }
    void onError(int error, const std::string& message) override { error = true; }
};

class TestPublisherServer : public PublisherServer
{
public:
    std::atomic<bool> started;
    std::atomic<bool> stopped;
    std::atomic<bool> error;

    explicit TestPublisherServer(const std::string& address)
        : PublisherServer(address),
          started(false),
          stopped(false),
          error(false)
    {
    }

protected:
    void onStarted() override { started = true; }
    void onStopped() override { stopped = true; }
    void onError(int error, const std::string& message) override { error = true; }
};

TEST_CASE("Nanomsg subscriber client & publisher server", "[CppServer][Nanomsg]")
{
    const std::string server_address = "tcp://*:6672";
    const std::string client_address = "tcp://localhost:6672";

    // Create and start Nanomsg publisher server
    auto server = std::make_shared<TestPublisherServer>(server_address);
    REQUIRE(server->Start());
    while (!server->IsStarted())
        Thread::Yield();

    // Create and connect Nanomsg subscriber client
    auto client = std::make_shared<TestSubscriberClient>(client_address, "foo");
    REQUIRE(client->Connect());
    while (!client->IsConnected())
        Thread::Yield();

    while (client->socket().bytes_received() == 0)
    {
        // Publish messages to all subscribed clients
        server->Send("test");
        server->Send("footest");
        Thread::Yield();
    }

    // Disconnect the client
    REQUIRE(client->Disconnect());
    while (client->IsConnected())
        Thread::Yield();

    // Stop the server
    REQUIRE(server->Stop());
    while (server->IsStarted())
        Thread::Yield();

    // Check the server state
    REQUIRE(server->started);
    REQUIRE(server->stopped);
    REQUIRE(server->socket().accepted_connections() == 1);
    REQUIRE(server->socket().messages_sent() > 0);
    REQUIRE(server->socket().bytes_sent() > 0);
    REQUIRE(!server->error);

    // Check the client state
    REQUIRE(client->connected);
    REQUIRE(client->disconnected);
    REQUIRE(client->socket().established_connections() == 1);
    REQUIRE(client->socket().messages_received() > 0);
    REQUIRE(client->socket().bytes_received() > 0);
    REQUIRE(!client->error);
}

TEST_CASE("Nanomsg publish/subscribe random test", "[CppServer][Nanomsg]")<commit_msg>Fix tests<commit_after>//
// Created by <NAME> on 03.02.2017.
//

#include "catch.hpp"

#include "server/nanomsg/publisher_server.h"
#include "server/nanomsg/subscriber_client.h"
#include "threads/thread.h"

#include <atomic>
#include <chrono>
#include <memory>
#include <vector>

using namespace CppCommon;
using namespace CppServer::Nanomsg;

class TestSubscriberClient : public SubscriberClient
{
public:
    std::atomic<bool> connected;
    std::atomic<bool> disconnected;
    std::atomic<bool> error;

    explicit TestSubscriberClient(const std::string& address, const std::string& topic = "")
        : SubscriberClient(address, topic),
          connected(false),
          disconnected(false),
          error(false)
    {
    }

protected:
    void onConnected() override { connected = true; }
    void onDisconnected() override { disconnected = true; }
    void onError(int error, const std::string& message) override { error = true; }
};

class TestPublisherServer : public PublisherServer
{
public:
    std::atomic<bool> started;
    std::atomic<bool> stopped;
    std::atomic<bool> error;

    explicit TestPublisherServer(const std::string& address)
        : PublisherServer(address),
          started(false),
          stopped(false),
          error(false)
    {
    }

protected:
    void onStarted() override { started = true; }
    void onStopped() override { stopped = true; }
    void onError(int error, const std::string& message) override { error = true; }
};

TEST_CASE("Nanomsg subscriber client & publisher server", "[CppServer][Nanomsg]")
{
    const std::string server_address = "tcp://*:6672";
    const std::string client_address = "tcp://localhost:6672";

    // Create and start Nanomsg publisher server
    auto server = std::make_shared<TestPublisherServer>(server_address);
    REQUIRE(server->Start());
    while (!server->IsStarted())
        Thread::Yield();

    // Create and connect Nanomsg subscriber client
    auto client = std::make_shared<TestSubscriberClient>(client_address, "foo");
    REQUIRE(client->Connect());
    while (!client->IsConnected())
        Thread::Yield();

    // Sleep for a while...
    Thread::Sleep(100);

    // Publish messages to all subscribed clients
    server->Send("test");

    // Wait for all data processed...
    while (client->socket().bytes_received() != 4)
        Thread::Yield();

    // Disconnect the client
    REQUIRE(client->Disconnect());
    while (client->IsConnected())
        Thread::Yield();

    // Stop the server
    REQUIRE(server->Stop());
    while (server->IsStarted())
        Thread::Yield();

    // Check the server state
    REQUIRE(server->started);
    REQUIRE(server->stopped);
    REQUIRE(server->socket().accepted_connections() == 1);
    REQUIRE(server->socket().messages_sent() == 1);
    REQUIRE(server->socket().bytes_sent() == 4);
    REQUIRE(!server->error);

    // Check the client state
    REQUIRE(client->connected);
    REQUIRE(client->disconnected);
    REQUIRE(client->socket().established_connections() == 1);
    REQUIRE(client->socket().messages_received() == 1);
    REQUIRE(client->socket().bytes_received() == 4);
    REQUIRE(!client->error);
}

TEST_CASE("Nanomsg publish/subscribe random test", "[CppServer][Nanomsg]")
{'fraction_non_alphanumeric': 0.1462737231908656, 'fraction_numerical': 0.007560561641721956, 'mean_word_length': 3.636623748211731, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 78, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6693571', 'n_tokens_mistral': 2039, 'n_tokens_neox': 1911, 'n_words': 425}