Merge pull request #964 from Xertis/main

Implementation of socket:peek

dev/tests/network_tcp.lua

@@ -13,6 +13,7 @@ for i=1,3 do
             print("client-listener started")
             local received_text = ""
             while client:is_alive() and #received_text < #text do
+                client:peek(math.random(0, 100))
                 local received = client:recv(512)
                 if received then
                     received_text = received_text .. utf8.tostring(received)

doc/en/scripting/builtins/libnetwork.md

@@ -92,6 +92,21 @@ socket:recv_async(
     [optional] usetable: bool=false
 ) -> nil|table|Bytearray
 
+-- `peek` and `peek_async` are analogous to the `recv` and `recv_async` methods
+-- with the exception that `peek` and `peek_async` do not advance the socket buffer position
+-- This means they do not remove bytes from the socket, so the bytes can be received after
+socket:peek(
+    length: int,
+    [optional] usetable: boolean=false
+) -> nil|table|Bytearray
+
+socket:peek_async(
+    length: int,
+    [optional] usetable: boolean=false
+) -> nil|table|Bytearray
+
+Translated with DeepL.com (free version)
+
 -- Closes the connection
 socket:close()
 

doc/ru/scripting/builtins/libnetwork.md

@@ -92,6 +92,19 @@ socket:recv_async(
     [опционально] usetable: boolean=false
 ) -> nil|table|Bytearray
 
+-- `peek` и `peek_async` являются аналогами методов `recv` и `recv_async`
+-- за тем исключением, что `peek` и `peek_async` не двигают позицию буфера сокета
+-- Это означает, что они не удаляют байты из сокета, а значит байты могут быть прочитаны после
+socket:peek(
+    length: int,
+    [опционально] usetable: boolean=false
+) -> nil|table|Bytearray
+
+socket:peek_async(
+    length: int,
+    [опционально] usetable: boolean=false
+) -> nil|table|Bytearray
+
 -- Оборачивает сокет в io_stream (см. ../io_stream.md)
 socket:as_stream(
     [опционально] binary_mode: boolean=true

res/scripts/classes.lua

@@ -39,6 +39,7 @@ end
 local Socket = {__index={
     send=function(self, ...) return network.__send(self.id, ...) end,
     recv=function(self, ...) return network.__recv(self.id, ...) end,
+    peek=function(self, ...) return network.__peek(self.id, ...) end,
     recv_async=function(self, length, usetable)
         while self:is_alive() do
             local available = self:available()
@@ -49,6 +50,16 @@ local Socket = {__index={
         end
         return self:recv(length, usetable)
     end,
+    peek_async=function(self, length, usetable)
+        while self:is_alive() do
+            local available = self:available()
+            if available >= length then
+                return self:peek(length, usetable)
+            end
+            coroutine.yield()
+        end
+        return self:peek(length, usetable)
+    end,
     as_stream=network.__as_stream,
     close=function(self) return network.__close(self.id) end,
     available=function(self) return network.__available(self.id) or 0 end,

src/logic/scripting/lua/libs/libnetwork.cpp

@@ -237,7 +237,7 @@ static int l_udp_server_send_to(lua::State* L, network::Network& network) {
     return 0;
 }
 
-static int l_recv(lua::State* L, network::Network& network) {
+static int read(lua::State* L, network::Network& network, int (*fn)(network::TcpConnection*, char*, int)) {
     u64id_t id = lua::tointeger(L, 1);
     int length = lua::tointeger(L, 2);
 
@@ -248,11 +248,12 @@ static int l_recv(lua::State* L, network::Network& network) {
     }
 
     auto tcpConnection = dynamic_cast<network::TcpConnection*>(connection);
+    auto tcp = dynamic_cast<network::TcpConnection*>(connection);
 
     length = glm::min(length, tcpConnection->available());
     util::Buffer<char> buffer(length);
-    
+
-    int size = tcpConnection->recv(buffer.data(), length);
+    int size = fn(tcp, buffer.data(), length);
     if (size == -1) {
         return 0;
     }
@@ -268,6 +269,18 @@ static int l_recv(lua::State* L, network::Network& network) {
     }
 }
 
+static int l_recv(lua::State* L, network::Network& network) {
+    return read(L, network, [](auto* tcp, char* buf, int len) {
+        return tcp->recv(buf, len);
+    });
+}
+
+static int l_peek(lua::State* L, network::Network& network) {
+    return read(L, network, [](auto* tcp, char* buf, int len) {
+        return tcp->peek(buf, len);
+    });
+}
+
 static int l_available(lua::State* L, network::Network& network) {
     u64id_t id = lua::tointeger(L, 1);
 
@@ -566,6 +579,7 @@ const luaL_Reg networklib[] = {
     {"__close", wrap<l_close>},
     {"__send", wrap<l_send>},
     {"__recv", wrap<l_recv>},
+    {"__peek", wrap<l_peek>},
     {"__available", wrap<l_available>},
     {"__is_alive", wrap<l_is_alive>},
     {"__is_connected", wrap<l_is_connected>},

src/network/Sockets.cpp

@@ -196,15 +196,26 @@ public:
         });
     }
 
-    int recv(char* buffer, size_t length) override {
+    int read(char* buffer, size_t length) {
-        std::lock_guard lock(mutex);
-
         if (state != ConnectionState::CONNECTED && readBatch.empty()) {
             return -1;
         }
         int size = std::min(readBatch.size(), length);
         std::memcpy(buffer, readBatch.data(), size);
-        readBatch.erase(readBatch.begin(), readBatch.begin() + size);
+        return size;
+    }
+
+    int peek(char* buffer, size_t length) override {
+        std::lock_guard lock(mutex);
+        return read(buffer, length);
+    }
+
+    int recv(char* buffer, size_t length) override {
+        std::lock_guard lock(mutex);
+        int size = read(buffer, length);
+        if (size != -1) {
+            readBatch.erase(readBatch.begin(), readBatch.begin() + size);
+        }
         return size;
     }
 

src/network/commons.hpp

@@ -79,6 +79,7 @@ namespace network {
     class ReadableConnection : public Connection {
     public:
         virtual int recv(char* buffer, size_t length) = 0;
+        virtual int peek(char* buffer, size_t length) = 0;
         virtual int available() = 0;
     };