common/src/main/java/HTTPClient/StreamDemultiplexor.java
/**
     * reads an array of bytes from the master stream.
     */
reads an array of bytes from the master stream.
reads an array of bytes from the master stream .
int read(byte[] b, int off, int len, ResponseHandler resph, int timeout)
	    throws IOException
    {
	if (resph.exception != null)
	{
	    resph.exception.fillInStackTrace();
	    throw resph.exception;
	}
	if (resph.eof)
	    return -1;
	ResponseHandler head;
	while ((head = (ResponseHandler) RespHandlerList.getFirst()) != null  &&
		head != resph)
	{
	    try
		{ head.stream.readAll(timeout); }
	    catch (IOException ioe)
	    {
		if (ioe instanceof InterruptedIOException)
		    throw ioe;
		else
		{
		    resph.exception.fillInStackTrace();
		    throw resph.exception;
		}
	    }
	}
	synchronized (this)
	{
	    if (resph.exception != null)
	    {
		resph.exception.fillInStackTrace();
		throw resph.exception;
	    }
	    if (resph.resp.cd_type != CD_HDRS)
		Log.write(Log.DEMUX, "Demux: Reading for stream " +
				     resph.stream.hashCode());
	    if (Timer != null)  Timer.hyber();
	    try
	    {
		int rcvd = -1;
		if (timeout != cur_timeout)
		{
		    Log.write(Log.DEMUX, "Demux: Setting timeout to " +
					 timeout + " ms");
		    Sock.setSoTimeout(timeout);
		    cur_timeout = timeout;
		}
		switch (resph.resp.cd_type)
		{
		    case CD_HDRS:
			rcvd = Stream.read(b, off, len);
			if (rcvd == -1)
			    throw new EOFException("Premature EOF encountered");
			break;
		    case CD_0:
			rcvd = -1;
			close(resph);
			break;
		    case CD_CLOSE:
			rcvd = Stream.read(b, off, len);
			if (rcvd == -1)
			    close(resph);
			break;
		    case CD_CONTLEN:
			int cl = resph.resp.ContentLength;
			if (len > cl - resph.stream.count)
			    len = cl - resph.stream.count;
			rcvd = Stream.read(b, off, len);
			if (rcvd == -1)
			    throw new EOFException("Premature EOF encountered");
			if (resph.stream.count+rcvd == cl)
			    close(resph);
			break;
		    case CD_CHUNKED:
			if (chunk_len == -1)	
			    chunk_len = Codecs.getChunkLength(Stream);
			if (chunk_len > 0)		
			{
			    if (len > chunk_len)  len = (int) chunk_len;
			    rcvd = Stream.read(b, off, len);
			    if (rcvd == -1)
				throw new EOFException("Premature EOF encountered");
			    chunk_len -= rcvd;
			    if (chunk_len == 0)	
			    {
				Stream.read();	
				Stream.read();	
				chunk_len = -1;
			    }
			}
			else	
			{
			    resph.resp.readTrailers(Stream);
			    rcvd = -1;
			    close(resph);
			    chunk_len = -1;
			}
			break;
		    case CD_MP_BR:
			byte[] endbndry = resph.getEndBoundary(Stream);
			int[]  end_cmp  = resph.getEndCompiled(Stream);
			rcvd = Stream.read(b, off, len);
			if (rcvd == -1)
			    throw new EOFException("Premature EOF encountered");
			int ovf = Stream.pastEnd(endbndry, end_cmp);
			if (ovf != -1)
			{
			    rcvd -= ovf;
			    close(resph);
			}
			break;
		    default:
			throw new Error("Internal Error in StreamDemultiplexor: " +
					"Invalid cd_type " + resph.resp.cd_type);
		}
		restartTimer();
		return rcvd;
	    }
	    catch (InterruptedIOException ie)	
	    {
		restartTimer();
		throw ie;
	    }
	    catch (IOException ioe)
	    {
		Log.write(Log.DEMUX, "Demux: ", ioe);
		close(ioe, true);
		throw resph.exception;		
	    }
	    catch (ParseException pe)
	    {
		Log.write(Log.DEMUX, "Demux: ", pe);
		close(new IOException(pe.toString()), true);
		throw resph.exception;		
	    }
	}
    }
int read ( byte [ ] b , int off , int len , ResponseHandler resph , int timeout ) throws IOException { if ( resph . exception != null ) { resph . exception . fillInStackTrace ( ) ; throw resph . exception ; } if ( resph . eof ) return - 1 ; ResponseHandler head ; while ( ( head = ( ResponseHandler ) RespHandlerList . getFirst ( ) ) != null && head != resph ) { try { head . stream . readAll ( timeout ) ; } catch ( IOException ioe ) { if ( ioe instanceof InterruptedIOException ) throw ioe ; else { resph . exception . fillInStackTrace ( ) ; throw resph . exception ; } } } synchronized ( this ) { if ( resph . exception != null ) { resph . exception . fillInStackTrace ( ) ; throw resph . exception ; } if ( resph . resp . cd_type != CD_HDRS ) Log . write ( Log . DEMUX , "Demux: Reading for stream " + resph . stream . hashCode ( ) ) ; if ( Timer != null ) Timer . hyber ( ) ; try { int rcvd = - 1 ; if ( timeout != cur_timeout ) { Log . write ( Log . DEMUX , "Demux: Setting timeout to " + timeout + " ms" ) ; Sock . setSoTimeout ( timeout ) ; cur_timeout = timeout ; } switch ( resph . resp . cd_type ) { case CD_HDRS : rcvd = Stream . read ( b , off , len ) ; if ( rcvd == - 1 ) throw new EOFException ( "Premature EOF encountered" ) ; break ; case CD_0 : rcvd = - 1 ; close ( resph ) ; break ; case CD_CLOSE : rcvd = Stream . read ( b , off , len ) ; if ( rcvd == - 1 ) close ( resph ) ; break ; case CD_CONTLEN : int cl = resph . resp . ContentLength ; if ( len > cl - resph . stream . count ) len = cl - resph . stream . count ; rcvd = Stream . read ( b , off , len ) ; if ( rcvd == - 1 ) throw new EOFException ( "Premature EOF encountered" ) ; if ( resph . stream . count + rcvd == cl ) close ( resph ) ; break ; case CD_CHUNKED : if ( chunk_len == - 1 ) chunk_len = Codecs . getChunkLength ( Stream ) ; if ( chunk_len > 0 ) { if ( len > chunk_len ) len = ( int ) chunk_len ; rcvd = Stream . read ( b , off , len ) ; if ( rcvd == - 1 ) throw new EOFException ( "Premature EOF encountered" ) ; chunk_len -= rcvd ; if ( chunk_len == 0 ) { Stream . read ( ) ; Stream . read ( ) ; chunk_len = - 1 ; } } else { resph . resp . readTrailers ( Stream ) ; rcvd = - 1 ; close ( resph ) ; chunk_len = - 1 ; } break ; case CD_MP_BR : byte [ ] endbndry = resph . getEndBoundary ( Stream ) ; int [ ] end_cmp = resph . getEndCompiled ( Stream ) ; rcvd = Stream . read ( b , off , len ) ; if ( rcvd == - 1 ) throw new EOFException ( "Premature EOF encountered" ) ; int ovf = Stream . pastEnd ( endbndry , end_cmp ) ; if ( ovf != - 1 ) { rcvd -= ovf ; close ( resph ) ; } break ; default : throw new Error ( "Internal Error in StreamDemultiplexor: " + "Invalid cd_type " + resph . resp . cd_type ) ; } restartTimer ( ) ; return rcvd ; } catch ( InterruptedIOException ie ) { restartTimer ( ) ; throw ie ; } catch ( IOException ioe ) { Log . write ( Log . DEMUX , "Demux: " , ioe ) ; close ( ioe , true ) ; throw resph . exception ; } catch ( ParseException pe ) { Log . write ( Log . DEMUX , "Demux: " , pe ) ; close ( new IOException ( pe . toString ( ) ) , true ) ; throw resph . exception ; } } }
reads an array of bytes from the master stream.
reads an array of bytes from the master stream .
// read the headers and data for all responses preceding us. // Now we can read from the stream. // it's a new chunk // it's data // got the whole chunk // CR // LF // the footers (trailers) // don't intercept this one // set by retry_requests // set by retry_requests