Add recipients to WriteStream

lua/autorun/netstream.lua

@@ -66,6 +66,19 @@
 				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({
@@ -87,7 +100,7 @@
 				ErrorNoHalt("Receiving too many ReadStream requests!")
 				return
 			end
 
 			for _, v in ipairs(queue) do
 				if v.identifier == stream.identifier then
 					ErrorNoHalt("Tried to start a new ReadStream for an already existing stream!")
@@ -146,7 +159,7 @@
 	__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,14 +183,24 @@
 
 		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
 			end
 		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,28 +219,45 @@
 				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,
 					downloads = 0,
 					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 @@
 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 @@
 	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,9 +369,9 @@
 		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)
 	net.WriteUInt(#stream.chunks, 32)
 	net.WriteUInt(stream.identifier, 32)
@@ -347,7 +395,7 @@
 	if not isfunction(callback) then
 		error("bad argument #2 to 'ReadStream' (function expected, got " .. type(callback) .. ")", 2)
 	end
 
 	local numchunks = net.ReadUInt(32)
 	if numchunks == nil then
 		return

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

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

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
 

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
 

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