file.lua

local fs = require('fs')
local iStream = require('stream').iStream

local file = {}

local ReadStream = iStream:extend()
file.ReadStream = ReadStream

function ReadStream:initialize(fd)
  self.fd = fd
  self.offset = 0
  self.chunkSize = 10
end

function ReadStream:read(callback)
  fs.read(self.fd, self.offset, self.chunkSize, function (err, chunk)
    if err then
      return callback(err)
    end
    local bytesRead = #chunk
    if bytesRead == 0 then
      return callback()
    end
    self.offset = self.offset + bytesRead
    callback(nil, chunk)
  end)
end

local WriteStream = iStream:extend()
file.WriteStream = WriteStream

function WriteStream:initialize(fd)
  self.fd = fd
  self.offset = 0
end

function WriteStream:write(chunk, callback)
  if not chunk then
    return callback()
  end
  fs.write(self.fd, self.offset, chunk, function (err, bytesWritten)
    if err then
      return callback(err)
    end
    self.offset = self.offset + bytesWritten
    callback()
  end)
end

function file.createReadStream(path, callback)
  fs.open(path, "r", "0666", function (err, fd)
    if err then return callback(err) end
    local stream = ReadStream:new(fd)
  end)
end

return file

stream.lua

local Emitter = require('core').Emitter

local stream = {}

local iStream = Emitter:extend()
stream.iStream = iStream

-- TODO: close input and output streams on error
function iStream:pipe(dest, callback)
  local consume, onRead
  consume = function (err)
    if err then return callback(err) end
    self:read(onRead)
  end
  onRead = function (err, chunk)
    if err then return callback(err) end
    return dest:write(chunk, chunk and consume or callback)
  end
  consume()
end

return stream

test.lua

local file = require('file')
local fs = require('fs')
local newFiber = require('fiber').new

local function copy(input, output, callback)
  fs.open(input, "r", "0666", function (err, fd)
    if err then return callback(err) end
    local input = file.ReadStream:new(fd)
    fs.open(output, "w", "0666", function (err, fd)
      if err then return callback(err) end
      local output = file.WriteStream:new(fd)
      input:pipe(output, callback)
    end)
  end)
end

copy(__filename, __filename .. ".out", function (err)
  if (err) then error(err) end
  p("done")
end)

local function scopy(input, output, wrap)
  local open, readable, writable, chunk
  open = wrap(fs.open, true)
  readable = file.ReadStream:new(open(input, "r", "0644"))
  writable = file.WriteStream:new(open(output, "w", "0644"))
  readable.read = wrap(readable.read, true)
  writable.write = wrap(writable.write, true)
  repeat
    chunk = readable:read()
    writable:write(chunk)
  until not chunk
end

newFiber(function (wrap)
  scopy(__filename, __filename .. ".out2", wrap)
  p("done2")
end)