<commit_before>#include <vector>
#include <future>
#include <edisense_types.h>
#include <list>
#include <zmqpp/message.hpp>
#include <zmqpp/socket.hpp>
#include <sstream>
#include <zmqpp/context.hpp>
#include "client.h"

using namespace edisense_comms;
using namespace std;

Client::Client() {
  context = new zmqpp::context;

  serverSocket = new zmqpp::socket(*context, zmqpp::socket_type::reply);
  clientSocket = new zmqpp::socket(*context, zmqpp::socket_type::request);

  clientSocket->set(zmqpp::socket_option::receive_timeout, 1000);
  clientSocket->set(zmqpp::socket_option::send_timeout, 1000);
  serverSocket->set(zmqpp::socket_option::send_timeout, 1000);
}

void Client::start(ClientServer *subscriber) {
  this->subscriber = subscriber;
  run = true;
  server = std::thread(&Client::startServer, this);
  server.detach();
}

void Client::stop() {
  run = false;
  serverSocket->disconnect(buildEndpoint("*", SERVER_SOCKET_PORT));
}

void Client::startServer() {

  zmqpp::endpoint_t acceptAll = buildEndpoint("*", SERVER_SOCKET_PORT);
  serverSocket->bind(acceptAll);

  while (run) {
    zmqpp::message message;
    serverSocket->receive(message);
    string requestName;
    message >> requestName;
    dispositionRequest(requestName, message);
  }

  serverSocket->unbind(acceptAll);
}

future<list<pair<string, PutResult>>> Client::put(node_t sender, transaction_t tid, list<string> &recipients, device_t deviceId, time_t timestamp, time_t expiration, blob data) {
  return async(launch::async, &Client::remotePut, this, sender, tid, std::ref(recipients), deviceId, timestamp, expiration, data);
}

future<list<GetResult>> Client::get(transaction_t tid, list<string> &recipients, device_t deviceId, time_t begin, time_t end) {
  return async(launch::async, &Client::remoteGet, this, tid, std::ref(recipients), deviceId, begin, end);
}

bool Client::dispositionRequest(string topic, zmqpp::message &request) { // TODO extract to functions
  bool wasRequestProcessed = false;
  zmqpp::message response;
  if (topic == "get") {
    wasRequestProcessed = true;
    transaction_t tid;
    device_t deviceId;
    uint32_t begin;
    uint32_t end;
    request >> tid >> deviceId >> begin >> end;

    GetResult getResult = subscriber->handleGetRequest(tid, deviceId, begin, end);

    response.add((uint8_t) getResult.status);
    response.add(getResult.moved_to);
    response.add((uint32_t) getResult.values->size());
    for (Data data : *getResult.values) {
      response.add((uint32_t)data.data.size());
      response.add_raw(&data.data[0], data.data.size());
      response.add((uint32_t) data.expiration);
      response.add((uint32_t) data.timestamp);
    }
    serverSocket->send(response);
  } else if (topic == "put") {
    wasRequestProcessed = true;
    node_t sender;
    transaction_t tid;
    device_t deviceId;
    uint32_t timestamp;
    uint32_t expiry;
    uint32_t blobSize;
    request >> sender >> tid >> deviceId >> timestamp >> expiry;
    request >> blobSize;
    unsigned char* rawPoint;
    rawPoint = (unsigned char *) request.raw_data(blobSize);
    blob point(rawPoint, rawPoint + blobSize);

    PutResult result = subscriber->handlePutRequest(sender, tid, deviceId, timestamp, expiry, point);
    response.add((uint8_t) result.status);
    response.add(result.moved_to);
    serverSocket->send(response);
  }
  return wasRequestProcessed;<commit_msg>added print stmt to client.cpp<commit_after>#include <vector>
#include <future>
#include <edisense_types.h>
#include <list>
#include <zmqpp/message.hpp>
#include <zmqpp/socket.hpp>
#include <sstream>
#include <zmqpp/context.hpp>
#include "client.h"

using namespace edisense_comms;
using namespace std;

Client::Client() {
  context = new zmqpp::context;

  serverSocket = new zmqpp::socket(*context, zmqpp::socket_type::reply);
  clientSocket = new zmqpp::socket(*context, zmqpp::socket_type::request);

  clientSocket->set(zmqpp::socket_option::receive_timeout, 1000);
  clientSocket->set(zmqpp::socket_option::send_timeout, 1000);
  serverSocket->set(zmqpp::socket_option::send_timeout, 1000);
}

void Client::start(ClientServer *subscriber) {
  this->subscriber = subscriber;
  run = true;
  server = std::thread(&Client::startServer, this);
  server.detach();
}

void Client::stop() {
  run = false;
  serverSocket->disconnect(buildEndpoint("*", SERVER_SOCKET_PORT));
}

void Client::startServer() {

  zmqpp::endpoint_t acceptAll = buildEndpoint("*", SERVER_SOCKET_PORT);
  serverSocket->bind(acceptAll);

  while (run) {
    zmqpp::message message;
    serverSocket->receive(message);
    string requestName;
    message >> requestName;
    dispositionRequest(requestName, message);
  }

  serverSocket->unbind(acceptAll);
}

future<list<pair<string, PutResult>>> Client::put(node_t sender, transaction_t tid, list<string> &recipients, device_t deviceId, time_t timestamp, time_t expiration, blob data) {
  return async(launch::async, &Client::remotePut, this, sender, tid, std::ref(recipients), deviceId, timestamp, expiration, data);
}

future<list<GetResult>> Client::get(transaction_t tid, list<string> &recipients, device_t deviceId, time_t begin, time_t end) {
  return async(launch::async, &Client::remoteGet, this, tid, std::ref(recipients), deviceId, begin, end);
}

bool Client::dispositionRequest(string topic, zmqpp::message &request) { // TODO extract to functions
  bool wasRequestProcessed = false;
  zmqpp::message response;
  if (topic == "get") {
    wasRequestProcessed = true;
    transaction_t tid;
    device_t deviceId;
    uint32_t begin;
    uint32_t end;
    request >> tid >> deviceId >> begin >> end;

    GetResult getResult = subscriber->handleGetRequest(tid, deviceId, begin, end);

    response.add((uint8_t) getResult.status);
    response.add(getResult.moved_to);
    response.add((uint32_t) getResult.values->size());
    for (Data data : *getResult.values) {
      response.add((uint32_t)data.data.size());
      response.add_raw(&data.data[0], data.data.size());
      response.add((uint32_t) data.expiration);
      response.add((uint32_t) data.timestamp);
    }
    serverSocket->send(response);
  } else if (topic == "put") {
    wasRequestProcessed = true;
    node_t sender;
    transaction_t tid;
    device_t deviceId;
    uint32_t timestamp;
    uint32_t expiry;
    uint32_t blobSize;
    request >> sender >> tid >> deviceId >> timestamp >> expiry;
    request >> blobSize;

    cout << sender << " [sender] " << tid << " [tid] " << deviceId << " [did] "  << timestamp << " [time] " 
      << expiry << " [expire] " << blobSize << " [size]" << endl;

    unsigned char* rawPoint;
    rawPoint = (unsigned char *) request.raw_data(blobSize);
    blob point(rawPoint, rawPoint + blobSize);

    PutResult result = subscriber->handlePutRequest(sender, tid, deviceId, timestamp, expiry, point);
    response.add((uint8_t) result.status);
    response.add(result.moved_to);
    serverSocket->send(response);
  }
  return wasRequestProcessed;
{'fraction_non_alphanumeric': 0.14171199541941024, 'fraction_numerical': 0.009447466361294017, 'mean_word_length': 4.982020547945205, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 103, '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': '2431802', 'n_tokens_mistral': 2331, 'n_tokens_neox': 2234, 'n_words': 479}