<commit_before>// Copyright (C) 2012 Bloomberg Finance L.P.

#ifndef BUILDING_NODE_EXTENSION
#define BUILDING_NODE_EXTENSION
#endif

#include <v8.h>
#include <node.h>
#include <node_version.h>
#include <uv.h>

#include <blpapi_session.h>
#include <blpapi_eventdispatcher.h>

#include <blpapi_event.h>
#include <blpapi_message.h>
#include <blpapi_element.h>
#include <blpapi_name.h>
#include <blpapi_request.h>
#include <blpapi_subscriptionlist.h>
#include <blpapi_defs.h>

#include <deque>
#include <sstream>

#include <cmath>
#include <ctime>
#include <cstdlib>

#ifdef _WIN32
#include <time.h>
#endif

#ifndef uv_mutex_t
# ifdef _WIN32
#  define uv_mutex_t CRITICAL_SECTION
#  define uv_mutex_init(x) InitializeCriticalSection(x)
#  define uv_mutex_lock(x) EnterCriticalSection(x)
#  define uv_mutex_unlock(x) LeaveCriticalSection(x)
#  define uv_mutex_destroy(x) DeleteCriticalSection(x)
# else
#  define uv_mutex_t pthread_mutex_t
#  define uv_mutex_init(x) pthread_mutex_init(x, NULL)
#  define uv_mutex_lock(x) pthread_mutex_lock(x)
#  define uv_mutex_unlock(x) pthread_mutex_unlock(x)
#  define uv_mutex_destroy(x) pthread_mutex_destroy(x)
# endif
#endif

#define BLPAPI_EXCEPTION_TRY try {
#define BLPAPI_EXCEPTION_CATCH \
    } catch (blpapi::Exception& e) { \
        ThrowException(Exception::Error( \
                String::New(e.description().c_str(), \
                            e.description().length()))); \
    }
#define BLPAPI_EXCEPTION_CATCH_RETURN \
    } catch (blpapi::Exception& e) { \
        return ThrowException(Exception::Error( \
                String::New(e.description().c_str(), \
                            e.description().length()))); \
    }

using namespace node;
using namespace v8;

namespace BloombergLP {
namespace blpapijs {

class Session : public ObjectWrap,
                public blpapi::EventHandler {
public:
    Session(const std::string& serverHost, int serverPort,
            const std::string& authenticationOptions);
    ~Session();

    static void Initialize(Handle<Object> target);
    static Handle<Value> New(const Arguments& args);

    static Handle<Value> Start(const Arguments& args);
    static Handle<Value> Authorize(const Arguments& args);
    static Handle<Value> Stop(const Arguments& args);
    static Handle<Value> Destroy(const Arguments& args);
    static Handle<Value> OpenService(const Arguments& args);
    static Handle<Value> Subscribe(const Arguments& args);
    static Handle<Value> Resubscribe(const Arguments& args);
    static Handle<Value> Request(const Arguments& args);

private:
    Session();
    Session(const Session&);
    Session& operator=(const Session&);

    static Handle<Value> subscribe(const Arguments& args, bool resubscribe);
    static void formFields(std::string* str, Handle<Object> array);
    static void formOptions(std::string* str, Handle<Value> array);
    static Handle<Value> elementToValue(const blpapi::Element& e);
    static Handle<Value> elementValueToValue(const blpapi::Element& e,
                                             int idx = 0);

    bool processEvent(const blpapi::Event& ev, blpapi::Session* session);
    static void processEvents(uv_async_t *async, int status);
    void processMessage(blpapi::Event::EventType et,
                        const blpapi::Message& msg);

    void emit(int argc, Handle<Value> argv[]);

    static uv_async_t s_async;
    static Persistent<String> s_emit;
    static Persistent<String> s_event_type;
    static Persistent<String> s_message_type;
    static Persistent<String> s_topic_name;
    static Persistent<String> s_correlations;
    static Persistent<String> s_value;
    static Persistent<String> s_class_id;
    static Persistent<String> s_data;

    blpapi::SessionOptions d_options;
    blpapi::Session *d_session;
    blpapi::Identity d_identity;
    Persistent<Object> d_session_ref;
    std::deque<blpapi::Event> d_que;
    uv_mutex_t d_que_mutex;
    bool d_started;
    bool d_stopped;
};

uv_async_t Session::s_async;
Persistent<String> Session::s_emit;
Persistent<String> Session::s_event_type;
Persistent<String> Session::s_message_type;
Persistent<String> Session::s_topic_name;
Persistent<String> Session::s_correlations;
Persistent<String> Session::s_value;
Persistent<String> Session::s_class_id;
Persistent<String> Session::s_data;

Session::Session(const std::string& serverHost, int serverPort,
                 const std::string& authenticationOptions)
    : d_started(false)
    , d_stopped(false)
{
    d_options.setServerHost(serverHost.c_str());
    d_options.setServerPort(serverPort);
    if (authenticationOptions.length())
        d_options.setAuthenticationOptions(authenticationOptions.c_str());

    BLPAPI_EXCEPTION_TRY
    d_session = new blpapi::Session(d_options, this);
    BLPAPI_EXCEPTION_CATCH

    uv_mutex_init(&d_que_mutex);

#if NODE_VERSION_AT_LEAST(0, 7, 9)
    uv_ref(reinterpret_cast<uv_handle_t *>(&s_async));
#else
    uv_ref(uv_default_loop());
#endif
}

Session::~Session()
{
    // Ref on the event loop is released in Destroy

    uv_mutex_destroy(&d_que_mutex);
}

void
Session::Initialize(Handle<Object> target)
{
    HandleScope scope;
    Local<FunctionTemplate> t = FunctionTemplate::New(Session::New);

    t->InstanceTemplate()->SetInternalFieldCount(1);

    NODE_SET_PROTOTYPE_METHOD(t, "start", Start);
    NODE_SET_PROTOTYPE_METHOD(t, "authorize", Authorize);
    NODE_SET_PROTOTYPE_METHOD(t, "stop", Stop);
    NODE_SET_PROTOTYPE_METHOD(t, "destroy", Destroy);
    NODE_SET_PROTOTYPE_METHOD(t, "openService", OpenService);
    NODE_SET_PROTOTYPE_METHOD(t, "subscribe", Subscribe);
    NODE_SET_PROTOTYPE_METHOD(t, "resubscribe", Resubscribe);
    NODE_SET_PROTOTYPE_METHOD(t, "request", Request);

    target->Set(String::NewSymbol("Session"), t->GetFunction());

    uv_async_init(uv_default_loop(), &s_async, Session::processEvents);
#if !NODE_VERSION_AT_LEAST(0, 7, 9)
    uv_unref(uv_default_loop());
#endif

    s_emit = NODE_PSYMBOL("emit");
    s_event_type = NODE_PSYMBOL("eventType");
    s_message_type = NODE_PSYMBOL("messageType");
    s_topic_name = NODE_PSYMBOL("topicName");
    s_correlations = NODE_PSYMBOL("correlations");
    s_value = NODE_PSYMBOL("value");
    s_class_id = NODE_PSYMBOL("classId");
    s_data = NODE_PSYMBOL("data");
}

Handle<Value>
Session::New(const Arguments& args)
{
    HandleScope scope;

    std::string serverHost;
    int serverPort = 0;
    std::string authenticationOptions;

    if (args.Length() > 0 && args[0]->IsObject()) {
        Local<Object> o = args[0]->ToObject();

        // Capture the host name
        Local<Value> h = o->Get(String::New("host"));
        if (h->IsUndefined())
            h = o->Get(String::New("serverHost"));
        if (!h->IsUndefined()) {
            String::AsciiValue hv(h);
            if (hv.length())
                serverHost.assign(*hv, hv.length());
        }
        if (0 == serverHost.length())
            return ThrowException(Exception::Error(String::New(
                        "Configuration missing 'serverHost'.")));

        // Capture the port number
        Local<Value> p = o->Get(String::New("port"));
        if (p->IsUndefined())
            p = o->Get(String::New("serverPort"));
        if (p->IsInt32())
            serverPort = p->ToInt32()->Value();
        if (0 == serverPort)
            return ThrowException(Exception::Error(String::New(
                        "Configuration missing non-zero 'serverPort'.")));

        // Capture optional authentication options
        Local<Value> ao = o->Get(String::New("authenticationOptions"));
        if (!ao->IsUndefined()) {
            String::AsciiValue aov(ao);
            if (aov.length())
                authenticationOptions.assign(*aov, aov.length());
        }
    } else {
        return ThrowException(Exception::Error(String::New(
                        "Configuration object must be passed as parameter.")));
    }

    Session *session = new Session(serverHost, serverPort,
                                   authenticationOptions);
    session->Wrap(args.This());
    return scope.Close(args.This());
}

Handle<Value>
Session::Start(const Arguments& args)
{
    HandleScope scope;

    Session* session = ObjectWrap::Unwrap<Session>(args.This());

    if (session->d_started)
        return ThrowException(Exception::Error(String::New(
                        "Session has already been started.")));
    if (session->d_stopped)
        return ThrowException(Exception::Error(String::New(
                        "Stopped sessions can not be restarted.")));

    BLPAPI_EXCEPTION_TRY
    session->d_session->startAsync();
    BLPAPI_EXCEPTION_CATCH_RETURN

    session->d_session_ref = Persistent<Object>::New(args.This());
    session->d_started = true;

    return scope.Close(args.This());
}

Handle<Value>
Session::Authorize(const Arguments& args)
{
    HandleScope scope;

    if (args.Length() < 1 || !args[0]->IsString()) {
        return ThrowException(Exception::Error(String::New(
                "Service URI string must be provided as first parameter.")));
    }
    if (args.Length() < 2 || !args[1]->IsInt32()) {
        return ThrowException(Exception::Error(String::New(
                "Integer correlation identifier must be provided "
                "as second parameter.")));
    }
    if (args.Length() > 2) {<commit_msg>Update Persistent::New to pass Isolate.<commit_after>// Copyright (C) 2012 Bloomberg Finance L.P.

#ifndef BUILDING_NODE_EXTENSION
#define BUILDING_NODE_EXTENSION
#endif

#include <v8.h>
#include <node.h>
#include <node_version.h>
#include <uv.h>

#include <blpapi_session.h>
#include <blpapi_eventdispatcher.h>

#include <blpapi_event.h>
#include <blpapi_message.h>
#include <blpapi_element.h>
#include <blpapi_name.h>
#include <blpapi_request.h>
#include <blpapi_subscriptionlist.h>
#include <blpapi_defs.h>

#include <deque>
#include <sstream>

#include <cmath>
#include <ctime>
#include <cstdlib>

#ifdef _WIN32
#include <time.h>
#endif

#ifndef uv_mutex_t
# ifdef _WIN32
#  define uv_mutex_t CRITICAL_SECTION
#  define uv_mutex_init(x) InitializeCriticalSection(x)
#  define uv_mutex_lock(x) EnterCriticalSection(x)
#  define uv_mutex_unlock(x) LeaveCriticalSection(x)
#  define uv_mutex_destroy(x) DeleteCriticalSection(x)
# else
#  define uv_mutex_t pthread_mutex_t
#  define uv_mutex_init(x) pthread_mutex_init(x, NULL)
#  define uv_mutex_lock(x) pthread_mutex_lock(x)
#  define uv_mutex_unlock(x) pthread_mutex_unlock(x)
#  define uv_mutex_destroy(x) pthread_mutex_destroy(x)
# endif
#endif

#define BLPAPI_EXCEPTION_TRY try {
#define BLPAPI_EXCEPTION_CATCH \
    } catch (blpapi::Exception& e) { \
        ThrowException(Exception::Error( \
                String::New(e.description().c_str(), \
                            e.description().length()))); \
    }
#define BLPAPI_EXCEPTION_CATCH_RETURN \
    } catch (blpapi::Exception& e) { \
        return ThrowException(Exception::Error( \
                String::New(e.description().c_str(), \
                            e.description().length()))); \
    }

using namespace node;
using namespace v8;

namespace BloombergLP {
namespace blpapijs {

class Session : public ObjectWrap,
                public blpapi::EventHandler {
public:
    Session(const std::string& serverHost, int serverPort,
            const std::string& authenticationOptions);
    ~Session();

    static void Initialize(Handle<Object> target);
    static Handle<Value> New(const Arguments& args);

    static Handle<Value> Start(const Arguments& args);
    static Handle<Value> Authorize(const Arguments& args);
    static Handle<Value> Stop(const Arguments& args);
    static Handle<Value> Destroy(const Arguments& args);
    static Handle<Value> OpenService(const Arguments& args);
    static Handle<Value> Subscribe(const Arguments& args);
    static Handle<Value> Resubscribe(const Arguments& args);
    static Handle<Value> Request(const Arguments& args);

private:
    Session();
    Session(const Session&);
    Session& operator=(const Session&);

    static Handle<Value> subscribe(const Arguments& args, bool resubscribe);
    static void formFields(std::string* str, Handle<Object> array);
    static void formOptions(std::string* str, Handle<Value> array);
    static Handle<Value> elementToValue(const blpapi::Element& e);
    static Handle<Value> elementValueToValue(const blpapi::Element& e,
                                             int idx = 0);

    bool processEvent(const blpapi::Event& ev, blpapi::Session* session);
    static void processEvents(uv_async_t *async, int status);
    void processMessage(blpapi::Event::EventType et,
                        const blpapi::Message& msg);

    void emit(int argc, Handle<Value> argv[]);

    static uv_async_t s_async;
    static Persistent<String> s_emit;
    static Persistent<String> s_event_type;
    static Persistent<String> s_message_type;
    static Persistent<String> s_topic_name;
    static Persistent<String> s_correlations;
    static Persistent<String> s_value;
    static Persistent<String> s_class_id;
    static Persistent<String> s_data;

    blpapi::SessionOptions d_options;
    blpapi::Session *d_session;
    blpapi::Identity d_identity;
    Persistent<Object> d_session_ref;
    std::deque<blpapi::Event> d_que;
    uv_mutex_t d_que_mutex;
    bool d_started;
    bool d_stopped;
};

uv_async_t Session::s_async;
Persistent<String> Session::s_emit;
Persistent<String> Session::s_event_type;
Persistent<String> Session::s_message_type;
Persistent<String> Session::s_topic_name;
Persistent<String> Session::s_correlations;
Persistent<String> Session::s_value;
Persistent<String> Session::s_class_id;
Persistent<String> Session::s_data;

Session::Session(const std::string& serverHost, int serverPort,
                 const std::string& authenticationOptions)
    : d_started(false)
    , d_stopped(false)
{
    d_options.setServerHost(serverHost.c_str());
    d_options.setServerPort(serverPort);
    if (authenticationOptions.length())
        d_options.setAuthenticationOptions(authenticationOptions.c_str());

    BLPAPI_EXCEPTION_TRY
    d_session = new blpapi::Session(d_options, this);
    BLPAPI_EXCEPTION_CATCH

    uv_mutex_init(&d_que_mutex);

#if NODE_VERSION_AT_LEAST(0, 7, 9)
    uv_ref(reinterpret_cast<uv_handle_t *>(&s_async));
#else
    uv_ref(uv_default_loop());
#endif
}

Session::~Session()
{
    // Ref on the event loop is released in Destroy

    uv_mutex_destroy(&d_que_mutex);
}

void
Session::Initialize(Handle<Object> target)
{
    HandleScope scope;
    Local<FunctionTemplate> t = FunctionTemplate::New(Session::New);

    t->InstanceTemplate()->SetInternalFieldCount(1);

    NODE_SET_PROTOTYPE_METHOD(t, "start", Start);
    NODE_SET_PROTOTYPE_METHOD(t, "authorize", Authorize);
    NODE_SET_PROTOTYPE_METHOD(t, "stop", Stop);
    NODE_SET_PROTOTYPE_METHOD(t, "destroy", Destroy);
    NODE_SET_PROTOTYPE_METHOD(t, "openService", OpenService);
    NODE_SET_PROTOTYPE_METHOD(t, "subscribe", Subscribe);
    NODE_SET_PROTOTYPE_METHOD(t, "resubscribe", Resubscribe);
    NODE_SET_PROTOTYPE_METHOD(t, "request", Request);

    target->Set(String::NewSymbol("Session"), t->GetFunction());

    uv_async_init(uv_default_loop(), &s_async, Session::processEvents);
#if !NODE_VERSION_AT_LEAST(0, 7, 9)
    uv_unref(uv_default_loop());
#endif

    s_emit = NODE_PSYMBOL("emit");
    s_event_type = NODE_PSYMBOL("eventType");
    s_message_type = NODE_PSYMBOL("messageType");
    s_topic_name = NODE_PSYMBOL("topicName");
    s_correlations = NODE_PSYMBOL("correlations");
    s_value = NODE_PSYMBOL("value");
    s_class_id = NODE_PSYMBOL("classId");
    s_data = NODE_PSYMBOL("data");
}

Handle<Value>
Session::New(const Arguments& args)
{
    HandleScope scope;

    std::string serverHost;
    int serverPort = 0;
    std::string authenticationOptions;

    if (args.Length() > 0 && args[0]->IsObject()) {
        Local<Object> o = args[0]->ToObject();

        // Capture the host name
        Local<Value> h = o->Get(String::New("host"));
        if (h->IsUndefined())
            h = o->Get(String::New("serverHost"));
        if (!h->IsUndefined()) {
            String::AsciiValue hv(h);
            if (hv.length())
                serverHost.assign(*hv, hv.length());
        }
        if (0 == serverHost.length())
            return ThrowException(Exception::Error(String::New(
                        "Configuration missing 'serverHost'.")));

        // Capture the port number
        Local<Value> p = o->Get(String::New("port"));
        if (p->IsUndefined())
            p = o->Get(String::New("serverPort"));
        if (p->IsInt32())
            serverPort = p->ToInt32()->Value();
        if (0 == serverPort)
            return ThrowException(Exception::Error(String::New(
                        "Configuration missing non-zero 'serverPort'.")));

        // Capture optional authentication options
        Local<Value> ao = o->Get(String::New("authenticationOptions"));
        if (!ao->IsUndefined()) {
            String::AsciiValue aov(ao);
            if (aov.length())
                authenticationOptions.assign(*aov, aov.length());
        }
    } else {
        return ThrowException(Exception::Error(String::New(
                        "Configuration object must be passed as parameter.")));
    }

    Session *session = new Session(serverHost, serverPort,
                                   authenticationOptions);
    session->Wrap(args.This());
    return scope.Close(args.This());
}

Handle<Value>
Session::Start(const Arguments& args)
{
    HandleScope scope;

    Session* session = ObjectWrap::Unwrap<Session>(args.This());

    if (session->d_started)
        return ThrowException(Exception::Error(String::New(
                        "Session has already been started.")));
    if (session->d_stopped)
        return ThrowException(Exception::Error(String::New(
                        "Stopped sessions can not be restarted.")));

    BLPAPI_EXCEPTION_TRY
    session->d_session->startAsync();
    BLPAPI_EXCEPTION_CATCH_RETURN

    session->d_session_ref =
#if NODE_VERSION_AT_LEAST(0, 11, 0)
        Persistent<Object>::New(Isolate::GetCurrent(), args.This());
#else
        Persistent<Object>::New(args.This());
#endif
    session->d_started = true;

    return scope.Close(args.This());
}

Handle<Value>
Session::Authorize(const Arguments& args)
{
    HandleScope scope;

    if (args.Length() < 1 || !args[0]->IsString()) {
        return ThrowException(Exception::Error(String::New(
                "Service URI string must be provided as first parameter.")));
    }
    if (args.Length() < 2 || !args[1]->IsInt32()) {
        return ThrowException(Exception::Error(String::New(
                "Integer correlation identifier must be provided "
                "as second parameter.")));
    }
    if (args.Length() > 2) {
{'fraction_non_alphanumeric': 0.12706392887383572, 'fraction_numerical': 0.003916172734970364, 'mean_word_length': 3.6624722427831236, 'pattern_counts': {'":': 0, '<': 138, '<?xml version=': 0, '>': 192, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3652626', 'n_tokens_mistral': 6124, 'n_tokens_neox': 5872, 'n_words': 1196}