Copy (over a web API) is complete; I get a broken pipe anyway. How to solve it?

Question

For one of my projects, I implement a Java 7 FileSystem over the Box API Java SDK (the new one).

However, for downloading files, when you want to have a stream to the content, it only provides methods taking OutputStream as an argument; specifically, I am using this one at the moment.

But this doesn't sit well with the JDK API; I need to be able to implement FileSystemProvider#newInputStream()... Therefore I elected to use Pipe{Input,Output}Stream.

Moreover, since the Box SDK API methods are synchronous (not that it matters here), I wrap them in a Future. My code is as follows (imports ommitted for brevity):

@ParametersAreNonnullByDefault
public final class BoxFileInputStream
    extends InputStream
{
    private final Future<Void> future;
    private final PipedInputStream in;

    public BoxFileInputStream(final ExecutorService executor,
        final BoxFile file)
    {
        in = new PipedInputStream(16384);
        future = executor.submit(new Callable<Void>()
        {
            @Override
            public Void call()
                throws IOException
            {
                try {
                    file.download(new PipedOutputStream(in));
                    return null;
                } catch (BoxAPIException e) {
                    throw BoxIOException.wrap(e);
                }
            }
        });
    }

    @Override
    public int read()
        throws IOException
    {
        try {
            return in.read();
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public int read(final byte[] b)
        throws IOException
    {
        try {
            return in.read(b);
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public int read(final byte[] b, final int off, final int len)
        throws IOException
    {
        try {
            return in.read(b, off, len);
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public long skip(final long n)
        throws IOException
    {
        try {
            return in.skip(n);
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public int available()
        throws IOException
    {
        try {
            return in.available();
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public void close()
        throws IOException
    {
        IOException streamException = null;
        IOException futureException = null;

        try {
            in.close();
        } catch (IOException e) {
            streamException = e;
        }

        try {
            future.get(5L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            futureException = new BoxIOException("donwload interrupted", e);
        } catch (ExecutionException e) {
            futureException = new BoxIOException("download failure",
                e.getCause());
        } catch (CancellationException e) {
            futureException = new BoxIOException("download cancelled", e);
        } catch (TimeoutException e) {
            futureException = new BoxIOException("download timeout", e);
        }

        if (futureException != null) {
            if (streamException != null)
                futureException.addSuppressed(streamException);
            throw futureException;
        }

        if (streamException != null)
            throw streamException;
    }

    @Override
    public synchronized void mark(final int readlimit)
    {
        in.mark(readlimit);
    }

    @Override
    public synchronized void reset()
        throws IOException
    {
        try {
            in.reset();
        } catch (IOException e) {
            future.cancel(true);
            throw new BoxIOException("download failure", e);
        }
    }

    @Override
    public boolean markSupported()
    {
        return in.markSupported();
    }
}

The code consistenly fails with the following stack trace (that is in int read(byte[]):

Exception in thread "main" com.github.fge.filesystem.box.exceptions.BoxIOException: download failure
    at com.github.fge.filesystem.box.io.BoxFileInputStream.read(BoxFileInputStream.java:81)
    at java.nio.file.Files.copy(Files.java:2735)
    at java.nio.file.Files.copy(Files.java:2854)
    at java.nio.file.CopyMoveHelper.copyToForeignTarget(CopyMoveHelper.java:126)
    at java.nio.file.Files.copy(Files.java:1230)
    at Main.main(Main.java:37)
    [ IDEA specific stack trace elements follow -- irrelevant]
Caused by: java.io.IOException: Pipe broken
    at java.io.PipedInputStream.read(PipedInputStream.java:322)
    at java.io.PipedInputStream.read(PipedInputStream.java:378)
    at java.io.InputStream.read(InputStream.java:101)
    at com.github.fge.filesystem.box.io.BoxFileInputStream.read(BoxFileInputStream.java:78)
    ... 10 more

But when it fails, the download is already complete...

OK, the thing is, I can grab the file size and hack around it but I'd prefer not to if at all possible; how can I modify this code so as to avoid EPIPE?

Answer 1

The SDK also provides BoxAPIRequest and BoxAPIResponse classes that let you make manual requests for advanced use-cases. These classes still automatically handle authentication, errors, back-off, etc. but give you more granular control over the request.

In your case, you could do make a download request manually by doing:

// Note: this example assumes you already have a BoxAPIConnection.
URL url = new URL("files/" + file.getID() + "/content")
BoxAPIRequest request = new BoxAPIRequest(api, url, "GET");
BoxAPIResponse response = request.send();

InputStream bodyStream = response.getBody();
// Use the stream.
response.disconnect();