EventMachine 分块读写文件

Question

我在 REST API 服务器中使用 EventMachine 和 EM-Synchrony。当接收到正文中包含大型二进制文件的 POST 请求时，我以块的形式接收它，将这些块写入 Tempfile，而不是阻塞反应器。

然后，在某些时候，我需要分块读取此文件并将这些块写入最终文件。这是可行的，但它会按预期阻塞反应器，并且无法找到一种方法使其在不阻塞的情况下工作。

我有时会调用这个函数，将临时文件和新文件名传递给它：

def self.save(tmp_file, new_file)
   tmp = File.open(tmp_file, "rb")
   newf = File.open(new_file, "wb")
   md5  = Digest::MD5.new

   each_chunk(tmp, CHUNKSIZE) do |chunk|
     newf << chunk
     md5.update chunk
   end

   md5.hexdigest
end

def self.each_chunk(file, chunk_size=1024)
  yield file.read(chunk_size) until file.eof?
end

我一直在 StackOverflow 上阅读所有其他类似的问题，尝试使用 EM#next_tick，这可能是解决方案（没有那么多 EM 经验）但不能让它工作，也许我把它放在错误的地方。

另外，我尝试过 EM#defer，但我需要该函数在返回 md5 之前等待读/写过程完成，就像在我的主文件中一样，在调用此函数后，我使用返回值。

如果有人可以帮助我，我将不胜感激。

编辑 1

我需要保存函数仅在完成文件读/写后返回，因为在调用函数中我正在等待最终的 md5 值，如下所示：

def copy_and_update(...)
  checksum = SomeModule.save(temp_file, new_file)
  do_database_update({:checksum => checksum}) # only with the final md5 value
end

Answer 1

你需要在里面注入一些东西来分解它：

def self.each_chunk(file, chunk_size=1024)
  chunk_handler = lambda {
    unless (file.eof?)
      yield file.read(chunk_size)


     EM.next_tick(&chunk_handler)
    end
  }

  EM.next_tick(&chunk_handler)
end

这样做有点麻烦，但这是异步编程。