local chunk = handle:read(2048) if not chunk then handle:close() end return chunk end
else return source.empty(io_err or \) end end
一个source每次调用时返回的下一个数据块。当有没有更多的数据时返回nil。如果有错误，source可以返回nill和一个错误消息。 阿德里安发现，虽然不是刻意为之，source接口与Lua5.0的迭代器的思想兼容。也就是说，数据source可以很好地与循环结合使用。把文件源作为一个迭代器，重写第一个例子： local process = normalize(\) for chunk in source.file(io.stdin) do io.write(process(chunk)) end
io.write(process(nil))
注意filter最后一次调用获得最后处理过的数据块。循环结束时source返回了nil。因此我们需要在循环外最后再调用一次。 4 在调用之间保持状态
通常情况下，source在某些事件发生后，需要改变自己的行为。一个简单的例子是，一个文件source，当到达文件末端时，无论他被调用多少次都必须保证返回nil。从而避免读取超过文件末端。
实现这种source的方式是创建一个工厂，带有一个额外的状态变量。source可以通过词法作用域使用它。在检测到到达文件末端时，文件source可以设置自己的文件句柄为nil。此后所有的source被调用时都检查句柄是否有效，做出响应操作。
function source.file(handle, io_err) if handle then
return function()
if not handle then return nil end local chunk = handle:read(2048) if not chunk then handle:close() handle = nil end
return chunk end else
return source.empty(io_err or \) end end
另一个实现方式是改变source的接口，让它更灵活一些。允许source返回一个数据块之外的第二个值。如果返回的数据块是nil，额外的值告诉我们发生了什么。第二个nil表示没有数据了，source已经空了。其他的取值可以是一个错误信
6
息。另一方面，如果数据块不是nil，第二个返回值表示source是否需要更换。如果第二个返回值不是nil，继续使用同一个source，否则这就是另一个source，我需要获取剩下的数据。
这个额外的自由对于编写source函数的人来说是很好的。但是对使用它的用户来说有困难。幸运的是，下面的fancy source能够把它转化成简单的source，并不需要替换，使用下面这个工厂: function source.simplify(src) return function()
local chunk, err_or_new = src() --或者是新的src返回，否则返回同样的src -- 注意无论怎样src都返回一个src，要不是错误的src，要不是新的src，或者是原来的src
src = err_or_new or src – 更新自身
if not chunk then return nil, err_or_new – chunk为空，返回错误的或者新的src
else return chunk end end end
这个简化转换工厂允许我们编写一个fancy source，却使用简单source来实现。因此，接下来的函数只是产生一个简单source，接受source的函数认为他是一个简单source。
回到文件source的例子，接口的扩展带来了更优雅的实现。在没有数据时，这个新的source替换为一个空source，不需要反复检查句柄了。 function source.file(handle, io_err) if handle then
return source.simplify(function() local chunk = handle:read(2048) if not chunk then handle:close()
return \, source.empty() end
return chunk end) else
return source.empty(io_err or \) end end
如果我们使用coroutine协程，lua5.0的新特性，这个方法可以更强大。协程的资料比较少，就像一个词法作用域，协程起初看起来很怪，但是一旦你适应了这个概念，可以节省你的大量时间。我得承认使用协程来实现file source是多余的。下面来看如何实现一个链接source的工厂 function source.cat(...)
local co = coroutine.create(function() local i = 1
while i <= table.getn(arg) do local chunk, err = arg[i]()
7
if chunk then coroutine.yield(chunk) elseif err then return nil, err else i = i + 1 end end end)
return function()
return shift(coroutine.resume(co)) end end
这个工厂创建了2个函数，第一个是一个辅助函数，以协程的方式完成了所有的工作。它从一个source中读取一个数据块。如果数据块是nil，移动到下一个source，否则它调用yield返回数据块。第二个函数是source自身，只是调用resume执行辅助的协程，无论返回怎样的数据块都返回给用户（跳过了第一个返回值，这个值告诉我们协程是否已经终止）。思考一下，当你编写同样的函数而不是用协程，你就会体会到如此实现的简单性。在我们要把filter接口变得更加强大时，将要再次使用协程。 5 链接source
用filter链接一个source意味着什么？最有用的解释是链接source-filter后形成一个新的source。source产生数据在返回前传递给filter，这里的工厂例子： function source.chain(src, f)
return source.simplify(function() local chunk, err = src()
if not chunk then return f(nil), source.empty(err) else return f(chunk) end end) end
考虑一个函数，从一个source获得输入数据。通过链接一个简单的source跟着一个或者多个filter。这个相同的函数能够得到处理过的数据，无需考虑这个数据是通过背后的filter得到的。
6 sinks
正如我们定义了最初的source接口，同样我们定一个sink接口。把所有遵循接口的函数称为sink。下面2个简单的工厂返回了sink。table工厂创建了一个sink表，保存了所有获得的数据。这些数据之后能有效的链接成一个单一的string，调用table.concat库函数即可。另一个例子介绍了一个空sink，直接丢弃接受到的数据。
function sink.table(t) t = t or {}
local f = function(chunk, err)
if chunk then table.insert(t, chunk) end
8
return 1 end
return f, t end
local function null() return 1 end
function sink.null() return null end
sink 接受连续的数据块直到数据结束，通过返回一个nil块表示。错误通过额外的参数返回，在nil数据块之后跟随一个错误消息。如果一个sink检测到一个自身的错误，并且不准备被再次调用，应该返回nil，之后的错误消息是可选的。非nil的返回值表示source将会接受更多的数据。最后，正如source可以选择被替换，sink也可以，只要接口一致。同样的，实现一个sink.simplify工厂是很简单的。可以用它转换一个fancy sink变成一个简单的sink。
作为一个例子，创建一个source从标准输入读取数据，链接到一个标准化行结束转换的filter，使用一个sink把所有数据放到一个表中，最后打印出来。 local load = source.chain(source.file(io.stdin), normalize(\)) local store, t = sink.table() while 1 do
local chunk = load()
store(chunk)
if not chunk then break end end
print(table.concat(t))
有一次，正如我们用source和filter创建一个链接source-filter的工厂，很容易用sink和filter创建一个工厂产生新的sink。这个新的sink在调用原始的sink之前，通过filter处理所有的接受到的数据。如此实现： function sink.chain(f, snk) return function(chunk, err) local r, e = snk(f(chunk))
if not r then return nil, e end
if not chunk then return snk(nil, err) end return 1 end end 7 Pumps
在我们的例子中一直存在一个循环，到目前为止我们设计的东西都是被动式的，因此每次都有这个循环，Sources, sinks, filters:没有一个是靠它们自己工作的。把数
9
据从source推送到sink的工作是一个普遍的要求，我们将提供2个辅助函数来做这些工作。
function pump.step(src, snk) local chunk, src_err = src()
local ret, snk_err = snk(chunk, src_err)
return chunk and ret and not src_err and not snk_err, src_err or snk_err end
function pump.all(src, snk, step) step = step or pump.step while true do
local ret, err = step(src, snk)
if not ret then return not err, err end end end
pump.step 函数把一个数据块从source传递到sink。pump.all 函数调用一个可选的step函数，调用step函数推送所有的数据到sink。现在使用所有的东西，编写一个程序。使用Base64编码方式，从磁盘的二进制文件中读取数据保存到另一个文件中。
local load = source.chain(
source.file(io.open(\, \)), encode(\) )
local store = sink.chain( wrap(76),
sink.file(io.open(\, \)), )
pump.all(load, store)
我们切分的filter并不符合直觉，是人为故意的。作为一种选择，我们可以把Base64编码filter和行包裹filter链接起来，然后链接这个复合filter到文件source或者文件sink上，这无关紧要。
8 最后的重要变更
事实证明我们依然有一个问题，David Burgess编写了一个gzip filter，他发现解压filter把一个小的输入数据扩大为大量的数据。虽然我们希望可以忽视这个问题，但是得承认不可以。唯一的解决方式是允许filter返回部分结果，这也是我们选择的方法。在调用这个filter处理输入数据之后，用户不得不循环调用这个filter询问是否还有更多的输出。注意这些额外的调用不能把更多的数据发送到filter。 更加特殊的情况是，传递一个输入数据块到filter之后搜集第一个输出数据块。用户重复调用filter，传递空字符串，获得额外的输出数据块。当filter自身返回一个空串时，用户知道没有输出数据了，然后接着处理下一个输入数据块。最后，在用户传递一个nill告知filter没有更多的输入数据了，filter仍然会在一个数据块中返
10