diff --git a/lua/autorun/netstream.lua b/lua/autorun/netstream.lua index c0bf9b4f74..ab3b570ca9 100644 --- a/lua/autorun/netstream.lua +++ b/lua/autorun/netstream.lua @@ -66,6 +66,19 @@ local WriteStreamQueue = { timer.Remove("netstream_queueclean") end end, + + CleanStream = function(self, cleanStream) + for k, stream in pairs(self.queue) do + if stream == cleanStream then + stream:Remove() + self.queue[k] = nil + break + end + end + if next(self.queue)==nil then + timer.Remove("netstream_queueclean") + end + end, }, __call = function(t) return setmetatable({ @@ -146,7 +159,7 @@ local WritingDataItem = { __index = { Write = function(self, ply, chunkidx) local client = self.clients[ply] - if client.finished then return false end + if client==nil or client.finished then return false end if chunkidx == #self.chunks+1 then self:Finished(ply) return true end if client.downloads+#self.chunks-client.progress >= net.Stream.MaxTries * #self.chunks then self:Finished(ply) return false end @@ -170,6 +183,15 @@ local WritingDataItem = { Finished = function(self, ply) self.clients[ply].finished = true + if self.protected then + local readyToClean = true + for _, client in pairs(self.clients) do + if not client.finished then readyToClean = false break end + end + if readyToClean then + net.Stream.WriteStreams:CleanStream(self) + end + end if self.callback then local ok, err = xpcall(self.callback, debug.traceback, ply) if not ok then ErrorNoHalt(err) end @@ -177,7 +199,8 @@ local WritingDataItem = { end, GetProgress = function(self, ply) - return self.clients[ply].progress / #self.chunks + local client = self.clients[ply] + return client and (client.progress / #self.chunks) or 0 end, Remove = function(self) @@ -196,21 +219,28 @@ local WritingDataItem = { net.WriteUInt(0, 32) if SERVER then net.Send(sendTo) else net.SendToServer() end end - end + end, }, - __call = function(t, data, callback) + __call = function(t, data, callback, recipients) local chunks = {} for i=1, math.ceil(#data / net.Stream.SendSize) do local datachunk = string.sub(data, (i - 1) * net.Stream.SendSize + 1, i * net.Stream.SendSize) chunks[i] = { data = datachunk, crc = util.CRC(datachunk) } end - return setmetatable({ - timeout = CurTime()+net.Stream.Timeout, - chunks = chunks, - callback = callback, - lasttouched = 0, + local clients, protected + if recipients then + clients = {} + for k, v in ipairs(recipients) do + clients[v] = { + finished = false, + downloads = 0, + progress = 0 + } + end + protected = true + else clients = setmetatable({},{__index = function(t,k) local r = { finished = false, @@ -218,6 +248,16 @@ local WritingDataItem = { progress = 0, } t[k]=r return r end}) + protected = false + end + + return setmetatable({ + timeout = CurTime()+net.Stream.Timeout, + chunks = chunks, + callback = callback, + lasttouched = 0, + clients = clients, + protected = protected, }, t) end } @@ -297,7 +337,7 @@ local ReadingDataItem = { setmetatable(ReadingDataItem, ReadingDataItem) -function net.WriteStream(data, callback, dontcompress) +function net.WriteStream(data, callback, dontcompress, recipients) if not isstring(data) then error("bad argument #1 to 'WriteStream' (string expected, got " .. type(data) .. ")", 2) end @@ -309,6 +349,14 @@ function net.WriteStream(data, callback, dontcompress) if compressed then data = util.Compress(data) or "" end + if recipients ~= nil then + if not istable(recipients) then error("bad argument #4 to 'WriteStream' (table expected, got " .. type(recipients) .. ")", 2) end + if CLIENT then error("recipients argument can't be used in clientside!", 2) end + if #recipients==0 then error("recipients argument is empty!", 2) end + for k, v in ipairs(recipients) do + if not v:IsPlayer() then error("recipients contains non-player entry in key ("..k..")!", 2) end + end + end if #data == 0 then net.WriteUInt(0, 32) @@ -321,7 +369,7 @@ function net.WriteStream(data, callback, dontcompress) return end - local stream = net.Stream.WriteStreams:Add(WritingDataItem(data, callback, compressed)) + local stream = net.Stream.WriteStreams:Add(WritingDataItem(data, callback, compressed, recipients)) if not stream then return end --print("WriteStream", #stream.chunks, stream.identifier, compressed) diff --git a/lua/entities/gmod_wire_customprop/init.lua b/lua/entities/gmod_wire_customprop/init.lua index b8f5f6f54d..a193b9bbb5 100644 --- a/lua/entities/gmod_wire_customprop/init.lua +++ b/lua/entities/gmod_wire_customprop/init.lua @@ -81,7 +81,7 @@ end function ENT:TransmitData(recip) net.Start(shared.classname) shared.writeReliableEntity(self) - local stream = net.WriteStream(self.wiremeshdata, nil, true) + local stream = net.WriteStream(self.wiremeshdata, nil, true, recip and {recip} or nil) if recip then net.Send(recip) else net.Broadcast() end return stream diff --git a/lua/entities/gmod_wire_expression2/core/files.lua b/lua/entities/gmod_wire_expression2/core/files.lua index 44ac708524..9447eff05d 100644 --- a/lua/entities/gmod_wire_expression2/core/files.lua +++ b/lua/entities/gmod_wire_expression2/core/files.lua @@ -361,7 +361,7 @@ flushFileBuffer = function() net.WriteUInt(0, 32) finished_cb() else - net.WriteStream(data, finished_cb) + net.WriteStream(data, finished_cb, nil, {ply}) end net.Send(ply) end diff --git a/lua/entities/gmod_wire_expression2/core/init.lua b/lua/entities/gmod_wire_expression2/core/init.lua index afbda09b2c..b4d607f8d2 100644 --- a/lua/entities/gmod_wire_expression2/core/init.lua +++ b/lua/entities/gmod_wire_expression2/core/init.lua @@ -347,7 +347,7 @@ if SERVER then } ) net.Start("e2_functiondata") - net.WriteStream(data) + net.WriteStream(data, nil, nil, {ply}) net.Send(ply) end diff --git a/lua/wire/stools/expression2.lua b/lua/wire/stools/expression2.lua index 23806fb604..a4c0a377e8 100644 --- a/lua/wire/stools/expression2.lua +++ b/lua/wire/stools/expression2.lua @@ -267,7 +267,7 @@ if SERVER then net.Start("wire_expression2_download") net.WriteEntity(targetEnt) net.WriteBit(uploadandexit or false) - net.WriteStream(datastr) + net.WriteStream(datastr, nil, nil, {ply}) net.Send(ply) return end @@ -282,7 +282,7 @@ if SERVER then net.Start("wire_expression2_download_wantedfiles_list") net.WriteEntity(targetEnt) net.WriteBit(uploadandexit or false) - net.WriteStream(datastr) + net.WriteStream(datastr, nil, nil, {ply}) net.Send(ply) targetEnt.DownloadAllowedPlayers = targetEnt.DownloadAllowedPlayers or {} @@ -309,7 +309,7 @@ if SERVER then net.Start("wire_expression2_download") net.WriteEntity(targetEnt) net.WriteBit(uploadandexit or false) - net.WriteStream(datastr) + net.WriteStream(datastr, nil, nil, {ply}) net.Send(ply) end diff --git a/lua/wire/wirenet.lua b/lua/wire/wirenet.lua index 9e29d10cc6..7d43d2b0cf 100644 --- a/lua/wire/wirenet.lua +++ b/lua/wire/wirenet.lua @@ -60,7 +60,7 @@ if SERVER then end else net.WriteBool(true) - net.WriteStream(data, nil, false) + net.WriteStream(data, nil, false, ply and {ply} or nil) end if ply then net.Send(ply) else net.Broadcast() end