Connecting an input stream to an outputstream

Question

update in java9: https://docs.oracle.com/javase/9/docs/api/java/io/InputStream.html#transferTo-java.io.OutputStream-

I saw some similar, but not-quite-what-i-need threads.

I have a server, which will basically take input from a client, client A, and forward it, byte for byte, to another client, client B.

I'd like to connect my inputstream of client A with my output stream of client B. Is that possible? What are ways to do that?

Also, these clients are sending each other messages, which are somewhat time sensitive, so buffering won't do. I do not want a buffer of say 500 and a client sends 499 bytes and then my server holds off on forwarding the 500 bytes because it hasn't received the last byte to fill the buffer.

Right now, I am parsing each message to find its length, then reading length bytes, then forwarding them. I figured (and tested) this would be better than reading a byte and forwarding a byte over and over because that would be very slow. I also did not want to use a buffer or a timer for the reason I stated in my last paragraph — I do not want messages waiting a really long time to get through simply because the buffer isn't full.

What's a good way to do this?

Adao

Answer 1

Just because you use a buffer doesn't mean the stream has to fill that buffer. In other words, this should be okay:

public static void copyStream(InputStream input, OutputStream output)
    throws IOException
{
    byte[] buffer = new byte[1024]; // Adjust if you want
    int bytesRead;
    while ((bytesRead = input.read(buffer)) != -1)
    {
        output.write(buffer, 0, bytesRead);
    }
}

That should work fine - basically the read call will block until there's some data available, but it won't wait until it's all available to fill the buffer. (I suppose it could, and I believe FileInputStream usually will fill the buffer, but a stream attached to a socket is more likely to give you the data immediately.)

I think it's worth at least trying this simple solution first.

Justinajustine

Yes, I think this clears things up. I think I was getting confused with readFully() which does require the buffer to fill. – Adao

I have tried your code and I also tried reading message by message by reading the message's length then doing a byte[] buf = length; inputstream.read(buf)....the latter method was faster, and I'm not sure why. It seems to execute more lines of code yet it's faster. Almost 2x as fast. – Adao

Also, since my server has multiple sources and a single sink, I think I have to read messages within the multiple sources because simply reading and forwarding buffers may interleave messages between clients and scramble them. – Adao

@Jon Skeet Is there any way to use this without knowing the exact size of the byte array you will need? – Empirical

@Zibbobz: Any array size will work - the bigger it is, the fewer reads will be needed, but the more memory it takes while it's working. It's not like it has to be the actual length of the stream. – Justinajustine

@Jon Skeet This seems to throw a NullPointerException when I try it. Do you know if it is particularly un-suited for handling, say, PDF files? Because I happen to know for a fact that's what we're dealing with. – Empirical

@Zibbobz: No, it sounds like either input or output is null. You should be able to see that easily by debugging. – Justinajustine

Don't forget the flush() (and close()) your streams. – Erwin

@sgibly: that would be the caller's responsibility. – Justinajustine

@JonSkeet true for the close() calls, but I would definitely flush() the output stream here. It's too common to forget to do that. – Erwin

@sgibly: Well given that a close() will flush it anyway, I don't think it's worth it, personally. Of course, if you take code like this you should feel very free to add it :) – Justinajustine

@JonSkeet You are correct :-) I just got burnt before by not using flush, and I can see why that can happen - #2732760 – Erwin

Btw, the close() contract in the API doesn't say anything about flushing before closing. See docs.oracle.com/javase/8/docs/api/java/io/OutputStream.html, so if anything like that happens, it's very specific to the output stream implementation. – Erwin

@sgibly: I'd say that it's poorly documented rather than the intention being that everyone has to call flush... – Justinajustine

Answer 2

How about just using

void feedInputToOutput(InputStream in, OutputStream out) {
   IOUtils.copy(in, out);
}

and be done with it?

from jakarta apache commons i/o library which is used by a huge amount of projects already so you probably already have the jar in your classpath already.

Epiclesis

or just use the function itself, since calling another function with the exact same parameters is not needed.... – Earp

yes, that is what I do personally. I guess I only typed the extra method name as documentation but it's not needed. – Epiclesis

From what I can tell, this method is blocking until the while input is passed through. Therefore, this should be done in an async thread for the question asker. – Tavarez

Answer 3

JDK 9 has added InputStream#transferTo(OutputStream out) for this functionality.

Pommel

Answer 4

For completeness, guava also has a handy utility for this

ByteStreams.copy(input, output);

Debility

Answer 5

You can use a circular buffer :

Code

// buffer all data in a circular buffer of infinite size
CircularByteBuffer cbb = new CircularByteBuffer(CircularByteBuffer.INFINITE_SIZE);
class1.putDataOnOutputStream(cbb.getOutputStream());
class2.processDataFromInputStream(cbb.getInputStream());

Maven dependency

<dependency>
    <groupId>org.ostermiller</groupId>
    <artifactId>utils</artifactId>
    <version>1.07.00</version>
</dependency>

Mode details

http://ostermiller.org/utils/CircularBuffer.html

Cresting

Answer 6

Asynchronous way to achieve it.

void inputStreamToOutputStream(final InputStream inputStream, final OutputStream out) {
    Thread t = new Thread(new Runnable() {

        public void run() {
            try {
                int d;
                while ((d = inputStream.read()) != -1) {
                    out.write(d);
                }
            } catch (IOException ex) {
                //TODO make a callback on exception.
            }
        }
    });
    t.setDaemon(true);
    t.start();
}

Readymade

This is to transfer data from one stream to another without block your current thread. – Vernacularize

Answer 7

BUFFER_SIZE is the size of chucks to read in. Should be > 1kb and < 10MB.

private static final int BUFFER_SIZE = 2 * 1024 * 1024;
private void copy(InputStream input, OutputStream output) throws IOException {
    try {
        byte[] buffer = new byte[BUFFER_SIZE];
        int bytesRead = input.read(buffer);
        while (bytesRead != -1) {
            output.write(buffer, 0, bytesRead);
            bytesRead = input.read(buffer);
        }
    //If needed, close streams.
    } finally {
        input.close();
        output.close();
    }
}

Weidner

Should be a lot less than 10MB. This is TCP we're talking about. Any size greater than the socket receive buffer is completely pointless, and they are measured in kilobytes, not megabytes. – Expunge

Answer 8

Use org.apache.commons.io.IOUtils

InputStream inStream = new ...
OutputStream outStream = new ...
IOUtils.copy(inStream, outStream);

or copyLarge for size >2GB

Autography

Answer 9

This is a Scala version that is clean and fast (no stackoverflow):

  import scala.annotation.tailrec
  import java.io._

  implicit class InputStreamOps(in: InputStream) {
    def >(out: OutputStream): Unit = pipeTo(out)

    def pipeTo(out: OutputStream, bufferSize: Int = 1<<10): Unit = pipeTo(out, Array.ofDim[Byte](bufferSize))

    @tailrec final def pipeTo(out: OutputStream, buffer: Array[Byte]): Unit = in.read(buffer) match {
      case n if n > 0 =>
        out.write(buffer, 0, n)
        pipeTo(out, buffer)
      case _ =>
        in.close()
        out.close()
    }
  }

This enables to use > symbol e.g. inputstream > outputstream and also pass in custom buffers/sizes.

Urethrectomy

Could you provide some similar Java implementation? – Cookgeneral

@Luchostein: I was responding to the buggy Scala answer by George Pligor below – Urethrectomy

Answer 10

In case you are into functional this is a function written in Scala showing how you could copy an input stream to an output stream using only vals (and not vars).

def copyInputToOutputFunctional(inputStream: InputStream, outputStream: OutputStream,bufferSize: Int) {
  val buffer = new Array[Byte](bufferSize);
  def recurse() {
    val len = inputStream.read(buffer);
    if (len > 0) {
      outputStream.write(buffer.take(len));
      recurse();
    }
  }
  recurse();
}

Note that this is not recommended to use in a java application with little memory available because with a recursive function you could easily get a stack overflow exception error

Polygnotus

-1: How is a recursive Scala solution relevant to a Java question? – Gerladina

The method recurse is tail recursive. If you anotate it with @tailrec you wont have stack overflow problems. – Tungstite

This answer verifies that all pure java coders are suffering from the pressure of their bosses and need serious anger management! – Polygnotus