Is it possible to ask for connection kept open for some limited time

[samadpaydar3]

Hi,

I have this code

        AtomicReference<InputStream> result = new AtomicReference<>();
        Consumer<InputStream> consumer = inputStream -> {
            result.set(inputStream);
        };

        Unirest
                .get("http://universities.hipolabs.com/search?country=United+States")
                .thenConsume(rawResponse -> consumer.accept(rawResponse.getContent()));
        InputStream inputStream = result.get();
        read(inputStream);

The idea is to use consumers to let the response's input stream be used/read by other part/layer of code. So, the consumer above uses AtomicReference to mimic returning the response's inputStream.

The read method is as below:

    private void read(InputStream inputStream) {
        try {
            byte[] buffer = new byte[4096];
            while (true) {
                int count = inputStream.read(buffer);
                int available = inputStream.available();
                System.out.println("Read " + count + " bytes, available " + available);

                if (count == -1) {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

and this is the output of the first code:

Read 2530 bytes, available 0
java.net.SocketException: Socket closed
    at java.base/sun.nio.ch.NioSocketImpl.ensureOpenAndConnected(NioSocketImpl.java:165)
    at java.base/sun.nio.ch.NioSocketImpl.beginRead(NioSocketImpl.java:231)
    at java.base/sun.nio.ch.NioSocketImpl.implRead(NioSocketImpl.java:299)
    at java.base/sun.nio.ch.NioSocketImpl.read(NioSocketImpl.java:350)
    at java.base/sun.nio.ch.NioSocketImpl$1.read(NioSocketImpl.java:803)
    at java.base/java.net.Socket$SocketInputStream.read(Socket.java:976)
    at org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
    at org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197)
    at org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176)
    at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135)
    at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:148)

So, the problem is when reading from the inputStream that is returned by the consumer, only one chunk of the response is read, and next read attempts fails, as socket is closed.

Now, is there any way (e.g. config) to ask unirest keep the connection open for a while (e.g. 1 minute)?

I have used configurations like below, but it didn't work. Unirest.config().socketTimeout(100000);

P.S: if the consumer directly reads the inputStream (like the code below), there's no problem and it can read the whole response (412,223 bytes). But for some design issue, I need the consumer to return/expose the inputStream to other part of code.

        Consumer<InputStream> consumer = inputStream -> {
            read(inputStream);
        };

        Unirest
                .get("http://universities.hipolabs.com/search?country=United+States")
                .thenConsume(rawResponse -> consumer.accept(rawResponse.getContent()));

Another point is that if I add a delay before reading from the returned inputStream (like below), the first chunk of the data is still read.

        AtomicReference<InputStream> result = new AtomicReference<>();
        Consumer<InputStream> consumer = inputStream -> {
            result.set(inputStream);
        };
        Unirest
                .get("http://universities.hipolabs.com/search?country=United+States")
                .thenConsume(rawResponse -> {
                    consumer.accept(rawResponse.getContent());
                });
        InputStream inputStream = result.get();
        try {
            Thread.sleep(10000);
        } catch (Exception e) {
            e.printStackTrace();
        }
        read(inputStream);

[ryber]

The problem is not the connection or the socket exactly. It's that thenConsume expect all the work to be done inside itself. Once thenConsume ends it closes the InputStream. The whole thenConsume method was really designed for very large responses where getting direct access to the InputStream was important. I suppose you could copy the input stream but that would be a blocking operation and you would end up with something like a ByteArrayInputStream with ALL the bytes in memory (which is what thenConsume was avoiding).

Not closing the InputStream is risky as people could end up with a lot of leakage if they don't close them properly. If you want that much control over it I would recommend using Apache HttpClient or the HttpClient in java 9+ native.

[samadpaydar3]

Thank you @ryber for the quick and helpful response.

[samadpaydar3]

Do you think my use case (returninig/exposing the response's input stream into downstream code) can be implemented using other methods (I mean not using thenConsume, but using something like asObject). I believe the same problem (socket being closed) applies to other methods. Correct?

[ryber]

Well, why do you want an InputStream? are your responses really that big? If not then reading the entire response and returning a supplier with the result should be fine.

[samadpaydar3]

Yes. I think the only way is to read the entire response in the consumer.

For my question about using other methods, like asObject, I noticed that they don't work, as you have mentioned here https://github.com/Kong/unirest-java/issues/303 the response's body can't be cast to InputStream.

Thanks for your help.

[ryber]

the consumes method works with the raw response before its been read. Thats why things like asObject aren't there. It really has a specific use case, if you are ok with reading the entire response then your method can just return a Supplier<?> or something and you can populate it with the completed response

[samadpaydar3]

Yes. Thank you for your help. I will move forward with reading the entire response.