Working with in the manual.

src/ftp.lua

@@ -114,7 +114,7 @@ function metat.__index:send(sendt)
         if err then data:close() end
         return ret, err
     end
-    local sink = socket.sink("close-when-empty", data)
+    local sink = socket.sink("close-when-done", data)
     socket.try(ltn12.pump.all(sendt.source, sink, checkstep))
     if string.find(code, "1..") then socket.try(self.tp:check("2..")) end
     return 1

src/smtp.lua

@@ -51,10 +51,10 @@ function metat.__index:rcpt(to)
     return socket.try(self.tp:check("2.."))
 end
 
-function metat.__index:data(src)
+function metat.__index:data(src, step)
     socket.try(self.tp:command("DATA"))
     socket.try(self.tp:check("3.."))
-    socket.try(self.tp:source(src))
+    socket.try(self.tp:source(src, step))
     socket.try(self.tp:send("\r\n.\r\n"))
     return socket.try(self.tp:check("2.."))
 end
@@ -78,7 +78,7 @@ function metat.__index:send(mailt)
     else
         self:rcpt(mailt.rcpt)
     end
-    self:data(ltn12.source.chain(mailt.source, stuff()))
+    self:data(ltn12.source.chain(mailt.source, stuff()), mailt.step)
 end
 
 function open(server, port)

src/tp.lua

@@ -87,13 +87,9 @@ function metat.__index:getcontrol()
     return self.control
 end
 
-function metat.__index:source(src, instr)
-    while true do
-        local chunk, err = src()
-        if not chunk then return not err, err end
-        local ret, err = self.control:send(chunk)
-        if not ret then return nil, err end
-    end
+function metat.__index:source(source, step)
+    local sink = socket.sink("keep-open", self.control)
+    return ltn12.pump.all(source, sink, step or ltn12.pump.step)
 end
 
 -- closes the underlying control