JkInputStream


1   /*
2    *  Copyright 1999-2004 The Apache Software Foundation
3    *
4    *  Licensed under the Apache License, Version 2.0 (the "License");
5    *  you may not use this file except in compliance with the License.
6    *  You may obtain a copy of the License at
7    *
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    *
10   *  Unless required by applicable law or agreed to in writing, software
11   *  distributed under the License is distributed on an "AS IS" BASIS,
12   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   *  See the License for the specific language governing permissions and
14   *  limitations under the License.
15   */
16  
17  package org.apache.jk.common;
18  
19  import java.io.IOException  ;
20  import java.io.InputStream  ;
21  import org.apache.jk.core.JkHandler;
22  import org.apache.jk.core.Msg;
23  import org.apache.jk.core.MsgContext;
24  import org.apache.tomcat.util.buf.ByteChunk;
25  
26  
27  /** Generic input stream impl on top of ajp
28   */
29  public class JkInputStream extends InputStream   {
30      private static org.apache.commons.logging.Log log=
31          org.apache.commons.logging.LogFactory.getLog( JkInputStream.class );
32  
33      public JkInputStream() {
34      }
35  
36      public int available() throws IOException   {
37          if( log.isDebugEnabled() )
38              log.debug( "available(): "  + blen + " " + pos );
39          return blen-pos;
40      }
41  
42      public void close() throws IOException   {
43          if( log.isDebugEnabled() )
44              log.debug( "cloae() " );
45          this.closed=true;
46      }
47  
48      public void mark(int readLimit) {
49      }
50  
51      public boolean markSupported() {
52          return false;
53      }
54  
55      public void reset() throws IOException   {
56          throw new IOException  ("reset() not supported");
57      }
58  
59      public int read() throws IOException   {
60          if( contentLength == -1 ) {
61              return doRead1();
62      }
63      if( available <= 0 ) {
64              if( log.isDebugEnabled() )
65                  log.debug("doRead() nothing available" );
66              return -1;
67          }
68      available--;
69  
70          return doRead1();
71      }
72      
73      public int read(byte[] b) throws IOException   {
74          int rd=read( b, 0, b.length);
75          if( log.isDebugEnabled() )
76              log.debug("read(" + b + ")=" + rd + " / " + b.length);
77          return rd;
78      }
79      
80      public int read(byte[] b, int off, int len) throws IOException   {
81          int rd=-1;
82      if( contentLength == -1 ) {
83          rd=doRead1(b,off,len);
84          return rd;
85      }
86      if( available <= 0 ) {
87              if( log.isDebugEnabled() ) log.debug("doRead() nothing available" );
88          return -1;
89          }
90          
91      rd=doRead1( b,off, len );
92      available -= rd;
93      if( log.isDebugEnabled() )
94              log.debug("Read: " + new String  ( b,off, len ));
95      return rd;
96      }
97  
98      public long skip(long n) throws IOException   {
99          if (n > Integer.MAX_VALUE) {
100             throw new IOException  ("can't skip than many:  " + n);
101         }
102         // XXX if n is big, split this in multiple reads
103         byte[] b = new byte[(int)n];
104         return read(b, 0, b.length);
105     }
106 
107 
108     // -------------------- Jk specific methods --------------------
109 
110     Msg bodyMsg=new MsgAjp();
111     MsgContext mc;
112 
113     // Total length of the body - maximum we can read
114     // If -1, we don't use any limit, and we don't count available
115     int contentLength;
116     // How much remains unread.
117     int available;
118 
119     boolean closed=false;
120 
121     // Ajp13 specific -  needs refactoring for the new model
122     public static final int MAX_PACKET_SIZE=8192;
123     public static final int H_SIZE=4;  // Size of basic packet header
124     public static final int  MAX_READ_SIZE = MAX_PACKET_SIZE - H_SIZE - 2;
125     public static final byte JK_AJP13_GET_BODY_CHUNK    = 6;
126 
127     
128     // Holds incoming chunks of request body data
129     // XXX We do a copy that could be avoided !
130     byte []bodyBuff = new byte[9000];
131     int blen;  // Length of current chunk of body data in buffer
132     int pos;   // Current read position within that buffer
133 
134     boolean end_of_stream=false; // true if we've received an empty packet
135     
136     private int doRead1() throws IOException   {
137         if(pos >= blen) {
138             if( ! refillReadBuffer()) {
139         return -1;
140         }
141         }
142         int i=bodyBuff[pos++] & 0xFF;
143         if( log.isDebugEnabled() ) log.debug("doRead1 " + (char)i );
144         return i;  // prevent sign extension of byte value
145     }
146 
147     public int doRead1(byte[] b, int off, int len) throws IOException   
148     {
149     if(pos >= blen) {
150         if( ! refillReadBuffer()) {
151         return -1;
152         }
153     }
154 
155     if(pos + len <= blen) { // Fear the off by one error
156         // Sanity check b.length > off + len?
157         System.arraycopy(bodyBuff, pos, b, off, len);
158         if( log.isDebugEnabled() )
159         log.debug("doRead1: " + pos + " " + len + " " + blen);
160             if( log.isTraceEnabled() )
161                 log.trace("Data: \n" + new String  ( b, off, len ));
162         pos += len;
163         return len;
164     }
165 
166     // Not enough data (blen < pos + len) or chunked encoded
167     int toCopy = len;
168     while(toCopy > 0) {
169         int bytesRemaining = blen - pos;
170         if(bytesRemaining < 0) 
171         bytesRemaining = 0;
172         int c = bytesRemaining < toCopy ? bytesRemaining : toCopy;
173 
174         System.arraycopy(bodyBuff, pos, b, off, c);
175         if( log.isDebugEnabled() )
176         log.debug("doRead2: " + pos + " " + len + " " +
177                           blen + " " + c);
178             if( log.isTraceEnabled() )
179                 log.trace("Data: \n" + new String  ( b, off, (len<blen-1)?len:blen-1 ));
180 
181         toCopy    -= c;
182 
183         off       += c;
184         pos       += c; // In case we exactly consume the buffer
185 
186         if(toCopy > 0) 
187         if( ! refillReadBuffer()) { // Resets blen and pos
188             break;
189         }
190     }
191 
192     return len - toCopy;
193     }
194 
195     /** Must be called after the request is parsed, before
196      *  any input
197      */
198     public void setContentLength( int i ) {
199         contentLength=i;
200         available=i;
201     }
202 
203     /** Must be called when the stream is created
204      */
205     public void setMsgContext( MsgContext mc ) {
206         this.mc=mc;
207     }
208 
209     /** Must be called before or after each request
210      */
211     public void recycle() {
212         available=0;
213         blen = 0;
214         pos = 0;
215         closed=false;
216         end_of_stream = false;
217         contentLength=-1;
218     }
219 
220     /**
221      */
222     public int doRead(ByteChunk responseChunk ) throws IOException   {
223         if( log.isDebugEnabled())
224             log.debug( "doRead " + pos + " " + blen + " " + available + " " + end_of_stream+
225                        " " + responseChunk.getOffset()+ " " + responseChunk.getLength());
226         if( end_of_stream ) {
227             return -1;
228         }
229         if( blen == pos ) {
230             if ( !refillReadBuffer() ){
231                 return -1;
232             }
233         }
234         responseChunk.setBytes( bodyBuff, pos, blen );
235         pos=blen;
236         return blen;
237     }
238     
239     /** Receive a chunk of data. Called to implement the
240      *  'special' packet in ajp13 and to receive the data
241      *  after we send a GET_BODY packet
242      */
243     public boolean receive() throws IOException  
244     {
245         mc.setType( JkHandler.HANDLE_RECEIVE_PACKET );
246         bodyMsg.reset();
247         int err = mc.getSource().receive(bodyMsg, mc);
248         if( log.isDebugEnabled() )
249             log.info( "Receiving: getting request body chunk " + err + " " + bodyMsg.getLen() );
250         
251         if(err < 0) {
252         throw new IOException  ();
253     }
254 
255         pos=0;
256         blen=0;
257 
258         // No data received.
259     if( bodyMsg.getLen() == 0 ) { // just the header
260             // Don't mark 'end of stream' for the first chunk.
261             // end_of_stream = true;
262         return false;
263     }
264         blen = bodyMsg.peekInt();
265 
266         if( blen == 0 ) {
267             return false;
268         }
269 
270         if( blen > bodyBuff.length ) {
271             bodyMsg.dump("Body");
272         }
273         
274         if( log.isTraceEnabled() ) {
275             bodyMsg.dump("Body buffer");
276         }
277         
278         int cpl=bodyMsg.getBytes(bodyBuff);
279 
280         if( log.isDebugEnabled() )
281             log.debug( "Copy into body buffer2 " + bodyBuff + " " + cpl + " " + blen );
282 
283         if( log.isTraceEnabled() )
284             log.trace( "Data:\n" + new String  ( bodyBuff, 0, cpl ));
285 
286     return (blen > 0);
287     }
288     
289     /**
290      * Get more request body data from the web server and store it in the 
291      * internal buffer.
292      *
293      * @return true if there is more data, false if not.    
294      */
295     private boolean refillReadBuffer() throws IOException   
296     {
297     // If the server returns an empty packet, assume that that end of
298     // the stream has been reached (yuck -- fix protocol??).
299         if (end_of_stream) {
300             if( log.isDebugEnabled() ) log.debug("refillReadBuffer: end of stream " );
301             return false;
302         }
303 
304     // Why not use outBuf??
305     bodyMsg.reset();
306     bodyMsg.appendByte(JK_AJP13_GET_BODY_CHUNK);
307     bodyMsg.appendInt(MAX_READ_SIZE);
308         
309     if( log.isDebugEnabled() )
310             log.debug("refillReadBuffer " + Thread.currentThread());
311 
312         mc.setType( JkHandler.HANDLE_SEND_PACKET );
313     mc.getSource().send(bodyMsg, mc);
314 
315         // In JNI mode, response will be in bodyMsg. In TCP mode, response need to be
316         // read
317 
318         //bodyMsg.dump("refillReadBuffer ");
319         
320         boolean moreData=receive();
321         if( !moreData ) {
322             end_of_stream=true;
323         }
324         return moreData;
325     }
326 
327 }
328
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags