发送数据

Question

这是我刚才的问题：Send big file over reactive stream

我设法用发送文件在阿卡流FileIO.fromPath（Paths.get（file.toURI（））），它工作正常。但是，我想在发送文件之前对其进行压缩和加密。我创建了方法，打开FileInputStream，通过压缩流，然后通过加密流路由它，现在我想使用Akka流将其引导到套接字中。

文件 - >的FileInputStream - > CompressedInputStream - > EncryptedInputStream？ - >春/阿卡流

的一点是，我可以即时而读它由一块块压缩/加密文件（我不是在磁盘上创建额外的文件），并且我不知道如何通过Akka/Spring流发送InputStream（压缩和加密）（Spring流，我的意思是项目Reactor API下的Akka流）。

问题是：如何压缩，加密和发送文件，而无需将整个压缩/加密文件保存到磁盘上？

Answer 1

我有（意外）发现：StreamConverters.fromInputStream！

http://doc.akka.io/docs/akka/2.4.17/java/stream/stages-overview.html#fromInputStream

Answer 2

实际上，有一个专门用于处理输入流等资源的源代码。这就是所谓的Source.unfoldResource：

Source<ByteString, NotUsed> source = Source.unfoldResource(
    () -> prepareEncryptedStream(), 
    is -> readChunk(is, 4096), 
    InputStream::close 
); 

Optional<ByteString> readChunk(InputStream is, int size) throws IOException { 
    byte[] data = new byte[size]; 
    int read = is.read(data); 
    if (read < 0) { 
     return Optional.empty(); 
    } 
    return Optional.of(ByteString.fromArray(data, 0, read)); 
} 

InputStream prepareEncryptedStream() { ... } 

这里prepareCompressedFile()是应该回到你想创建一个反应流加密流的方法，并readChunk()是一个便捷方法读取指定大小的InputStream一个ByteString。

如果您可以将压缩和加密例程表示为ByteString -> ByteString函数，那么您并不需要这些;所有你需要做的就是将这些程序传递给map()流量：

Source<ByteString, CompletionStage<IOResult>> source = 
    FileIO.fromPath(Paths.get("somewhere")) 
     .map(bs -> compress(bs)) 
     .map(bs -> encrypt(bs)); 

ByteString encrypt(ByteString bs) { ... } 

ByteString compress(ByteString bs) { ... }