<commit_before>#ifndef PROTOQUEUE_H
#define PROTOQUEUE_H

#include <functional>
#include <iostream>
#include <memory>
#include <sstream>
#include <string>

#include <zmq.hpp>

#include "address.h"
#include "port.h"
#include "protocontext.h"
#include "topic.h"
#include "type.h"


template <typename T>
class ProtoQueue {
  using socket_ptr = std::unique_ptr<zmq::socket_t, std::function<void(zmq::socket_t*)>>;

  public:
    ProtoQueue() : address_{}, port_{0}, type_{ZMQ_PAIR} {}

    void Send(T t) {
        t.CheckInitialized();
        std::string output;
        t.SerializeToString(&output);
        zmq::message_t message{output.length()};
        memcpy(message.data(), output.data(), output.length());
        socket_->send(message);
    }

    T Receive(bool block=true) {
        zmq::message_t message;
        T t;
        if ((block && socket_->recv(&message)) || socket_->recv(&message, ZMQ_NOBLOCK)) {
            t.ParseFromString(std::string{(char*) message.data(), message.size()});
        }
        return t;
    }

    void Bind() {
        auto close = [] (zmq::socket_t* socket) {
            socket->close();
            delete socket;
        };
        socket_ = socket_ptr(new zmq::socket_t{ProtoContext::Get().zmq, type_.value}, close);
        if (address_.value.empty()) {
            if (port_.value == 0) {
                try {
                    socket_->bind("tcp://*:*");
                } catch (zmq::error_t& e) {
                    std::cerr << "Socket Error [" << e.num() << "]: " << e.what() << std::endl;
                }
                char port_string[1024];
                size_t size = sizeof(port_string);
                socket_->getsockopt(ZMQ_LAST_ENDPOINT, &port_string, &size);
                address_.value = port_string;
                auto& address = address_.value;
                port_.value = std::stoi(address.substr(address.find(':', 4) + 1, address.length()));
            } else {
                try {
                    std::stringstream url;
                    url << "tcp://0.0.0.0:" << port_.value;
                    socket_->bind(url.str().data());
                } catch (zmq::error_t& e) {
                    std::cerr << "Socket Error [" << e.num() << "]: " << e.what() << std::endl;
                }
            }
        } else {
            try {
                auto& address = address_.value;
                socket_->bind(address.data());
                port_.value = std::stoi(address.substr(address.find(':', 4) + 1, address.length()));
            } catch (zmq::error_t& e) {
                std::cerr << "Socket Error [" << e.num() << "]: " << e.what() << std::endl;
            }
        }
    }

    void Connect() {
        if (port_.value <= 0) {
            throw std::runtime_error("Socket port must be above 0");
        }
        auto close = [] (zmq::socket_t* socket) {
            socket->close();
            delete socket;
        };
        socket_ = socket_ptr(new zmq::socket_t{ProtoContext::Get().zmq, type_.value}, close);
        if (type_.value == ZMQ_SUB) {
            socket_->setsockopt(ZMQ_SUBSCRIBE, topic_.value.data(), topic_.value.length());
        }
        try {
            std::stringstream url;
            url << "tcp://0.0.0.0:" << port_.value;
            address_.value = url.str();
            socket_->connect(url.str().data());
        } catch (zmq::error_t& e) {
            std::cerr << "Error connecting socket [" << e.num() << "]: " << e.what() << std::endl;
        }
    }

    void SetOption(const Port& port) {
        port_ = port;
    }
<commit_msg>Always surface the zmq exceptions<commit_after>#ifndef PROTOQUEUE_H
#define PROTOQUEUE_H

#include <functional>
#include <iostream>
#include <memory>
#include <sstream>
#include <string>

#include <zmq.hpp>

#include "address.h"
#include "port.h"
#include "protocontext.h"
#include "topic.h"
#include "type.h"


template <typename T>
class ProtoQueue {
  using socket_ptr = std::unique_ptr<zmq::socket_t, std::function<void(zmq::socket_t*)>>;

  public:
    ProtoQueue() : address_{}, port_{0}, type_{ZMQ_PAIR} {}

    void Send(T t) {
        t.CheckInitialized();
        std::string output;
        t.SerializeToString(&output);
        zmq::message_t message{output.length()};
        memcpy(message.data(), output.data(), output.length());
        socket_->send(message);
    }

    T Receive(bool block=true) {
        zmq::message_t message;
        T t;
        if ((block && socket_->recv(&message)) || socket_->recv(&message, ZMQ_NOBLOCK)) {
            t.ParseFromString(std::string{(char*) message.data(), message.size()});
        }
        return t;
    }

    void Bind() {
        auto close = [] (zmq::socket_t* socket) {
            socket->close();
            delete socket;
        };
        socket_ = socket_ptr(new zmq::socket_t{ProtoContext::Get().zmq, type_.value}, close);
        if (address_.value.empty()) {
            if (port_.value == 0) {
                socket_->bind("tcp://*:*");
                char port_string[1024];
                size_t size = sizeof(port_string);
                socket_->getsockopt(ZMQ_LAST_ENDPOINT, &port_string, &size);
                address_.value = port_string;
                auto& address = address_.value;
                port_.value = std::stoi(address.substr(address.find(':', 4) + 1, address.length()));
            } else {
                try {
                    std::stringstream url;
                    url << "tcp://0.0.0.0:" << port_.value;
                    socket_->bind(url.str().data());
                } catch (zmq::error_t& e) {
                    std::cerr << "Socket Error [" << e.num() << "]: " << e.what() << std::endl;
                    throw e;
                }
            }
        } else {
            try {
                auto& address = address_.value;
                socket_->bind(address.data());
                port_.value = std::stoi(address.substr(address.find(':', 4) + 1, address.length()));
            } catch (zmq::error_t& e) {
                std::cerr << "Socket Error [" << e.num() << "]: " << e.what() << std::endl;
                throw e;
            }
        }
    }

    void Connect() {
        if (port_.value <= 0) {
            throw std::runtime_error("Socket port must be above 0");
        }
        auto close = [] (zmq::socket_t* socket) {
            socket->close();
            delete socket;
        };
        socket_ = socket_ptr(new zmq::socket_t{ProtoContext::Get().zmq, type_.value}, close);
        if (type_.value == ZMQ_SUB) {
            socket_->setsockopt(ZMQ_SUBSCRIBE, topic_.value.data(), topic_.value.length());
        }
        try {
            std::stringstream url;
            url << "tcp://0.0.0.0:" << port_.value;
            address_.value = url.str();
            socket_->connect(url.str().data());
        } catch (zmq::error_t& e) {
            std::cerr << "Error connecting socket [" << e.num() << "]: " << e.what() << std::endl;
        }
    }

    void SetOption(const Port& port) {
        port_ = port;
    }

{'fraction_non_alphanumeric': 0.1673716864072194, 'fraction_numerical': 0.0056401579244218835, 'mean_word_length': 1.8520305589063129, 'pattern_counts': {'":': 0, '<': 109, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '2592559', 'n_tokens_mistral': 2219, 'n_tokens_neox': 2129, 'n_words': 418}