From 246e50d166565fb487bf392dec5bfbee29db95ed Mon Sep 17 00:00:00 2001 From: ruki Date: Sun, 10 Apr 2022 17:36:29 +0800 Subject: rename stream to socket stream --- xmake/modules/private/service/connect_service.lua | 8 +- .../private/service/server/remote_build_server.lua | 6 +- xmake/modules/private/service/server/server.lua | 6 +- xmake/modules/private/service/socket_stream.lua | 221 +++++++++++++++++++++ xmake/modules/private/service/stream.lua | 221 --------------------- 5 files changed, 231 insertions(+), 231 deletions(-) create mode 100644 xmake/modules/private/service/socket_stream.lua delete mode 100644 xmake/modules/private/service/stream.lua diff --git a/xmake/modules/private/service/connect_service.lua b/xmake/modules/private/service/connect_service.lua index eabcdfc32..9509b3aab 100644 --- a/xmake/modules/private/service/connect_service.lua +++ b/xmake/modules/private/service/connect_service.lua @@ -23,7 +23,7 @@ import("core.base.option") import("core.base.socket") import("core.base.scheduler") import("private.service.config") -import("private.service.stream") +import("private.service.socket_stream") import("private.service.message") import("private.service.client.remote_build_client") @@ -49,9 +49,9 @@ function _connect(addr, port) print("%s: connect %s:%d ..", client, addr, port) if sock then print("%s: connected!", client) - local sock_stream = stream(sock) - if sock_stream:send_msg(message.new_ping()) and sock_stream:flush() then - local msg = sock_stream:recv_msg() + local stream = socket_stream(sock) + if stream:send_msg(message.new_ping()) and stream:flush() then + local msg = stream:recv_msg() if msg then msg:dump() end diff --git a/xmake/modules/private/service/server/remote_build_server.lua b/xmake/modules/private/service/server/remote_build_server.lua index 57f697f59..a323132a6 100644 --- a/xmake/modules/private/service/server/remote_build_server.lua +++ b/xmake/modules/private/service/server/remote_build_server.lua @@ -20,7 +20,7 @@ -- imports import("private.service.config") -import("private.service.stream") +import("private.service.socket_stream") import("private.service.message") import("private.service.server.server") @@ -41,8 +41,8 @@ end -- handle ping message function remote_build_server:handle_ping(sock, msg) - local wstream = stream(sock) - if wstream:send_msg(message.new_ping()) and wstream:flush() then + local stream = socket_stream(sock) + if stream:send_msg(message.new_ping()) and stream:flush() then print("send ok") end end diff --git a/xmake/modules/private/service/server/server.lua b/xmake/modules/private/service/server/server.lua index d3035eddf..ea1831934 100644 --- a/xmake/modules/private/service/server/server.lua +++ b/xmake/modules/private/service/server/server.lua @@ -23,7 +23,7 @@ import("core.base.object") import("core.base.bytes") import("core.base.socket") import("core.base.scheduler") -import("private.service.stream") +import("private.service.socket_stream") import("private.service.message") -- define module @@ -121,9 +121,9 @@ end -- handle session function server:_handle_session(sock) print("%s: %s session connected", self, sock) - local rstream = stream(sock) + local stream = socket_stream(sock) while true do - local msg = rstream:recv_object() + local msg = stream:recv_object() if msg then self:_HANDLER(sock, message(msg)) else diff --git a/xmake/modules/private/service/socket_stream.lua b/xmake/modules/private/service/socket_stream.lua new file mode 100644 index 000000000..74f71c330 --- /dev/null +++ b/xmake/modules/private/service/socket_stream.lua @@ -0,0 +1,221 @@ +--!A cross-platform build utility based on Lua +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); +-- you may not use this file except in compliance with the License. +-- You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +-- +-- Copyright (C) 2015-present, TBOOX Open Source Group. +-- +-- @author ruki +-- @file socket_stream.lua +-- + +-- imports +import("core.base.object") +import("core.base.socket") +import("core.base.bytes") +import("private.service.message") + +-- define module +local socket_stream = socket_stream or object() + +-- init socket_stream +function socket_stream:init(sock) + self._SOCK = sock + self._BUFF = bytes(65536) + self._RCACHE = bytes(8192) + self._RCACHE_SIZE = 0 + self._WCACHE = bytes(8192) + self._WCACHE_SIZE = 0 +end + +-- flush data +function socket_stream:flush() + local cache = self._WCACHE + local cache_size = self._WCACHE_SIZE + if cache_size > 0 then + local sock = self._SOCK + local real = sock:send(cache, {block = true, last = cache_size}) + if real > 0 then + self._WCACHE_SIZE = 0 + return true + end + else + return true + end +end + +-- send the given bytes +function socket_stream:send(data, start, last) + start = start or 1 + last = last or data:size() + local size = last + 1 - start + assert(size <= data:size()) + + -- write data to cache first + local cache = self._WCACHE + local cache_size = self._WCACHE_SIZE + local cache_maxn = cache:size() + local cache_left = cache_maxn - cache_size + if size <= cache_left then + cache:copy2(cache_size + 1, data, start, last) + cache_size = cache_size + size + self._WCACHE_SIZE = cache_size + return true + elseif cache_left > 0 then + cache:copy2(cache_size + 1, data, start, start + cache_left - 1) + cache_size = cache_size + cache_left + start = start + cache_left + size = last + 1 - start + end + assert(cache_size == cache_maxn) + + -- send data to socket + local sock = self._SOCK + local real = sock:send(cache, {block = true}) + if real > 0 then + -- copy left data to cache + assert(size <= cache_maxn) + cache:copy2(1, data, start, last) + self._WCACHE_SIZE = size + return true + end +end + +-- send message +function socket_stream:send_msg(msg) + return self:send_object(msg:body()) +end + +-- send object +function socket_stream:send_object(obj) + local str, errors = string.serialize(obj, {strip = true, indent = false}) + if errors then + raise(errors) + end + if str then + return self:send_string(str) + end +end + +-- send string +function socket_stream:send_string(str) + local buff = self._BUFF + local size = #str + buff:u16be_set(1, size) + if self:send(buff, 1, 2) then + return self:send(bytes(str), 1, size) + end +end + +-- recv the given bytes +function socket_stream:recv(buff, size) + assert(size <= buff:size()) + + -- read data from cache first + local buffsize = 0 + local cache = self._RCACHE + local cache_size = self._RCACHE_SIZE + local cache_maxn = cache:size() + if size <= cache_size then + buff:copy(cache, 1, size) + cache_size = cache_size - size + self._RCACHE_SIZE = cache_size + return buff:slice(1, size) + elseif cache_size > 0 then + buff:copy(cache, 1, cache_size) + buffsize = cache_size + cache_size = 0 + end + assert(cache_size == 0) + + -- recv data from socket + local real = 0 + local data = nil + local wait = false + local sock = self._SOCK + while buffsize < size do + real, data = sock:recv(cache) + if real > 0 then + -- append data to buffer + local leftsize = size - buffsize + if real < leftsize then + buff:copy2(buffsize + 1, data) + buffsize = buffsize + real + else + buff:copy2(buffsize + 1, data, 1, leftsize) + buffsize = buffsize + leftsize + + -- move left cache to head + cache_size = real - leftsize + if cache_size > 0 then + cache:move(leftsize + 1, real) + end + self._RCACHE_SIZE = cache_size + return buff:slice(1, buffsize) + end + wait = false + elseif real == 0 and not wait then + if sock:wait(socket.EV_RECV, -1) == socket.EV_RECV then + wait = true + else + break + end + else + break + end + end +end + +-- recv u16be +function socket_stream:recv_u16be() + local data = self:recv(self._BUFF, 2) + if data then + return data:u16be(1) + end +end + +-- recv message +function socket_stream:recv_msg() + local body = self:recv_object() + if body then + return message(body) + end +end + +-- recv object +function socket_stream:recv_object() + local str = self:recv_string() + if str then + local obj, errors = str:deserialize() + if errors then + raise(errors) + end + return obj + end +end + +-- recv string +function socket_stream:recv_string() + local size = self:recv_u16be() + if size then + local data = self:recv(self._BUFF, size) + if data then + return data:str() + end + end +end + +function main(sock) + local instance = socket_stream() + instance:init(sock) + return instance +end diff --git a/xmake/modules/private/service/stream.lua b/xmake/modules/private/service/stream.lua deleted file mode 100644 index d7000090e..000000000 --- a/xmake/modules/private/service/stream.lua +++ /dev/null @@ -1,221 +0,0 @@ ---!A cross-platform build utility based on Lua --- --- Licensed under the Apache License, Version 2.0 (the "License"); --- you may not use this file except in compliance with the License. --- You may obtain a copy of the License at --- --- http://www.apache.org/licenses/LICENSE-2.0 --- --- Unless required by applicable law or agreed to in writing, software --- distributed under the License is distributed on an "AS IS" BASIS, --- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. --- See the License for the specific language governing permissions and --- limitations under the License. --- --- Copyright (C) 2015-present, TBOOX Open Source Group. --- --- @author ruki --- @file stream.lua --- - --- imports -import("core.base.object") -import("core.base.socket") -import("core.base.bytes") -import("private.service.message") - --- define module -local stream = stream or object() - --- init stream -function stream:init(sock) - self._SOCK = sock - self._BUFF = bytes(65536) - self._RCACHE = bytes(8192) - self._RCACHE_SIZE = 0 - self._WCACHE = bytes(8192) - self._WCACHE_SIZE = 0 -end - --- flush data -function stream:flush() - local cache = self._WCACHE - local cache_size = self._WCACHE_SIZE - if cache_size > 0 then - local sock = self._SOCK - local real = sock:send(cache, {block = true, last = cache_size}) - if real > 0 then - self._WCACHE_SIZE = 0 - return true - end - else - return true - end -end - --- send the given bytes -function stream:send(data, start, last) - start = start or 1 - last = last or data:size() - local size = last + 1 - start - assert(size <= data:size()) - - -- write data to cache first - local cache = self._WCACHE - local cache_size = self._WCACHE_SIZE - local cache_maxn = cache:size() - local cache_left = cache_maxn - cache_size - if size <= cache_left then - cache:copy2(cache_size + 1, data, start, last) - cache_size = cache_size + size - self._WCACHE_SIZE = cache_size - return true - elseif cache_left > 0 then - cache:copy2(cache_size + 1, data, start, start + cache_left - 1) - cache_size = cache_size + cache_left - start = start + cache_left - size = last + 1 - start - end - assert(cache_size == cache_maxn) - - -- send data to socket - local sock = self._SOCK - local real = sock:send(cache, {block = true}) - if real > 0 then - -- copy left data to cache - assert(size <= cache_maxn) - cache:copy2(1, data, start, last) - self._WCACHE_SIZE = size - return true - end -end - --- send message -function stream:send_msg(msg) - return self:send_object(msg:body()) -end - --- send object -function stream:send_object(obj) - local str, errors = string.serialize(obj, {strip = true, indent = false}) - if errors then - raise(errors) - end - if str then - return self:send_string(str) - end -end - --- send string -function stream:send_string(str) - local buff = self._BUFF - local size = #str - buff:u16be_set(1, size) - if self:send(buff, 1, 2) then - return self:send(bytes(str), 1, size) - end -end - --- recv the given bytes -function stream:recv(buff, size) - assert(size <= buff:size()) - - -- read data from cache first - local buffsize = 0 - local cache = self._RCACHE - local cache_size = self._RCACHE_SIZE - local cache_maxn = cache:size() - if size <= cache_size then - buff:copy(cache, 1, size) - cache_size = cache_size - size - self._RCACHE_SIZE = cache_size - return buff:slice(1, size) - elseif cache_size > 0 then - buff:copy(cache, 1, cache_size) - buffsize = cache_size - cache_size = 0 - end - assert(cache_size == 0) - - -- recv data from socket - local real = 0 - local data = nil - local wait = false - local sock = self._SOCK - while buffsize < size do - real, data = sock:recv(cache) - if real > 0 then - -- append data to buffer - local leftsize = size - buffsize - if real < leftsize then - buff:copy2(buffsize + 1, data) - buffsize = buffsize + real - else - buff:copy2(buffsize + 1, data, 1, leftsize) - buffsize = buffsize + leftsize - - -- move left cache to head - cache_size = real - leftsize - if cache_size > 0 then - cache:move(leftsize + 1, real) - end - self._RCACHE_SIZE = cache_size - return buff:slice(1, buffsize) - end - wait = false - elseif real == 0 and not wait then - if sock:wait(socket.EV_RECV, -1) == socket.EV_RECV then - wait = true - else - break - end - else - break - end - end -end - --- recv u16be -function stream:recv_u16be() - local data = self:recv(self._BUFF, 2) - if data then - return data:u16be(1) - end -end - --- recv message -function stream:recv_msg() - local body = self:recv_object() - if body then - return message(body) - end -end - --- recv object -function stream:recv_object() - local str = self:recv_string() - if str then - local obj, errors = str:deserialize() - if errors then - raise(errors) - end - return obj - end -end - --- recv string -function stream:recv_string() - local size = self:recv_u16be() - if size then - local data = self:recv(self._BUFF, size) - if data then - return data:str() - end - end -end - -function main(sock) - local instance = stream() - instance:init(sock) - return instance -end -- cgit v1.3.1