如何通过管道将 OutputStream 传递给 StreamingDataHandler？

Question

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

@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;
    }
}

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

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

有什么想法吗？

Answer 1

我按照 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;
    }
}

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

Answer 2

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

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

当然，棘手的部分是掌握这样一个多线程安全流：如果内部缓冲区太满，它会阻塞每一端。

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

Answer 3

包装模式？ :-)。

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

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

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

Answer 4

在我的应用程序中，我使用 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");
}

}