<commit_before>#ifndef __SOCKET_H__
#define __SOCKET_H__

#include <string>
#include <vector>
#include <cstdint>
#include <zmq.h>
#include <cassert>
#include <iostream>

#include "err.h"

typedef std::vector<uint8_t> DataVec;

struct Socket {

Socket(void *ctx, int type, std::string addr_) : addr(addr_) {
  sock = zmq_socket(ctx, type);
  if (!sock)
    throw Err("Couldnt create socket");

  if (type == ZMQ_REP || type == ZMQ_PUB) {
    std::cout << "binding to " << addr << std::endl;
    if (zmq_bind(sock, addr.c_str()) == -1)
      throw Err("couldnt connect");
  } else {
    std::cout << "connecting to " << addr << std::endl;
    if (zmq_connect(sock, addr.c_str()) == -1)
      throw Err("couldnt connect");
  }
}

  void send(DataVec &data) {
    zmq_msg_t *msg = new zmq_msg_t;
    int rc = zmq_msg_init_size (msg, data.size());
    assert(rc != -1);
    std::copy(&data[0], &data[data.size()], (char*)zmq_msg_data(msg));

    rc = zmq_msg_send(msg, sock, 0);//ZMQ_DONTWAIT for polling
    assert (rc != -1);
  }

  DataVec recv() {
    zmq_msg_t *msg = new zmq_msg_t;
    int rc = zmq_msg_init (msg);
    if (rc == -1)
      throw Err("msg init failed");
    
    rc = zmq_msg_recv(msg, sock, 0);//ZMQ_DONTWAIT for polling
    if (rc == -1)
      throw Err("recieving failed failed");

    DataVec data(rc);
    char *data_ptr = (char*)zmq_msg_data(msg);
    std::copy(data_ptr, data_ptr + rc, &data[0]);
    return data;
  }

  void subscribe(std::string pref) {
    zmq_setsockopt (sock, ZMQ_SUBSCRIBE, pref.c_str(), pref.size());
  }
  
  std::string addr;
  void *sock;
};

struct Context {
  void *ctx;
  static Context *s;

Context() : ctx(0) {
    init();
  }
  
  static Context &inst() {
    if (!Context::s) {
      Context::s = new Context();
    }
      
    return *Context::s;
  }

  void init() {
    ctx = zmq_ctx_new();
    if (!ctx)
      throw Err("Couldnt create context");<commit_msg>pub sub<commit_after>#ifndef __SOCKET_H__
#define __SOCKET_H__

#include <string>
#include <vector>
#include <cstdint>
#include <zmq.h>
#include <cassert>
#include <iostream>

#include "err.h"

typedef std::vector<uint8_t> DataVec;

struct Socket {

Socket(void *ctx, int type_, std::string addr_) : type(type_), addr(addr_) {
  sock = zmq_socket(ctx, type);
  if (!sock)
    throw Err("Couldnt create socket");

  if (type == ZMQ_REP || type == ZMQ_PUB) {
    std::cout << "binding to " << addr << std::endl;
    if (zmq_bind(sock, addr.c_str()) == -1)
      throw Err("couldnt connect");
  } else {
    std::cout << "connecting to " << addr << std::endl;
    if (zmq_connect(sock, addr.c_str()) == -1)
      throw Err("couldnt connect");
  }
}

  void send(DataVec &data) {
    zmq_msg_t *msg = new zmq_msg_t;
    int rc = zmq_msg_init_size (msg, data.size());
    assert(rc != -1);
    std::copy(&data[0], &data[data.size()], (char*)zmq_msg_data(msg));

    rc = zmq_msg_send(msg, sock, 0);//ZMQ_DONTWAIT for polling
    assert (rc != -1);
  }

  DataVec recv() {
    zmq_msg_t *msg = new zmq_msg_t;
    int rc = zmq_msg_init (msg);
    if (rc == -1)
      throw Err("msg init failed");
    
    rc = zmq_msg_recv(msg, sock, 0);//ZMQ_DONTWAIT for polling
    if (rc == -1)
      throw Err("recieving failed failed");

    DataVec data(rc);
    char *data_ptr = (char*)zmq_msg_data(msg);
    std::copy(data_ptr, data_ptr + rc, &data[0]);
    return data;
  }

  void subscribe(std::string pref) {
    assert(type = ZMQ_SUB);
    zmq_setsockopt (sock, ZMQ_SUBSCRIBE, pref.c_str(), pref.size());
  }
  
  std::string addr;
  int type;
  void *sock;
};

struct Context {
  void *ctx;
  static Context *s;

Context() : ctx(0) {
    init();
  }
  
  static Context &inst() {
    if (!Context::s) {
      Context::s = new Context();
    }
      
    return *Context::s;
  }

  void init() {
    ctx = zmq_ctx_new();
    if (!ctx)
      throw Err("Couldnt create context");
{'fraction_non_alphanumeric': 0.169238683127572, 'fraction_numerical': 0.006172839506172839, 'mean_word_length': 3.15491452991453, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 17, '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': '3150262', 'n_tokens_mistral': 1542, 'n_tokens_neox': 1508, 'n_words': 361}