replaced Stream:foreach() with Stream:iter()

* This allows the native syntax `for f in Stream:iter() do ... end` without using lambda
  functions.
  Also you can use `break` and `return` statements.
* On the other hand we cannot exploit the extended xpcall() semantics and had to introduce
  another lambda in Stream:play().
* In general the number of function calls per tick stays the same.
  Stream:gtick() itself could be used as an iterator, but Stream:iter() adds checking for CTRL+C,
  resetting of the sample cache and binding functions.

3 files changed, 53 insertions, 45 deletions

diff --git a/TODO b/TODO
index 58240ed..0fbacf9 100644
--- a/TODO
+++ b/TODO
@@ -52,7 +52,7 @@
   Perhaps bundle https://github.com/Xpra-org/xpra/.
   The HTML client might even allow integrating Tk windows into Jupyter cells?
 * Widget toolkit integration.
-  We could optionally integrate Gtk (lgi) or Tcl and drive their main loops from Stream:foreach() or Stream:play()
+  We could optionally integrate Gtk (lgi) or Tcl and drive their main loops from Stream:iter() or Stream:play()
   whenever their modules are loaded.
   This would allow creating GUIs to control streams without having to go via external MIDI
   applications. There would also have to be a new Stream class that reads values always from the
@@ -114,3 +114,6 @@
   Tk canvases.
 * Automatically detect whether terminal supports the Kitty graphics protocol:
   https://sw.kovidgoyal.net/kitty/graphics-protocol/#querying-support-and-available-transmission-mediums
+* Two C non-inlineable function calls per sample due to applause_push_sample() could be avoided by including our
+  own lockless ring-buffer implementation.
+* Document ZipStream semantics - see chapter Stream Algebra in my thesis.
diff --git a/applause.lua b/applause.lua
index bf94d5d..5d1ba5a 100644
--- a/applause.lua
+++ b/applause.lua
@@ -661,26 +661,28 @@ function Stream:play(first_port)
 	local old_stepmul = collectgarbage("setstepmul", 100)
 
 	local channels = self.channels
-	local _, err = xpcall(self.foreach, debug.traceback, self, function(frame)
-		-- Loop should get unrolled automatically
-		for i = 1, channels do
-			local sample = tonumber(frame[i])
-			assert(sample ~= nil)
-
-			-- NOTE: Invalid port Ids are currently silently
-			-- ignored. Perhaps it's better to check state or
-			-- to access output_ports_count from applause.c.
-			local state = C.applause_push_sample(first_port+i, sample)
-
-			-- React to buffer underruns.
-			-- This is done here instead of in the realtime thread
-			-- even though it is already overloaded, so as not to
-			-- affect other applications in the Jack graph.
-			if state == C.APPLAUSE_AUDIO_XRUN then
-				io.stderr:write("WARNING: Buffer underrun detected\n")
+	local _, err = xpcall(function()
+		for frame in self:iter() do
+			-- Loop should get unrolled automatically
+			for i = 1, channels do
+				local sample = tonumber(frame[i])
+				assert(sample ~= nil)
+
+				-- NOTE: Invalid port Ids are currently silently
+				-- ignored. Perhaps it's better to check state or
+				-- to access output_ports_count from applause.c.
+				local state = C.applause_push_sample(first_port+i, sample)
+
+				-- React to buffer underruns.
+				-- This is done here instead of in the realtime thread
+				-- even though it is already overloaded, so as not to
+				-- affect other applications in the Jack graph.
+				if state == C.APPLAUSE_AUDIO_XRUN then
+					io.stderr:write("WARNING: Buffer underrun detected\n")
+				end
 			end
 		end
-	end)
+	end, debug.traceback)
 
 	collectgarbage("setpause", old_pause)
 	collectgarbage("setstepmul", old_stepmul)
@@ -688,25 +690,26 @@ function Stream:play(first_port)
 	if err then error(err, 0) end
 end
 
---- Execute function for each frame generated by the stream.
+--- Iterator over each frame generated by the stream for use with for-in statements.
 -- This will process samples as fast as possible and may therefore
 -- not be well suited to process real-time input.
--- @func fnc
---   Function to execute.
---   It gets passed an array of samples, one for each channel.
---   If it returns true, the loop is terminated.
-function Stream:foreach(fnc)
+-- @treturn fnc Iterator function.
+-- @usage for f in iota(10):iter() do print(f[1]) end
+function Stream:iter()
 	-- NOTE: This implementation is for single-channel streams
-	-- only. See also MuxStream:foreach().
+	-- only. See also MuxStream:iter().
 	local clear = table.clear
 	local frame = table.new(1, 0)
 	local tick = self:gtick()
 
-	repeat
+	-- This could in principle use iterator state to avoid the closure,
+	-- but we want to bind table.clear() as well.
+	return function()
 		checkint()
 		frame[1] = tick()
 		clear(sampleCache)
-	until not frame[1] or fnc(frame)
+		return frame[1] and frame
+	end
 end
 
 --- Benchmark stream (time to generate all samples).
@@ -720,7 +723,7 @@ function Stream:benchmark()
 	end
 
 	benchmark(function()
-		self:foreach(function() end)
+		for _ in self:iter() do end
 	end)
 end
 
@@ -789,12 +792,12 @@ function Stream:totable()
 		channel_vectors[i] = table.new(self:len(), 0)
 	end
 
-	self:foreach(function(frame)
+	for frame in self:iter() do
 		-- Loop should be unrolled automatically
 		for i = 1, channels do
 			channel_vectors[i][#channel_vectors[i] + 1] = frame[i]
 		end
-	end)
+	end
 
 	-- Return a list of vectors, one per channel
 	return unpack(channel_vectors)
@@ -867,10 +870,10 @@ function Stream:pipe(prog, vbufmode, vbufsize)
 	local hnd = io.popen(prog, "w")
 	hnd:setvbuf(vbufmode or "full", vbufsize)
 
-	self:foreach(function(frame)
+	for frame in self:iter() do
 		hnd:write(unpack(frame))
 		hnd:write("\n")
-	end)
+	end
 
 	hnd:close()
 end
@@ -923,11 +926,11 @@ function Stream:gnuplot(file)
 
 	local second = sec()
 	local i = 1
-	self:foreach(function(frame)
+	for frame in self:iter() do
 		hnd:write(i/second, " ", unpack(frame))
 		hnd:write("\n")
 		i = i + 1
-	end)
+	end
 
 	hnd:close()
 
@@ -1274,14 +1277,14 @@ function MuxStream:demux(i, j)
 	              or MuxStream:new(unpack(self.streams, i, j))
 end
 
--- Overrides Stream:foreach()
--- NOTE: This could easily be integrated into Stream:foreach(),
+-- Overrides Stream:iter()
+-- NOTE: This could easily be integrated into Stream:iter(),
 -- however this results in the loop to be unrolled explicitly
 -- for single-channel streams.
-function MuxStream:foreach(fnc)
+function MuxStream:iter()
 	local clear = table.clear
 
-	local ticks = {}
+	local ticks = table.new(#self.streams, 0)
 	for i = 1, #self.streams do
 		ticks[i] = self.streams[i]:gtick()
 	end
@@ -1289,17 +1292,19 @@ function MuxStream:foreach(fnc)
 	local channels = self.channels
 	local frame = table.new(channels, 0)
 
-	repeat
+	return function()
 		checkint()
 		clear(sampleCache)
 
 		for i = 1, channels do
 			frame[i] = ticks[i]()
 			-- Since all streams must have the same
-			-- length, if one ends all end
+			-- length, if one ends, all end
 			if not frame[i] then return end
 		end
-	until fnc(frame)
+
+		return frame
+	end
 end
 
 --- Mux several streams with the source stream into a multi-channel stream.
@@ -1522,7 +1527,7 @@ end
 
 function ConcatStream:gtick()
 	local i = 1
-	local ticks = {}
+	local ticks = table.new(#self.streams, 0)
 
 	for k = 1, #self.streams do
 		ticks[k] = self.streams[k]:gtick()
diff --git a/sndfile-stream.lua b/sndfile-stream.lua
index 1c394b8..ba5a0f3 100644
--- a/sndfile-stream.lua
+++ b/sndfile-stream.lua
@@ -113,7 +113,7 @@ function Stream:save(filename, format)
 
 	local frame_buffer = sndfile.frame_type(channels)
 
-	self:foreach(function(frame)
+	for frame in self:iter() do
 		-- NOTE: This should be (hopefully) automatically
 		-- unrolled for single-channel streams
 		-- Otherwise each loop copies an entire frame.
@@ -129,7 +129,7 @@ function Stream:save(filename, format)
 		-- (i.e. multichannel streams)
 		-- FIXME: Check return value
 		hnd:writef(frame_buffer)
-	end)
+	end
 
 	hnd:close()
 end