Cory McWilliams
937e32fb99
git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@3357 ed5197a5-7fde-0310-b194-c3ffbd925b24
678 lines
26 KiB
C++
678 lines
26 KiB
C++
#include "Socket.h"
|
|
|
|
#include "Task.h"
|
|
#include "TaskTryCatch.h"
|
|
#include "Tls.h"
|
|
#include "TlsContextWrapper.h"
|
|
|
|
#include <assert.h>
|
|
#include <cstring>
|
|
#include <uv.h>
|
|
|
|
int Socket::_count = 0;
|
|
int Socket::_openCount = 0;
|
|
TlsContext* Socket::_defaultTlsContext = 0;
|
|
|
|
struct SocketResolveData {
|
|
uv_getaddrinfo_t resolver;
|
|
Socket* socket;
|
|
promiseid_t promise;
|
|
};
|
|
|
|
Socket::Socket(Task* task) {
|
|
v8::HandleScope scope(task->getIsolate());
|
|
++_count;
|
|
|
|
v8::Handle<v8::External> data = v8::External::New(task->getIsolate(), this);
|
|
|
|
v8::Local<v8::ObjectTemplate> socketTemplate = v8::ObjectTemplate::New(task->getIsolate());
|
|
socketTemplate->SetInternalFieldCount(1);
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "bind"), v8::FunctionTemplate::New(task->getIsolate(), bind, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "connect"), v8::FunctionTemplate::New(task->getIsolate(), connect, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "listen"), v8::FunctionTemplate::New(task->getIsolate(), listen, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "accept"), v8::FunctionTemplate::New(task->getIsolate(), accept, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "startTls"), v8::FunctionTemplate::New(task->getIsolate(), startTls, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "stopTls"), v8::FunctionTemplate::New(task->getIsolate(), stopTls, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "shutdown"), v8::FunctionTemplate::New(task->getIsolate(), shutdown, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "close"), v8::FunctionTemplate::New(task->getIsolate(), close, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "read"), v8::FunctionTemplate::New(task->getIsolate(), read, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "onError"), v8::FunctionTemplate::New(task->getIsolate(), onError, data));
|
|
socketTemplate->Set(v8::String::NewFromUtf8(task->getIsolate(), "write"), v8::FunctionTemplate::New(task->getIsolate(), write, data));
|
|
socketTemplate->SetAccessor(v8::String::NewFromUtf8(task->getIsolate(), "peerName"), getPeerName, 0, data);
|
|
socketTemplate->SetAccessor(v8::String::NewFromUtf8(task->getIsolate(), "peerCertificate"), getPeerCertificate, 0, data);
|
|
socketTemplate->SetAccessor(v8::String::NewFromUtf8(task->getIsolate(), "isConnected"), isConnected, 0, data);
|
|
|
|
v8::Local<v8::Object> socketObject = socketTemplate->NewInstance();
|
|
socketObject->SetInternalField(0, v8::External::New(task->getIsolate(), this));
|
|
_object = v8::Persistent<v8::Object, v8::CopyablePersistentTraits<v8::Object> >(task->getIsolate(), socketObject);
|
|
|
|
uv_tcp_init(task->getLoop(), &_socket);
|
|
++_openCount;
|
|
_socket.data = this;
|
|
_task = task;
|
|
}
|
|
|
|
Socket::~Socket() {
|
|
if (_tls) {
|
|
delete _tls;
|
|
_tls = 0;
|
|
}
|
|
--_count;
|
|
}
|
|
|
|
void Socket::close() {
|
|
if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(&_socket))) {
|
|
if (!_onRead.IsEmpty()) {
|
|
_onRead.Reset();
|
|
}
|
|
uv_close(reinterpret_cast<uv_handle_t*>(&_socket), onClose);
|
|
}
|
|
}
|
|
|
|
void Socket::reportError(const char* error) {
|
|
v8::Handle<v8::Value> exception = v8::Exception::Error(v8::String::NewFromUtf8(_task->getIsolate(), error));
|
|
if (!_onError.IsEmpty()) {
|
|
v8::Handle<v8::Function> callback = v8::Local<v8::Function>::New(_task->getIsolate(), _onError);
|
|
callback->Call(callback, 1, &exception);
|
|
} else {
|
|
_task->getIsolate()->ThrowException(exception);
|
|
}
|
|
}
|
|
|
|
void Socket::reportTlsErrors() {
|
|
char buffer[4096];
|
|
while (_tls && _tls->getError(buffer, sizeof(buffer))) {
|
|
reportError(buffer);
|
|
}
|
|
}
|
|
|
|
void Socket::startTls(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
if (!socket->_tls) {
|
|
TlsContext* context = 0;
|
|
|
|
if (args.Length() > 0 && !args[0].IsEmpty() && args[0]->IsObject()) {
|
|
if (TlsContextWrapper* wrapper = TlsContextWrapper::get(args[0])) {
|
|
context = wrapper->getContext();
|
|
}
|
|
} else {
|
|
if (!_defaultTlsContext) {
|
|
_defaultTlsContext = TlsContext::create();
|
|
}
|
|
context = _defaultTlsContext;
|
|
}
|
|
|
|
if (context) {
|
|
socket->_tls = context->createSession();
|
|
}
|
|
|
|
if (socket->_tls) {
|
|
socket->_tls->setHostname(socket->_peerName.c_str());
|
|
if (socket->_direction == kAccept) {
|
|
socket->_tls->startAccept();
|
|
} else if (socket->_direction == kConnect) {
|
|
socket->_tls->startConnect();
|
|
}
|
|
socket->_startTlsPromise = socket->_task->allocatePromise();
|
|
socket->processOutgoingTls();
|
|
args.GetReturnValue().Set(socket->_task->getPromise(socket->_startTlsPromise));
|
|
} else {
|
|
args.GetIsolate()->ThrowException(v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), "Failed to get TLS context")));
|
|
}
|
|
} else {
|
|
args.GetIsolate()->ThrowException(v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), "startTls with TLS already started")));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::stopTls(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
if (socket->_tls) {
|
|
socket->processOutgoingTls();
|
|
delete socket->_tls;
|
|
socket->_tls = 0;
|
|
} else {
|
|
args.GetIsolate()->ThrowException(v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), "stopTls with TLS already stopped")));
|
|
}
|
|
}
|
|
}
|
|
|
|
bool Socket::processSomeOutgoingTls(promiseid_t promise, uv_write_cb callback) {
|
|
char buffer[8192];
|
|
int result = _tls->readEncrypted(buffer, sizeof(buffer));
|
|
if (result > 0) {
|
|
char* rawBuffer = new char[sizeof(uv_write_t) + result];
|
|
uv_write_t* request = reinterpret_cast<uv_write_t*>(rawBuffer);
|
|
std::memset(request, 0, sizeof(*request));
|
|
request->data = reinterpret_cast<void*>(promise);
|
|
rawBuffer += sizeof(uv_write_t);
|
|
std::memcpy(rawBuffer, buffer, result);
|
|
|
|
uv_buf_t writeBuffer;
|
|
writeBuffer.base = rawBuffer;
|
|
writeBuffer.len = result;
|
|
|
|
int writeResult = uv_write(request, reinterpret_cast<uv_stream_t*>(&_socket), &writeBuffer, 1, callback);
|
|
if (writeResult != 0) {
|
|
std::string error = "uv_write: " + std::string(uv_strerror(writeResult));
|
|
reportError(error.c_str());
|
|
}
|
|
} else {
|
|
reportTlsErrors();
|
|
}
|
|
return result > 0;
|
|
}
|
|
|
|
void Socket::processOutgoingTls() {
|
|
while (processSomeOutgoingTls(-1, onWrite)) {}
|
|
}
|
|
|
|
void Socket::bind(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
v8::String::Utf8Value node(args[0]->ToString());
|
|
v8::String::Utf8Value port(args[1]->ToString());
|
|
|
|
SocketResolveData* data = new SocketResolveData();
|
|
std::memset(data, 0, sizeof(*data));
|
|
struct addrinfo hints;
|
|
hints.ai_family = PF_INET;
|
|
hints.ai_socktype = SOCK_STREAM;
|
|
hints.ai_protocol = IPPROTO_TCP;
|
|
hints.ai_flags = 0;
|
|
data->resolver.data = data;
|
|
data->socket = socket;
|
|
data->promise = socket->_task->allocatePromise();
|
|
|
|
int result = uv_getaddrinfo(socket->_task->getLoop(), &data->resolver, onResolvedForBind, *node, *port, &hints);
|
|
if (result != 0) {
|
|
std::string error = "uv_getaddrinfo: " + std::string(uv_strerror(result));
|
|
socket->_task->rejectPromise(data->promise, v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), error.c_str())));
|
|
delete data;
|
|
}
|
|
|
|
args.GetReturnValue().Set(socket->_task->getPromise(data->promise));
|
|
}
|
|
}
|
|
|
|
void Socket::onResolvedForBind(uv_getaddrinfo_t* resolver, int status, struct addrinfo* result) {
|
|
SocketResolveData* data = reinterpret_cast<SocketResolveData*>(resolver->data);
|
|
if (status != 0) {
|
|
std::string error = "uv_getaddrinfo: " + std::string(uv_strerror(status));
|
|
data->socket->_task->rejectPromise(data->promise, v8::Exception::Error(v8::String::NewFromUtf8(data->socket->_task->getIsolate(), error.c_str())));
|
|
} else {
|
|
int bindResult = uv_tcp_bind(&data->socket->_socket, result->ai_addr, 0);
|
|
if (bindResult != 0) {
|
|
std::string error = "uv_tcp_bind: " + std::string(uv_strerror(bindResult));
|
|
data->socket->_task->rejectPromise(data->promise, v8::Exception::Error(v8::String::NewFromUtf8(data->socket->_task->getIsolate(), error.c_str())));
|
|
} else {
|
|
data->socket->_task->resolvePromise(data->promise, v8::Undefined(data->socket->_task->getIsolate()));
|
|
}
|
|
}
|
|
delete data;
|
|
}
|
|
|
|
void Socket::connect(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
socket->_direction = kConnect;
|
|
v8::String::Utf8Value node(args[0]->ToString());
|
|
v8::String::Utf8Value port(args[1]->ToString());
|
|
|
|
socket->_peerName = *node;
|
|
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
|
|
SocketResolveData* data = new SocketResolveData();
|
|
std::memset(data, 0, sizeof(*data));
|
|
struct addrinfo hints;
|
|
hints.ai_family = PF_INET;
|
|
hints.ai_socktype = SOCK_STREAM;
|
|
hints.ai_protocol = IPPROTO_TCP;
|
|
hints.ai_flags = 0;
|
|
data->resolver.data = data;
|
|
data->socket = socket;
|
|
data->promise = promise;
|
|
|
|
int result = uv_getaddrinfo(socket->_task->getLoop(), &data->resolver, onResolvedForConnect, *node, *port, &hints);
|
|
if (result != 0) {
|
|
std::string error = "uv_getaddrinfo: " + std::string(uv_strerror(result));
|
|
socket->_task->rejectPromise(promise, v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), error.c_str())));
|
|
delete data;
|
|
}
|
|
|
|
args.GetReturnValue().Set(socket->_task->getPromise(promise));
|
|
}
|
|
}
|
|
|
|
void Socket::onResolvedForConnect(uv_getaddrinfo_t* resolver, int status, struct addrinfo* result) {
|
|
SocketResolveData* data = reinterpret_cast<SocketResolveData*>(resolver->data);
|
|
if (status != 0) {
|
|
std::string error = "uv_getaddrinfo: " + std::string(uv_strerror(status));
|
|
data->socket->_task->rejectPromise(data->promise, v8::Exception::Error(v8::String::NewFromUtf8(data->socket->_task->getIsolate(), error.c_str())));
|
|
} else {
|
|
uv_connect_t* request = new uv_connect_t();
|
|
std::memset(request, 0, sizeof(*request));
|
|
request->data = reinterpret_cast<void*>(data->promise);
|
|
int connectResult = uv_tcp_connect(request, &data->socket->_socket, result->ai_addr, onConnect);
|
|
if (connectResult != 0) {
|
|
std::string error("uv_tcp_connect: " + std::string(uv_strerror(connectResult)));
|
|
data->socket->_task->rejectPromise(data->promise, v8::Exception::Error(v8::String::NewFromUtf8(data->socket->_task->getIsolate(), error.c_str())));
|
|
}
|
|
}
|
|
delete data;
|
|
}
|
|
|
|
|
|
void Socket::onConnect(uv_connect_t* request, int status) {
|
|
promiseid_t promise = reinterpret_cast<intptr_t>(request->data);
|
|
if (promise != -1) {
|
|
Socket* socket = reinterpret_cast<Socket*>(request->handle->data);
|
|
if (status == 0) {
|
|
socket->_connected = true;
|
|
socket->_task->resolvePromise(promise, v8::Integer::New(socket->_task->getIsolate(), status));
|
|
} else {
|
|
std::string error("uv_tcp_connect: " + std::string(uv_strerror(status)));
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(socket->_task->getIsolate(), error.c_str()));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::listen(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
int backlog = args[0]->ToInteger()->Value();
|
|
if (socket->_onConnect.IsEmpty()) {
|
|
v8::Persistent<v8::Function, v8::CopyablePersistentTraits<v8::Function> > callback(args.GetIsolate(), args[1].As<v8::Function>());
|
|
socket->_onConnect = callback;
|
|
int result = uv_listen(reinterpret_cast<uv_stream_t*>(&socket->_socket), backlog, onNewConnection);
|
|
if (result != 0) {
|
|
std::string error = "uv_listen: " + std::string(uv_strerror(result));
|
|
args.GetIsolate()->ThrowException(v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), error.c_str(), v8::String::kNormalString, error.size())));
|
|
}
|
|
args.GetReturnValue().Set(v8::Integer::New(args.GetIsolate(), result));
|
|
} else {
|
|
args.GetIsolate()->ThrowException(v8::Exception::Error(v8::String::NewFromUtf8(args.GetIsolate(), "listen: Already listening.")));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::onNewConnection(uv_stream_t* server, int status) {
|
|
if (Socket* socket = reinterpret_cast<Socket*>(server->data)) {
|
|
v8::HandleScope handleScope(socket->_task->getIsolate());
|
|
TaskTryCatch tryCatch(socket->_task);
|
|
if (!socket->_onConnect.IsEmpty()) {
|
|
v8::Handle<v8::Function> callback = v8::Local<v8::Function>::New(socket->_task->getIsolate(), socket->_onConnect);
|
|
callback->Call(callback, 0, 0);
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::accept(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
v8::HandleScope handleScope(args.GetIsolate());
|
|
Socket* client = new Socket(socket->_task);
|
|
client->_direction = kAccept;
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
v8::Handle<v8::Value> promiseObject = socket->_task->getPromise(promise);
|
|
v8::Handle<v8::Object> result = v8::Local<v8::Object>::New(args.GetIsolate(), client->_object);
|
|
int status = uv_accept(reinterpret_cast<uv_stream_t*>(&socket->_socket), reinterpret_cast<uv_stream_t*>(&client->_socket));
|
|
if (status == 0) {
|
|
client->_connected = true;
|
|
socket->_task->resolvePromise(promise, result);
|
|
} else {
|
|
std::string error = "uv_accept: " + std::string(uv_strerror(status));
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(args.GetIsolate(), error.c_str(), v8::String::kNormalString, error.size()));
|
|
}
|
|
args.GetReturnValue().Set(promiseObject);
|
|
client->release();
|
|
}
|
|
}
|
|
|
|
void Socket::close(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
if (socket->_closePromise == -1) {
|
|
socket->_closePromise = socket->_task->allocatePromise();
|
|
args.GetReturnValue().Set(socket->_task->getPromise(socket->_closePromise));
|
|
socket->close();
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::shutdown(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
if (socket->_tls) {
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
socket->processTlsShutdown(promise);
|
|
args.GetReturnValue().Set(socket->_task->getPromise(promise));
|
|
} else {
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
socket->shutdownInternal(promise);
|
|
args.GetReturnValue().Set(socket->_task->getPromise(promise));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::shutdownInternal(promiseid_t promise) {
|
|
uv_shutdown_t* request = new uv_shutdown_t();
|
|
std::memset(request, 0, sizeof(*request));
|
|
request->data = reinterpret_cast<void*>(promise);
|
|
int result = uv_shutdown(request, reinterpret_cast<uv_stream_t*>(&_socket), onShutdown);
|
|
if (result != 0) {
|
|
std::string error = "uv_shutdown: " + std::string(uv_strerror(result));
|
|
_task->rejectPromise(promise, v8::Exception::Error(v8::String::NewFromUtf8(_task->getIsolate(), error.c_str())));
|
|
delete request;
|
|
}
|
|
}
|
|
|
|
void Socket::processTlsShutdown(promiseid_t promise) {
|
|
_tls->shutdown();
|
|
if (!processSomeOutgoingTls(promise, onTlsShutdown)) {
|
|
shutdownInternal(promise);
|
|
}
|
|
}
|
|
|
|
void Socket::onTlsShutdown(uv_write_t* request, int status) {
|
|
if (Socket* socket = reinterpret_cast<Socket*>(request->handle->data)) {
|
|
promiseid_t promise = reinterpret_cast<intptr_t>(request->data);
|
|
socket->processTlsShutdown(promise);
|
|
}
|
|
}
|
|
|
|
void Socket::onError(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
v8::Persistent<v8::Function, v8::CopyablePersistentTraits<v8::Function> > callback(args.GetIsolate(), args[0].As<v8::Function>());
|
|
socket->_onError = callback;
|
|
}
|
|
}
|
|
|
|
void Socket::read(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
v8::Persistent<v8::Function, v8::CopyablePersistentTraits<v8::Function> > callback(args.GetIsolate(), args[0].As<v8::Function>());
|
|
socket->_onRead = callback;
|
|
int result = uv_read_start(reinterpret_cast<uv_stream_t*>(&socket->_socket), allocateBuffer, onRead);
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
if (result != 0) {
|
|
std::string error = "uv_read_start: " + std::string(uv_strerror(result));
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(socket->_task->getIsolate(), error.c_str(), v8::String::kNormalString, error.size()));
|
|
} else {
|
|
socket->_task->resolvePromise(promise, v8::Undefined(socket->_task->getIsolate()));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::allocateBuffer(uv_handle_t* handle, size_t suggestedSize, uv_buf_t* buf) {
|
|
*buf = uv_buf_init(new char[suggestedSize], suggestedSize);
|
|
}
|
|
|
|
void Socket::onRead(uv_stream_t* stream, ssize_t readSize, const uv_buf_t* buffer) {
|
|
if (Socket* socket = reinterpret_cast<Socket*>(stream->data)) {
|
|
v8::HandleScope handleScope(socket->_task->getIsolate());
|
|
TaskTryCatch tryCatch(socket->_task);
|
|
v8::Handle<v8::Value> data;
|
|
|
|
if (readSize <= 0) {
|
|
socket->_connected = false;
|
|
v8::Local<v8::Function> callback = v8::Local<v8::Function>::New(socket->_task->getIsolate(), socket->_onRead);
|
|
if (!callback.IsEmpty()) {
|
|
data = v8::Undefined(socket->_task->getIsolate());
|
|
callback->Call(callback, 1, &data);
|
|
}
|
|
socket->close();
|
|
} else {
|
|
if (socket->_tls) {
|
|
socket->reportTlsErrors();
|
|
socket->_tls->writeEncrypted(buffer->base, readSize);
|
|
if (socket->_startTlsPromise != -1) {
|
|
TlsSession::HandshakeResult result = socket->_tls->handshake();
|
|
if (result == TlsSession::kDone) {
|
|
promiseid_t promise = socket->_startTlsPromise;
|
|
socket->_startTlsPromise = -1;
|
|
socket->_task->resolvePromise(promise, v8::Undefined(socket->_task->getIsolate()));
|
|
} else if (result == TlsSession::kFailed) {
|
|
promiseid_t promise = socket->_startTlsPromise;
|
|
socket->_startTlsPromise = -1;
|
|
char buffer[8192];
|
|
if (socket->_tls->getError(buffer, sizeof(buffer))) {
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(socket->_task->getIsolate(), buffer));
|
|
} else {
|
|
socket->_task->rejectPromise(promise, v8::Undefined(socket->_task->getIsolate()));
|
|
}
|
|
}
|
|
}
|
|
|
|
while (true) {
|
|
char plain[8192];
|
|
int result = socket->_tls->readPlain(plain, sizeof(plain));
|
|
if (result > 0) {
|
|
socket->notifyDataRead(plain, result);
|
|
} else if (result == TlsSession::kReadFailed) {
|
|
socket->reportTlsErrors();
|
|
socket->close();
|
|
break;
|
|
} else if (result == TlsSession::kReadZero) {
|
|
v8::Local<v8::Function> callback = v8::Local<v8::Function>::New(socket->_task->getIsolate(), socket->_onRead);
|
|
if (!callback.IsEmpty()) {
|
|
data = v8::Undefined(socket->_task->getIsolate());
|
|
callback->Call(callback, 1, &data);
|
|
}
|
|
break;
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if (socket->_tls) {
|
|
socket->processOutgoingTls();
|
|
}
|
|
} else {
|
|
socket->notifyDataRead(buffer->base, readSize);
|
|
}
|
|
}
|
|
}
|
|
delete[] buffer->base;
|
|
}
|
|
|
|
void Socket::notifyDataRead(const char* data, size_t length) {
|
|
v8::Local<v8::Function> callback = v8::Local<v8::Function>::New(_task->getIsolate(), _onRead);
|
|
if (!callback.IsEmpty()) {
|
|
if (data && length > 0) {
|
|
v8::Handle<v8::ArrayBuffer> buffer = v8::ArrayBuffer::New(_task->getIsolate(), length);
|
|
std::memcpy(buffer->GetContents().Data(), data, length);
|
|
v8::Handle<v8::Uint8Array> array = v8::Uint8Array::New(buffer, 0, length);
|
|
|
|
v8::Handle<v8::Value> arguments = array;
|
|
callback->Call(callback, 1, &arguments);
|
|
}
|
|
}
|
|
}
|
|
|
|
int Socket::writeBytes(std::function<int(const char*, size_t)> callback, v8::Handle<v8::Value> value, int* outLength) {
|
|
int result = -1;
|
|
|
|
if (value->IsArrayBufferView()) {
|
|
v8::Handle<v8::ArrayBufferView> array = v8::Handle<v8::ArrayBufferView>::Cast(value);
|
|
result = callback(reinterpret_cast<const char*>(array->Buffer()->GetContents().Data()), array->Buffer()->GetContents().ByteLength());
|
|
if (outLength) {
|
|
*outLength = array->Buffer()->GetContents().ByteLength();
|
|
}
|
|
} else if (value->IsString()) {
|
|
v8::Handle<v8::String> stringValue = v8::Handle<v8::String>::Cast(value);
|
|
if (stringValue->ContainsOnlyOneByte()) {
|
|
std::vector<uint8_t> bytes(stringValue->Length());
|
|
stringValue->WriteOneByte(bytes.data(), 0, bytes.size(), v8::String::NO_NULL_TERMINATION);
|
|
result = callback(reinterpret_cast<const char*>(bytes.data()), bytes.size());
|
|
if (outLength) {
|
|
*outLength = stringValue->Length();
|
|
}
|
|
} else {
|
|
v8::String::Utf8Value utf8(stringValue);
|
|
result = callback(*utf8, utf8.length());
|
|
if (outLength) {
|
|
*outLength = utf8.length();
|
|
}
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
int Socket::writeInternal(promiseid_t promise, const char* data, size_t length) {
|
|
char* rawBuffer = new char[sizeof(uv_write_t) + length];
|
|
uv_write_t* request = reinterpret_cast<uv_write_t*>(rawBuffer);
|
|
std::memcpy(rawBuffer + sizeof(uv_write_t), data, length);
|
|
|
|
uv_buf_t buffer;
|
|
buffer.base = rawBuffer + sizeof(uv_write_t);
|
|
buffer.len = length;
|
|
|
|
request->data = reinterpret_cast<void*>(promise);
|
|
return uv_write(request, reinterpret_cast<uv_stream_t*>(&_socket), &buffer, 1, onWrite);
|
|
}
|
|
|
|
void Socket::write(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
if (Socket* socket = Socket::get(args.Data())) {
|
|
promiseid_t promise = socket->_task->allocatePromise();
|
|
args.GetReturnValue().Set(socket->_task->getPromise(promise));
|
|
v8::Handle<v8::Value> value = args[0];
|
|
if (!value.IsEmpty()) {
|
|
if (socket->_tls) {
|
|
socket->reportTlsErrors();
|
|
int length = 0;
|
|
std::function<int(const char*, size_t)> writeFunction = std::bind(&TlsSession::writePlain, socket->_tls, std::placeholders::_1, std::placeholders::_2);
|
|
int result = socket->writeBytes(writeFunction, value, &length);
|
|
char buffer[8192];
|
|
if (result <= 0 && socket->_tls->getError(buffer, sizeof(buffer))) {
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(args.GetIsolate(), buffer));
|
|
} else if (result < length) {
|
|
socket->_task->rejectPromise(promise, v8::Integer::New(socket->_task->getIsolate(), result));
|
|
} else {
|
|
socket->_task->resolvePromise(promise, v8::Integer::New(socket->_task->getIsolate(), result));
|
|
}
|
|
socket->processOutgoingTls();
|
|
} else {
|
|
int length;
|
|
|
|
std::function<int(const char*, size_t)> writeFunction = std::bind(&Socket::writeInternal, socket, promise, std::placeholders::_1, std::placeholders::_2);
|
|
int result = socket->writeBytes(writeFunction, value, &length);
|
|
|
|
if (result != 0) {
|
|
std::string error = "uv_write: " + std::string(uv_strerror(result));
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(args.GetIsolate(), error.c_str(), v8::String::kNormalString, error.size()));
|
|
}
|
|
}
|
|
} else {
|
|
socket->_task->rejectPromise(promise, v8::Integer::New(args.GetIsolate(), -2));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::onWrite(uv_write_t* request, int status) {
|
|
if (Socket* socket = reinterpret_cast<Socket*>(request->handle->data)) {
|
|
v8::HandleScope handleScope(socket->_task->getIsolate());
|
|
promiseid_t promise = reinterpret_cast<intptr_t>(request->data);
|
|
if (promise != -1) {
|
|
if (status == 0) {
|
|
socket->_task->resolvePromise(promise, v8::Integer::New(socket->_task->getIsolate(), status));
|
|
} else {
|
|
std::string error = "uv_write: " + std::string(uv_strerror(status));
|
|
socket->_task->rejectPromise(promise, v8::String::NewFromUtf8(socket->_task->getIsolate(), error.c_str(), v8::String::kNormalString, error.size()));
|
|
}
|
|
}
|
|
}
|
|
delete[] reinterpret_cast<char*>(request);
|
|
}
|
|
|
|
void Socket::onClose(uv_handle_t* handle) {
|
|
--_openCount;
|
|
if (Socket* socket = reinterpret_cast<Socket*>(handle->data)) {
|
|
if (socket->_closePromise != -1) {
|
|
v8::HandleScope scope(socket->_task->getIsolate());
|
|
promiseid_t promise = socket->_closePromise;
|
|
socket->_closePromise = -1;
|
|
socket->_connected = false;
|
|
socket->_task->resolvePromise(promise, v8::Integer::New(socket->_task->getIsolate(), 0));
|
|
}
|
|
if (socket->_object.IsEmpty()) {
|
|
delete socket;
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::onShutdown(uv_shutdown_t* request, int status) {
|
|
if (Socket* socket = reinterpret_cast<Socket*>(request->handle->data)) {
|
|
promiseid_t promise = reinterpret_cast<intptr_t>(request->data);
|
|
if (status == 0) {
|
|
socket->_task->resolvePromise(promise, v8::Undefined(socket->_task->getIsolate()));
|
|
} else {
|
|
std::string error = "uv_shutdown: " + std::string(uv_strerror(status));
|
|
socket->_task->rejectPromise(promise, v8::Exception::Error(v8::String::NewFromUtf8(socket->_task->getIsolate(), error.c_str())));
|
|
}
|
|
}
|
|
delete request;
|
|
}
|
|
|
|
void Socket::getPeerName(v8::Local<v8::String> property, const v8::PropertyCallbackInfo<v8::Value>& info) {
|
|
if (Socket* socket = Socket::get(info.Data())) {
|
|
struct sockaddr_in6 addr;
|
|
int nameLength = sizeof(addr);
|
|
if (uv_tcp_getpeername(&socket->_socket, reinterpret_cast<sockaddr*>(&addr), &nameLength) == 0) {
|
|
char name[1024];
|
|
if (static_cast<size_t>(nameLength) > sizeof(struct sockaddr_in)) {
|
|
if (uv_ip6_name(&addr, name, sizeof(name)) == 0) {
|
|
info.GetReturnValue().Set(v8::String::NewFromUtf8(info.GetIsolate(), name));
|
|
}
|
|
} else {
|
|
if (uv_ip4_name(reinterpret_cast<struct sockaddr_in*>(&addr), name, sizeof(name)) == 0) {
|
|
info.GetReturnValue().Set(v8::String::NewFromUtf8(info.GetIsolate(), name));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::getPeerCertificate(v8::Local<v8::String> property, const v8::PropertyCallbackInfo<v8::Value>& info) {
|
|
if (Socket* socket = Socket::get(info.Data())) {
|
|
if (socket->_tls) {
|
|
std::vector<char> buffer(128 * 1024);
|
|
int result = socket->_tls->getPeerCertificate(buffer.data(), buffer.size());
|
|
if (result > 0) {
|
|
info.GetReturnValue().Set(v8::String::NewFromUtf8(info.GetIsolate(), buffer.data(), v8::String::kNormalString, result));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Socket::isConnected(v8::Local<v8::String> property, const v8::PropertyCallbackInfo<v8::Value>& info) {
|
|
if (Socket* socket = Socket::get(info.Data())) {
|
|
info.GetReturnValue().Set(v8::Boolean::New(socket->_task->getIsolate(), socket->_connected));
|
|
}
|
|
}
|
|
|
|
void Socket::create(const v8::FunctionCallbackInfo<v8::Value>& args) {
|
|
v8::HandleScope handleScope(args.GetIsolate());
|
|
if (Socket* socket = new Socket(Task::get(args.GetIsolate()))) {
|
|
v8::Handle<v8::Object> result = v8::Local<v8::Object>::New(args.GetIsolate(), socket->_object);
|
|
args.GetReturnValue().Set(result);
|
|
socket->release();
|
|
}
|
|
}
|
|
|
|
Socket* Socket::get(v8::Handle<v8::Value> socketObject) {
|
|
return reinterpret_cast<Socket*>(v8::Handle<v8::External>::Cast(socketObject)->Value());
|
|
}
|
|
|
|
void Socket::ref() {
|
|
if (++_refCount == 1) {
|
|
_object.ClearWeak();
|
|
}
|
|
}
|
|
|
|
void Socket::release() {
|
|
assert(_refCount >= 1);
|
|
if (--_refCount == 0) {
|
|
_object.SetWeak(this, onRelease, v8::WeakCallbackType::kParameter);
|
|
}
|
|
}
|
|
|
|
void Socket::onRelease(const v8::WeakCallbackInfo<Socket>& data) {
|
|
data.GetParameter()->_object.Reset();
|
|
data.GetParameter()->close();
|
|
}
|