<commit_before>#include <node.h>
#include <node_buffer.h>
#include <zmq.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "zmqstream.h"

using namespace v8;
using namespace node;

namespace zmqstream {
  // TODO: Is there any reason to have more than one Context?
  // Would that make managing blocking sockets (REQ, DEALER, PUSH) easier?
  ScopedContext gContext;
  Persistent<Function> Socket::constructor;

  //
  // ## ScopedContext
  //
  // At the moment, there's a 1:1 relationship between Node processes and ZMQ contexts. This scope-based wrapper
  // is used to ensure that global context is managed properly over the life of that process.
  //
  ScopedContext::ScopedContext() {
    context = zmq_ctx_new();
    assert(context != 0);
  }

  ScopedContext::~ScopedContext() {
    assert(zmq_ctx_destroy(context) == 0 || zmq_errno() == EINTR);
  }

  //
  // ## Helpers
  //
  // These "throw" helpers bail the calling context immediately.
  //
  #define THROW(str) return ThrowException(Exception::Error(String::New(str)));
  #define THROW_REF(str) return ThrowException(Exception::ReferenceError(String::New(str)));
  #define THROW_TYPE(str) return ThrowException(Exception::TypeError(String::New(str)));

  //
  // ZMQ provides its own error messages, so we'll pass those through to V8.
  //
  #define ZMQ_THROW() return ThrowException(Exception::Error(String::New(zmq_strerror(zmq_errno()))));
  #define ZMQ_CHECK(rc) if (!isSuccessRC(rc)) return ThrowException(Exception::Error(String::New(zmq_strerror(zmq_errno()))));

  //
  // PUSH shorthand for v8::Array.
  //
  #define PUSH(a, v) a->Set(a->Length(), v);

  //
  // Internal versions of Unwrap.
  //
  #define SOCKET_TO_THIS(obj) obj->handle_
  ; // For Sublime's syntax highlighter. Ignore.
  #define THIS_TO_SOCKET(obj) ObjectWrap::Unwrap<Socket>(obj)

  //
  // Internal version of NODE_DEFINE_CONSTANT to allow for names without the ZMQ_ prefix.
  //
  #define ZMQ_DEFINE_CONSTANT(target, name, constant)                  \
    (target)->Set(                                                     \
      v8::String::NewSymbol(name),                                     \
      v8::Integer::New(constant),                                      \
      static_cast<v8::PropertyAttribute>(v8::ReadOnly|v8::DontDelete)  \
    );

  //
  // Because we use ZMQ in a non-blocking way, EAGAIN holds a special place in our hearts.
  //
  static inline bool isEAGAIN(int rc) {
    return rc == -1 && zmq_errno() == EAGAIN;
  }

  //
  // ZMQ returns -1 on failure, so this is a simple test for that _ignoring_ EAGAIN.
  //
  static inline bool isSuccessRC(int rc) {
    // Ignore EAGAIN. EAGAIN is handled in a sane, call-specific manner.
    return rc != -1 || isEAGAIN(rc);
  }

  //
  // ## Socket
  //
  // Much like the native `net` module, a ZMQStream socket (perhaps obviously) is really just a Duplex stream that
  // you can `connect`, `bind`, etc. just like a native ZMQ socket.
  //
  Socket::Socket(int type) : ObjectWrap(), shouldDrain(false), shouldReadable(true) {
    this->socket = zmq_socket(gContext.context, type);
    assert(this->socket != 0);

    uv_os_sock_t fd;
    size_t size = sizeof fd;
    zmq_getsockopt(this->socket, ZMQ_FD, &fd, &size);

    assert(uv_poll_init_socket(uv_default_loop(), &readableHandle, fd) == 0);
    readableHandle.data = this;
    assert(uv_poll_init_socket(uv_default_loop(), &writableHandle, fd) == 0);
    writableHandle.data = this;
    assert(uv_idle_init(uv_default_loop(), &idleHandle) == 0);
    idleHandle.data = this;
  }

  Socket::~Socket() {
    if (this->socket) {
      assert(zmq_close(this->socket) == 0);
    }
  }

  //
  // ## Socket(options)
  //
  // Creates a new **options.type** ZMQ socket. Defaults to PAIR.
  //
  Handle<Value> Socket::New(const Arguments& args) {
    HandleScope scope;

    if (!args.IsConstructCall()) {
      Handle<Value> argv[1] = { args[0] };
      return constructor->NewInstance(1, argv);
    }

    Handle<Object> options;

    if (args.Length() < 1 || !args[0]->IsObject()) {
      options = Object::New();
    } else {
      options = args[0]->ToObject();
    }

    Handle<Integer> type = options->Get(String::NewSymbol("type"))->ToInteger();
    int32_t hwm = options->Get(String::NewSymbol("highWaterMark"))->ToInteger()->Int32Value();

    // Creates a new instance object of this type and wraps it.
    Socket* self = new Socket(type->Value());
    assert(self);<commit_msg>Added uv_close calls.<commit_after>#include <node.h>
#include <node_buffer.h>
#include <zmq.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "zmqstream.h"

using namespace v8;
using namespace node;

namespace zmqstream {
  // TODO: Is there any reason to have more than one Context?
  // Would that make managing blocking sockets (REQ, DEALER, PUSH) easier?
  ScopedContext gContext;
  Persistent<Function> Socket::constructor;

  //
  // ## ScopedContext
  //
  // At the moment, there's a 1:1 relationship between Node processes and ZMQ contexts. This scope-based wrapper
  // is used to ensure that global context is managed properly over the life of that process.
  //
  ScopedContext::ScopedContext() {
    context = zmq_ctx_new();
    assert(context != 0);
  }

  ScopedContext::~ScopedContext() {
    assert(zmq_ctx_destroy(context) == 0 || zmq_errno() == EINTR);
  }

  //
  // ## Helpers
  //
  // These "throw" helpers bail the calling context immediately.
  //
  #define THROW(str) return ThrowException(Exception::Error(String::New(str)));
  #define THROW_REF(str) return ThrowException(Exception::ReferenceError(String::New(str)));
  #define THROW_TYPE(str) return ThrowException(Exception::TypeError(String::New(str)));

  //
  // ZMQ provides its own error messages, so we'll pass those through to V8.
  //
  #define ZMQ_THROW() return ThrowException(Exception::Error(String::New(zmq_strerror(zmq_errno()))));
  #define ZMQ_CHECK(rc) if (!isSuccessRC(rc)) return ThrowException(Exception::Error(String::New(zmq_strerror(zmq_errno()))));

  //
  // PUSH shorthand for v8::Array.
  //
  #define PUSH(a, v) a->Set(a->Length(), v);

  //
  // Internal versions of Unwrap.
  //
  #define SOCKET_TO_THIS(obj) obj->handle_
  ; // For Sublime's syntax highlighter. Ignore.
  #define THIS_TO_SOCKET(obj) ObjectWrap::Unwrap<Socket>(obj)

  //
  // Internal version of NODE_DEFINE_CONSTANT to allow for names without the ZMQ_ prefix.
  //
  #define ZMQ_DEFINE_CONSTANT(target, name, constant)                  \
    (target)->Set(                                                     \
      v8::String::NewSymbol(name),                                     \
      v8::Integer::New(constant),                                      \
      static_cast<v8::PropertyAttribute>(v8::ReadOnly|v8::DontDelete)  \
    );

  //
  // Because we use ZMQ in a non-blocking way, EAGAIN holds a special place in our hearts.
  //
  static inline bool isEAGAIN(int rc) {
    return rc == -1 && zmq_errno() == EAGAIN;
  }

  //
  // ZMQ returns -1 on failure, so this is a simple test for that _ignoring_ EAGAIN.
  //
  static inline bool isSuccessRC(int rc) {
    // Ignore EAGAIN. EAGAIN is handled in a sane, call-specific manner.
    return rc != -1 || isEAGAIN(rc);
  }

  //
  // ## Socket
  //
  // Much like the native `net` module, a ZMQStream socket (perhaps obviously) is really just a Duplex stream that
  // you can `connect`, `bind`, etc. just like a native ZMQ socket.
  //
  Socket::Socket(int type) : ObjectWrap(), shouldDrain(false), shouldReadable(true) {
    this->socket = zmq_socket(gContext.context, type);
    assert(this->socket != 0);

    uv_os_sock_t fd;
    size_t size = sizeof fd;
    zmq_getsockopt(this->socket, ZMQ_FD, &fd, &size);

    assert(uv_poll_init_socket(uv_default_loop(), &readableHandle, fd) == 0);
    readableHandle.data = this;
    assert(uv_poll_init_socket(uv_default_loop(), &writableHandle, fd) == 0);
    writableHandle.data = this;
    assert(uv_idle_init(uv_default_loop(), &idleHandle) == 0);
    idleHandle.data = this;
  }

  Socket::~Socket() {
    if (this->socket) {
      assert(zmq_close(this->socket) == 0);
    }

    uv_close((uv_handle_t*)&writableHandle, NULL);
    uv_close((uv_handle_t*)&readableHandle, NULL);
    uv_close((uv_handle_t*)&idleHandle, NULL);
  }

  //
  // ## Socket(options)
  //
  // Creates a new **options.type** ZMQ socket. Defaults to PAIR.
  //
  Handle<Value> Socket::New(const Arguments& args) {
    HandleScope scope;

    if (!args.IsConstructCall()) {
      Handle<Value> argv[1] = { args[0] };
      return constructor->NewInstance(1, argv);
    }

    Handle<Object> options;

    if (args.Length() < 1 || !args[0]->IsObject()) {
      options = Object::New();
    } else {
      options = args[0]->ToObject();
    }

    Handle<Integer> type = options->Get(String::NewSymbol("type"))->ToInteger();
    int32_t hwm = options->Get(String::NewSymbol("highWaterMark"))->ToInteger()->Int32Value();

    // Creates a new instance object of this type and wraps it.
    Socket* self = new Socket(type->Value());
    assert(self);
{'fraction_non_alphanumeric': 0.1435632688927944, 'fraction_numerical': 0.006590509666080844, 'mean_word_length': 3.077474249888043, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '301365', 'n_tokens_mistral': 3064, 'n_tokens_neox': 2934, 'n_words': 741}