diff --git a/client.comet b/client.comet index 9ddf841..617ebe1 100644 --- a/client.comet +++ b/client.comet @@ -4,13 +4,12 @@ import io func main() -> int { socket.Socket s = new socket.Socket(socket.Family.INET, socket.Type.STREAM, 0) - io.println("Connecting...") s.connect("127.0.0.1", 2000) - - io.println("Sending data...") s.send("hi!") - io.println("Closing socket...") + small byte = s.readByte() + io.println("%c", byte) + s.close() return 0 diff --git a/server.comet b/server.comet index c2aa463..a11fe7b 100644 --- a/server.comet +++ b/server.comet @@ -10,9 +10,11 @@ func main() -> int { socket.Socket client = s.accept() io.println("Connection received!") - string response = client.recv(256) + string response = client.read(1024) io.println("Response: %s", response) + client.sendByte('a') + s.close() return 0 diff --git a/src/address.c b/src/address.c index 63d3e5b..9f956fe 100644 --- a/src/address.c +++ b/src/address.c @@ -1,6 +1,6 @@ #include "address.h" -ResultType(int64_t, objectPtr) impl_Address_INIT(int64_t* args, CometVM* vm) { +ResultType(int64_t, objectPtr) impl_SocketAddress_INIT(int64_t* args, CometVM* vm) { CometObject* self = (CometObject*)args[0]; self->fields[0] = args[1]; @@ -10,7 +10,7 @@ ResultType(int64_t, objectPtr) impl_Address_INIT(int64_t* args, CometVM* vm) { } CometStruct* defineAddressStruct(CometEnvironment* env) { - CometStruct* addressStruct = cometDefineStruct(env, "Address", NULL); + CometStruct* addressStruct = cometDefineStruct(env, "SocketAddress", NULL); List(StructField) fields = newList(StructField); append(fields, cometCreateField("ip", cometTypeString, FIELD_PUBLIC)); diff --git a/src/socket.c b/src/socket.c index b49f5f9..93f8f0d 100644 --- a/src/socket.c +++ b/src/socket.c @@ -112,7 +112,7 @@ ResultType(int64_t, objectPtr) impl_accept(int64_t* args, CometVM* vm) { return Success(int64_t, objectPtr, (int64_t)clientCometSocket); } -ResultType(int64_t, objectPtr) impl_recv(int64_t* args, CometVM* vm) { +ResultType(int64_t, objectPtr) impl_read(int64_t* args, CometVM* vm) { int amount = cometArgInt(args[0]); CometObject* self = (CometObject*)args[1]; @@ -120,7 +120,7 @@ ResultType(int64_t, objectPtr) impl_recv(int64_t* args, CometVM* vm) { char* buffer = malloc(amount + 1); if (!buffer) - return cometError(vm, "MemoryAllocFail", "Failed to allocate memory for buffer (recv)"); + return cometError(vm, "MemoryAllocFail", "Failed to allocate memory for buffer (read)"); ssize_t amountReceived = recv(fd, buffer, amount, 0); if (amountReceived > 0) { @@ -137,6 +137,95 @@ ResultType(int64_t, objectPtr) impl_recv(int64_t* args, CometVM* vm) { return Success(int64_t, objectPtr, serializedString); } +ResultType(int64_t, objectPtr) impl_readExact(int64_t* args, CometVM* vm) { + int amount = cometArgInt(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + char* buffer = malloc(amount + 1); + if (!buffer) + return cometError(vm, "MemoryAllocFail", "Failed to allocate memory for buffer (readExact)"); + + ssize_t amountReceived = recv(fd, buffer, amount, 0); + if (amountReceived != amount) { + char* buffer = malloc(128); + snprintf(buffer, 128, "Expected to read %d bytes but only got %zu", amount, amountReceived); + return cometError(vm, "UnexpectedEOF", buffer); + } else if (amountReceived == -1) { // connection closed by peer + char* errMsg = strerror(errno); + close(fd); + free(buffer); + return cometError(vm, "SocketError", errMsg); + } + + buffer[amountReceived] = 0; + + int64_t serializedString = cometSerializeString(buffer); + free(buffer); + return Success(int64_t, objectPtr, serializedString); +} + +ResultType(int64_t, objectPtr) impl_readByte(int64_t* args, CometVM* vm) { + CometObject* self = (CometObject*)args[0]; + int fd = self->fields[0]; + + char buffer[2]; + + ssize_t amountReceived = recv(fd, buffer, 1, 0); + if (amountReceived < 0) { + // connection closed by peer + char* errMsg = strerror(errno); + close(fd); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, buffer[0]); +} + +ResultType(int64_t, objectPtr) impl_readAll(int64_t* args, CometVM* vm) { + CometObject* self = (CometObject*)args[0]; + int fd = self->fields[0]; + + size_t capacity = 1024; + size_t length = 0; + char* buffer = malloc(capacity); + if (!buffer) { + return cometError(vm, "MemoryAllocFail", "Failed to allocate buffer (readAll)"); + } + + while (true) { + if (length == capacity) { + capacity *= 2; + + char* newBuff = realloc(buffer, capacity); + if (!newBuff) { + free(buffer); + return cometError(vm, "MemoryAllocFail", "Failed to reallocate buffer (readAll)"); + } + } + + size_t n = recv(fd, buffer + length, capacity - length, 0); + + if (n == 0) { // connection closed + break; + } + + if (n < 0) { + char* errMsg = strerror(errno); + free(buffer); + return cometError(vm, "SocketError", errMsg); + } + + length += n; + } + + int64_t serializedString = cometSerializeString(buffer); + free(buffer); + + return Success(int64_t, objectPtr, serializedString); +} + ResultType(int64_t, objectPtr) impl_send(int64_t* args, CometVM* vm) { char* buffer = cometArgString(args[0]); CometObject* self = (CometObject*)args[1]; @@ -152,6 +241,23 @@ ResultType(int64_t, objectPtr) impl_send(int64_t* args, CometVM* vm) { return Success(int64_t, objectPtr, 0); } +ResultType(int64_t, objectPtr) impl_sendByte(int64_t* args, CometVM* vm) { + char byte = cometArgSmall(args[0]); + CometObject* self = (CometObject*)args[1]; + + char buffer[1] = {byte}; + + int fd = self->fields[0]; + + ssize_t bytesSent = send(fd, buffer, 1, MSG_NOSIGNAL); + if (bytesSent < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + ResultType(int64_t, objectPtr) impl_close(int64_t* args, CometVM* vm) { CometObject* self = (CometObject*)args[0]; @@ -166,6 +272,158 @@ ResultType(int64_t, objectPtr) impl_close(int64_t* args, CometVM* vm) { return Success(int64_t, objectPtr, 0); } +ResultType(int64_t, objectPtr) impl_setBlocking(int64_t* args, CometVM* vm) { + bool blocking = cometArgBool(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + int flags = fcntl(fd, F_GETFL, 0); + if (flags == -1) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + if (blocking) + flags &= ~O_NONBLOCK; + else + flags |= O_NONBLOCK; + + int result = fcntl(fd, F_SETFL, flags); + if (result < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + +ResultType(int64_t, objectPtr) impl_setTimeout(int64_t* args, CometVM* vm) { + int ms = cometArgInt(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + struct timeval tv; + tv.tv_sec = ms / 1000; + tv.tv_usec = (ms % 1000) * 1000; + + if (setsockopt(fd, SOL_SOCKET, + SO_RCVTIMEO, + &tv, sizeof(tv)) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + if (setsockopt(fd, SOL_SOCKET, + SO_SNDTIMEO, + &tv, sizeof(tv)) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + +ResultType(int64_t, objectPtr) impl_setNoDelay(int64_t* args, CometVM* vm) { + bool noDelay = cometArgBool(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + if (setsockopt(fd, + IPPROTO_TCP, + TCP_NODELAY, + &noDelay, + sizeof(noDelay)) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + +ResultType(int64_t, objectPtr) impl_setReuseAddress(int64_t* args, CometVM* vm) { + bool reuseAddr = cometArgBool(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + if (setsockopt(fd, + SOL_SOCKET, + SO_REUSEADDR, + &reuseAddr, + sizeof(reuseAddr)) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + +ResultType(int64_t, objectPtr) impl_setKeepAlive(int64_t* args, CometVM* vm) { + bool keepAlive = cometArgBool(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + if (setsockopt(fd, + SOL_SOCKET, + SO_KEEPALIVE, + &keepAlive, + sizeof(keepAlive)) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, 0); +} + +ResultType(int64_t, objectPtr) impl_available(int64_t* args, CometVM* vm) { + CometObject* self = (CometObject*)args[0]; + int fd = self->fields[0]; + + int count = 0; + if (ioctl(fd, FIONREAD, &count) < 0) { + char* errMsg = strerror(errno); + return cometError(vm, "SocketError", errMsg); + } + + return Success(int64_t, objectPtr, count); +} + +ResultType(int64_t, objectPtr) impl_waitReadable(int64_t* args, CometVM* vm) { + int timeoutMs = cometArgInt(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + struct pollfd pfd = { + .fd = fd, + .events = POLLIN + }; + + int ret = poll(&pfd, 1, timeoutMs); + + return Success(int64_t, objectPtr, ret > 0 && (pfd.revents & POLLIN)); +} + +ResultType(int64_t, objectPtr) impl_waitWriteable(int64_t* args, CometVM* vm) { + int timeoutMs = cometArgInt(args[0]); + CometObject* self = (CometObject*)args[1]; + + int fd = self->fields[0]; + + struct pollfd pfd = { + .fd = fd, + .events = POLLOUT + }; + + int ret = poll(&pfd, 1, timeoutMs); + + return Success(int64_t, objectPtr, ret > 0 && (pfd.revents & POLLOUT)); +} + void defineSocketStruct(CometEnvironment* env, CometTypeMap* typeMap) { CometType familyEnum = defineFamilyEnum(env, typeMap); CometType typeEnum = defineTypeEnum(env, typeMap); @@ -192,9 +450,26 @@ void defineSocketStruct(CometEnvironment* env, CometTypeMap* typeMap) { append(methods, cometDefineMethod(env, "accept", socketStruct, FIELD_PUBLIC, socketStructType, 0, false)); // general - append(methods, cometDefineMethod(env, "recv", socketStruct, FIELD_PUBLIC, cometTypeString, 1, false, cometTypeInt)); + append(methods, cometDefineMethod(env, "read", socketStruct, FIELD_PUBLIC, cometTypeString, 1, false, cometTypeInt)); + append(methods, cometDefineMethod(env, "readAll", socketStruct, FIELD_PUBLIC, cometTypeString, 0, false)); + append(methods, cometDefineMethod(env, "readExact", socketStruct, FIELD_PUBLIC, cometTypeString, 1, false, cometTypeInt)); + append(methods, cometDefineMethod(env, "readByte", socketStruct, FIELD_PUBLIC, cometTypeSmall, 0, false)); append(methods, cometDefineMethod(env, "send", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeString)); + append(methods, cometDefineMethod(env, "sendByte", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeSmall)); append(methods, cometDefineMethod(env, "close", socketStruct, FIELD_PUBLIC, cometTypeVoid, 0, false)); + + // options + append(methods, cometDefineMethod(env, "setBlocking", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeBool)); + append(methods, cometDefineMethod(env, "setTimeout", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeInt)); + append(methods, cometDefineMethod(env, "setNoDelay", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeBool)); + append(methods, cometDefineMethod(env, "setReuseAddress", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeBool)); + append(methods, cometDefineMethod(env, "setKeepAlive", socketStruct, FIELD_PUBLIC, cometTypeVoid, 1, false, cometTypeBool)); + + // other + append(methods, cometDefineMethod(env, "available", socketStruct, FIELD_PUBLIC, cometTypeInt, 0, false)); + append(methods, cometDefineMethod(env, "waitReadable", socketStruct, FIELD_PUBLIC, cometTypeBool, 1, false, cometTypeInt)); + append(methods, cometDefineMethod(env, "waitWriteable", socketStruct, FIELD_PUBLIC, cometTypeBool, 1, false, cometTypeInt)); + /// END OF METHODS /// cometSetStructFieldsAndMethods(socketStruct, fields, methods); diff --git a/src/socket.h b/src/socket.h index 75adce9..90c6139 100644 --- a/src/socket.h +++ b/src/socket.h @@ -4,9 +4,14 @@ #include #include #include +#include #include +#include #include #include +#include +#include +#include #include "socket_enums.h" void defineSocketStruct(CometEnvironment* env, CometTypeMap* typeMap);