java 如何通过管道将 OutputStream 传送到 StreamingDataHandler？

How can you pipe an OutputStream to a StreamingDataHandler?

提问by Dan Lew

I've got a Java web service in JAX-WS that returns an OutputStream from another method. I can't seem to figure out how to stream the OutputStream into the returned DataHandler any other way than to create a temporary file, write to it, then open it back up again as an InputStream. Here's an example:

我在 JAX-WS 中有一个 Java Web 服务，它从另一个方法返回一个 OutputStream。除了创建一个临时文件，写入它，然后再将它作为 InputStream 重新打开之外，我似乎无法弄清楚如何将 OutputStream 流式传输到返回的 DataHandler 中。下面是一个例子：

@MTOM
@WebService
class Example {
    @WebMethod
    public @XmlMimeType("application/octet-stream") DataHandler service() {
        // Create a temporary file to write to
        File fTemp = File.createTempFile("my", "tmp");
        OutputStream out = new FileOutputStream(fTemp);

        // Method takes an output stream and writes to it
        writeToOut(out);
        out.close();

        // Create a data source and data handler based on that temporary file
        DataSource ds = new FileDataSource(fTemp);
        DataHandler dh = new DataHandler(ds);
        return dh;
    }
}

The main issue is that the writeToOut() method can return data that are far larger than the computer's memory. That's why the method is using MTOM in the first place - to stream the data. I can't seem to wrap my head around how to stream the data directly from the OutputStream that I need to provide to the returned DataHandler (and ultimately the client, who receives the StreamingDataHandler).

主要问题是 writeToOut() 方法可以返回远大于计算机内存的数据。这就是该方法首先使用 MTOM 的原因 - 流式传输数据。我似乎无法理解如何直接从我需要提供给返回的 DataHandler（以及接收 StreamingDataHandler 的客户端）的 OutputStream 中传输数据。

I've tried playing around with PipedInputStream and PipedOutputStream, but those don't seem to be quite what I need, because the DataHandler would need to be returned after the PipedOutputStream is written to.

我尝试过使用 PipedInputStream 和 PipedOutputStream，但这些似乎并不是我所需要的，因为在写入 PipedOutputStream 之后需要返回 DataHandler。

Any ideas?

有任何想法吗？

采纳答案by Dan Lew

I figured out the answer, along the lines that Christian was talking about (creating a new thread to execute writeToOut()):

我按照 Christian 所说的思路找到了答案（创建一个新线程来执行 writeToOut()）：

@MTOM
@WebService
class Example {
    @WebMethod
    public @XmlMimeType("application/octet-stream") DataHandler service() {
        // Create piped output stream, wrap it in a final array so that the
        // OutputStream doesn't need to be finalized before sending to new Thread.
        PipedOutputStream out = new PipedOutputStream();
        InputStream in = new PipedInputStream(out);
        final Object[] args = { out };

        // Create a new thread which writes to out.
        new Thread(
            new Runnable(){
                public void run() {
                    writeToOut(args);
                    ((OutputStream)args[0]).close();
                }
            }
        ).start();

        // Return the InputStream to the client.
        DataSource ds = new ByteArrayDataSource(in, "application/octet-stream");
        DataHandler dh = new DataHandler(ds);
        return dh;
    }
}

It is a tad more complex due to finalvariables, but as far as I can tell this is correct. When the thread is started, it blocks when it first tries to call out.write(); at the same time, the input stream is returned to the client, who unblocks the write by reading the data. (The problem with my previous implementations of this solution was that I wasn't properly closing the stream, and thus running into errors.)

由于final变量，它有点复杂，但据我所知这是正确的。当线程启动时，它在第一次尝试调用时阻塞out.write()；同时，输入流返回给客户端，客户端通过读取数据解除写入阻塞。（我之前实现这个解决方案的问题是我没有正确关闭流，因此遇到了错误。）

回答by Christian

Sorry, I only did this for C# and not java, but I think your method should launch a thread to run "writeToOut(out);" in parralel. You need to create a special stream and pass it to the new thread which gives that stream to writeToOut. After starting the thread you return that stream-object to your caller.

抱歉，我只为 C# 而不是 Java 做了这个，但我认为你的方法应该启动一个线程来运行“writeToOut(out);” 在并行。您需要创建一个特殊的流并将其传递给将该流提供给 writeToOut 的新线程。启动线程后，您将该流对象返回给您的调用者。

If you only have a method that writes to a stream and returns afterwards and another method that consumes a stream and returns afterwards, there is no other way.

如果您只有一个写入流并随后返回的方法和另一个消耗流并随后返回的方法，则没有其他方法。

Of coure the tricky part is to get hold of such a -multithreading safe- stream: It shall block each side if an internal buffer is too full.

当然，棘手的部分是获得这样的多线程安全流：如果内部缓冲区太满，它将阻塞每一方。

Don't know if a Java-pipe-stream works for that.

不知道 Java-pipe-stream 是否适用于此。

回答by Mihai Toader

Wrapper pattern ? :-).

包装图案 ? :-)。

Custom javax.activation.DataSource implementation (only 4 methods) to be able to do this ?

自定义 javax.activation.DataSource 实现（只有 4 个方法）能够做到这一点？

return new DataHandler(new DataSource() { 
  // implement getOutputStream to return the stream used inside writeToOut() 
  ... 
});   

I don't have the IDE available to test this so i'm only doing a suggestion. I would also need the writeToOut general layout :-).

我没有可用的 IDE 来测试这个，所以我只是做一个建议。我还需要 writeToOut 总体布局 :-)。

回答by yurin

In my application I use InputStreamDataSource implementation that take InputStream as constructor argument instead of File in FileDataSource. It works so far.

在我的应用程序中，我使用 InputStreamDataSource 实现，它将 InputStream 作为构造函数参数而不是 FileDataSource 中的 File。到目前为止它有效。

public class InputStreamDataSource implements DataSource {

ByteArrayOutputStream buffer = new ByteArrayOutputStream();
private final String name;

public InputStreamDataSource(InputStream inputStream, String name) {
    this.name = name;
    try {
        int nRead;
        byte[] data = new byte[16384];
        while ((nRead = inputStream.read(data, 0, data.length)) != -1) {
            buffer.write(data, 0, nRead);
        }

        buffer.flush();
        inputStream.close();
    } catch (IOException e) {
        e.printStackTrace();
    }

}

@Override
public String getContentType() {
    return new MimetypesFileTypeMap().getContentType(name);
}

@Override
public InputStream getInputStream() throws IOException {
    return new ByteArrayInputStream(buffer.toByteArray());
}

@Override
public String getName() {
    return name;
}

@Override
public OutputStream getOutputStream() throws IOException {
    throw new IOException("Read-only data");
}

}

}