public Response get ( HttpClientRequest < ByteBuf > req , long timeout , TimeUnit timeUnit ) { final String uri = req . getUri ( ) ; final Response result = new Response ( ) ; try { final Func1 < HttpClientResponse < ByteBuf > , Observable < byte [ ] > > process = response -> { result . status = response . getStatus ( ) . code ( ) ; result . headers = response . getHeaders ( ) ; final Func2 < ByteArrayOutputStream , ByteBuf , ByteArrayOutputStream > accumulator = ( baos , bb ) -> { try { bb . readBytes ( baos , bb . readableBytes ( ) ) ; } catch ( IOException e ) { throw new RuntimeException ( e ) ; } return baos ; } ; return response . getContent ( ) . reduce ( new ByteArrayOutputStream ( ) , accumulator ) . map ( ByteArrayOutputStream :: toByteArray ) ; } ; result . body = rxHttp . submit ( req ) . flatMap ( process ) . subscribeOn ( Schedulers . io ( ) ) . toBlocking ( ) . toFuture ( ) . get ( timeout , timeUnit ) ; return result ; } catch ( Exception e ) { throw new RuntimeException ( "failed to get url: " + uri , e ) ; } }
Perform an HTTP get in the allowed time .