StreamHandler


1   /*
2    * JBoss, the OpenSource J2EE webOS
3    *
4    * Distributable under LGPL license.
5    * See terms of license at gnu.org.
6    */
7   package org.jboss.remoting.stream;
8   
9   import java.io.IOException  ;
10  import java.io.InputStream  ;
11  import org.jboss.logging.Logger;
12  import org.jboss.remoting.Client;
13  import org.jboss.remoting.InvokerLocator;
14  
15  /**
16   * This is the server side proxy back to the orginal stream
17   * on the client side.  It implements InputStream, so can be
18   * passed and acted on by the server handler as a regular InputStream
19   * type.  For all the InputStream methods, it should behave EXACTLY
20   * like a local InputStream with the one exception being that it
21   * will sometimes throw IOExceptions based on network exceptions
22   * or in the case when the method does not throw an IOException, throwing
23   * a RuntimeException if network problem (however none of the method
24   * signatures are changed).
25   * <p/>
26   * Internally, it will use remoting to callback to the client.
27   *
28   * @author <a HREF="mailto:tom.elrod@jboss.com">Tom Elrod</a>
29   */
30  public class StreamHandler extends InputStream   //implements InvocationHandler
31  {
32     private InvokerLocator streamServerLocator = null;
33     private Client streamClient = null;
34  
35     private static final Logger log = Logger.getLogger(StreamHandler.class);
36  
37     // The remoting invocation methods the match the InputStream metnhods.
38     public static final String   READ = "read()";
39     public static final String   AVAILABLE = "available()";
40     public static final String   CLOSE = "close()";
41     public static final String   RESET = "reset()";
42     public static final String   MARKSUPPORTED = "markSupported()";
43     public static final String   MARKREADLIMIT = "mark(int readlimit)";
44     public static final String   SKIP = "skip(long n)";
45     public static final String   READBYTEARRAY = "read(byte b[])";
46     public static final String   READOFFSET = "read(byte b[], int off, int len)";
47  
48     /**
49      * Constructor requiring the locator url back to the client's
50      * StreamServer connector (actually the connector's server invoker).
51      *
52      * @param locatorURL
53      * @throws Exception
54      */
55     //private StreamHandler(String locatorURL) throws Exception
56     public StreamHandler(String   locatorURL) throws Exception  
57     {
58        streamServerLocator = new InvokerLocator(locatorURL);
59        streamClient = new Client(streamServerLocator);
60     }
61  
62     /**
63      * Returns the number of bytes that can be read (or skipped over) from
64      * this input stream without blocking by the next caller of a method for
65      * this input stream.  The next caller might be the same thread or or
66      * another thread.
67      * <p/>
68      * <p> The <code>available</code> method for class <code>InputStream</code>
69      * always returns <code>0</code>.
70      * <p/>
71      * <p> This method should be overridden by subclasses.
72      *
73      * @return the number of bytes that can be read from this input stream
74      *         without blocking.
75      * @throws java.io.IOException if an I/O error occurs.
76      */
77     public int available() throws IOException  
78     {
79        int readInt = 0;
80  
81        try
82        {
83           Integer   retInt = (Integer  ) streamClient.invoke(new StreamCallPayload(AVAILABLE));
84           if(retInt != null)
85           {
86              readInt = retInt.intValue();
87           }
88        }
89        catch(Throwable   throwable)
90        {
91           log.error("Error getting available from client stream.", throwable);
92           throw new IOException  (throwable.getMessage());
93        }
94        return readInt;
95     }
96  
97     /**
98      * Closes this input stream and releases any system resources associated
99      * with the stream.
100     * <p/>
101     * <p> The <code>close</code> method of <code>InputStream</code> does
102     * nothing.
103     *
104     * @throws java.io.IOException if an I/O error occurs.
105     */
106    public void close() throws IOException  
107    {
108       try
109       {
110          streamClient.invoke(new StreamCallPayload(CLOSE));
111       }
112       catch(Throwable   throwable)
113       {
114          log.error("Error closing client stream.", throwable);
115          throw new IOException  (throwable.getMessage());
116       }
117    }
118 
119    /**
120     * Repositions this stream to the position at the time the
121     * <code>mark</code> method was last called on this input stream.
122     * <p/>
123     * <p> The general contract of <code>reset</code> is:
124     * <p/>
125     * <p><ul>
126     * <p/>
127     * <li> If the method <code>markSupported</code> returns
128     * <code>true</code>, then:
129     * <p/>
130     * <ul><li> If the method <code>mark</code> has not been called since
131     * the stream was created, or the number of bytes read from the stream
132     * since <code>mark</code> was last called is larger than the argument
133     * to <code>mark</code> at that last call, then an
134     * <code>IOException</code> might be thrown.
135     * <p/>
136     * <li> If such an <code>IOException</code> is not thrown, then the
137     * stream is reset to a state such that all the bytes read since the
138     * most recent call to <code>mark</code> (or since the start of the
139     * file, if <code>mark</code> has not been called) will be resupplied
140     * to subsequent callers of the <code>read</code> method, followed by
141     * any bytes that otherwise would have been the next input data as of
142     * the time of the call to <code>reset</code>. </ul>
143     * <p/>
144     * <li> If the method <code>markSupported</code> returns
145     * <code>false</code>, then:
146     * <p/>
147     * <ul><li> The call to <code>reset</code> may throw an
148     * <code>IOException</code>.
149     * <p/>
150     * <li> If an <code>IOException</code> is not thrown, then the stream
151     * is reset to a fixed state that depends on the particular type of the
152     * input stream and how it was created. The bytes that will be supplied
153     * to subsequent callers of the <code>read</code> method depend on the
154     * particular type of the input stream. </ul></ul>
155     * <p/>
156     * <p> The method <code>reset</code> for class <code>InputStream</code>
157     * does nothing and always throws an <code>IOException</code>.
158     *
159     * @throws java.io.IOException if this stream has not been marked or if the
160     *                             mark has been invalidated.
161     * @see java.io.InputStream#mark(int)
162     * @see java.io.IOException
163     */
164    public synchronized void reset() throws IOException  
165    {
166       try
167       {
168          streamClient.invoke(new StreamCallPayload(RESET));
169       }
170       catch(Throwable   throwable)
171       {
172          log.error("Error reseting client stream.", throwable);
173          throw new IOException  (throwable.getMessage());
174       }
175    }
176 
177    /**
178     * Tests if this input stream supports the <code>mark</code> and
179     * <code>reset</code> methods. Whether or not <code>mark</code> and
180     * <code>reset</code> are supported is an invariant property of a
181     * particular input stream instance. The <code>markSupported</code> method
182     * of <code>InputStream</code> returns <code>false</code>.
183     *
184     * @return <code>true</code> if this stream instance supports the mark
185     *         and reset methods; <code>false</code> otherwise.
186     * @see java.io.InputStream#mark(int)
187     * @see java.io.InputStream#reset()
188     */
189    public boolean markSupported()
190    {
191       boolean supported = false;
192 
193       try
194       {
195          Boolean   bSupported = (Boolean  ) streamClient.invoke(new StreamCallPayload(MARKSUPPORTED));
196          if(bSupported != null)
197          {
198             supported = bSupported.booleanValue();
199          }
200       }
201       catch(Throwable   throwable)
202       {
203          log.error("Error getting markSupported from client stream.", throwable);
204          throw new RuntimeException  (throwable.getMessage(), throwable);
205       }
206       return supported;
207    }
208 
209    /**
210     * Marks the current position in this input stream. A subsequent call to
211     * the <code>reset</code> method repositions this stream at the last marked
212     * position so that subsequent reads re-read the same bytes.
213     * <p/>
214     * <p> The <code>readlimit</code> arguments tells this input stream to
215     * allow that many bytes to be read before the mark position gets
216     * invalidated.
217     * <p/>
218     * <p> The general contract of <code>mark</code> is that, if the method
219     * <code>markSupported</code> returns <code>true</code>, the stream somehow
220     * remembers all the bytes read after the call to <code>mark</code> and
221     * stands ready to supply those same bytes again if and whenever the method
222     * <code>reset</code> is called.  However, the stream is not required to
223     * remember any data at all if more than <code>readlimit</code> bytes are
224     * read from the stream before <code>reset</code> is called.
225     * <p/>
226     * <p> The <code>mark</code> method of <code>InputStream</code> does
227     * nothing.
228     *
229     * @param readlimit the maximum limit of bytes that can be read before
230     *                  the mark position becomes invalid.
231     * @see java.io.InputStream#reset()
232     */
233    public synchronized void mark(int readlimit)
234    {
235       try
236       {
237          StreamCallPayload payload = new StreamCallPayload(MARKREADLIMIT);
238          payload.setParams(new Object  []{new Integer  (readlimit)});
239          streamClient.invoke(payload);
240       }
241       catch(Throwable   throwable)
242       {
243          log.error("Error marking with read limit on client stream.", throwable);
244          throw new RuntimeException  (throwable.getMessage(), throwable);
245       }
246    }
247 
248    /**
249     * Skips over and discards <code>n</code> bytes of data from this input
250     * stream. The <code>skip</code> method may, for a variety of reasons, end
251     * up skipping over some smaller number of bytes, possibly <code>0</code>.
252     * This may result from any of a number of conditions; reaching end of file
253     * before <code>n</code> bytes have been skipped is only one possibility.
254     * The actual number of bytes skipped is returned.  If <code>n</code> is
255     * negative, no bytes are skipped.
256     * <p/>
257     * <p> The <code>skip</code> method of <code>InputStream</code> creates a
258     * byte array and then repeatedly reads into it until <code>n</code> bytes
259     * have been read or the end of the stream has been reached. Subclasses are
260     * encouraged to provide a more efficient implementation of this method.
261     *
262     * @param n the number of bytes to be skipped.
263     * @return the actual number of bytes skipped.
264     * @throws java.io.IOException if an I/O error occurs.
265     */
266    public long skip(long n) throws IOException  
267    {
268       long numSkipped = -1;
269 
270       try
271       {
272          StreamCallPayload payload = new StreamCallPayload(SKIP);
273          payload.setParams(new Object  []{new Long  (n)});
274          Long   ret = (Long  ) streamClient.invoke(payload);
275          if(ret != null)
276          {
277             numSkipped = ret.longValue();
278          }
279       }
280       catch(Throwable   throwable)
281       {
282          log.error("Error skipping on client stream.", throwable);
283          throw new IOException  (throwable.getMessage());
284       }
285 
286       return numSkipped;
287    }
288 
289    /**
290     * Reads some number of bytes from the input stream and stores them into
291     * the buffer array <code>b</code>. The number of bytes actually read is
292     * returned as an integer.  This method blocks until input data is
293     * available, end of file is detected, or an exception is thrown.
294     * <p/>
295     * <p> If <code>b</code> is <code>null</code>, a
296     * <code>NullPointerException</code> is thrown.  If the length of
297     * <code>b</code> is zero, then no bytes are read and <code>0</code> is
298     * returned; otherwise, there is an attempt to read at least one byte. If
299     * no byte is available because the stream is at end of file, the value
300     * <code>-1</code> is returned; otherwise, at least one byte is read and
301     * stored into <code>b</code>.
302     * <p/>
303     * <p> The first byte read is stored into element <code>b[0]</code>, the
304     * next one into <code>b[1]</code>, and so on. The number of bytes read is,
305     * at most, equal to the length of <code>b</code>. Let <i>k</i> be the
306     * number of bytes actually read; these bytes will be stored in elements
307     * <code>b[0]</code> through <code>b[</code><i>k</i><code>-1]</code>,
308     * leaving elements <code>b[</code><i>k</i><code>]</code> through
309     * <code>b[b.length-1]</code> unaffected.
310     * <p/>
311     * <p> If the first byte cannot be read for any reason other than end of
312     * file, then an <code>IOException</code> is thrown. In particular, an
313     * <code>IOException</code> is thrown if the input stream has been closed.
314     * <p/>
315     * <p> The <code>read(b)</code> method for class <code>InputStream</code>
316     * has the same effect as: <pre><code> read(b, 0, b.length) </code></pre>
317     *
318     * @param b the buffer into which the data is read.
319     * @return the total number of bytes read into the buffer, or
320     *         <code>-1</code> is there is no more data because the end of
321     *         the stream has been reached.
322     * @throws java.io.IOException  if an I/O error occurs.
323     * @throws NullPointerException if <code>b</code> is <code>null</code>.
324     * @see java.io.InputStream#read(byte[], int, int)
325     */
326    public int read(byte b[]) throws IOException  
327    {
328       if(b == null)
329       {
330          throw new NullPointerException  ("can not read for a null byte array.");
331       }
332       else
333       {
334          if(b.length == 0)
335          {
336             return 0;
337          }
338       }
339 
340       int retByte = -1;
341 
342       try
343       {
344          StreamCallPayload payload = new StreamCallPayload(READBYTEARRAY);
345          payload.setParams(new Object  []{b});
346          StreamCallPayload ret = (StreamCallPayload) streamClient.invoke(payload);
347          if(ret != null)
348          {
349             Object  [] retVals = ret.getParams();
350             byte[] retBytes = (byte[]) retVals[0];
351             Integer   retInt = (Integer  ) retVals[1];
352 
353             retByte = retInt.intValue();
354 
355             if(retByte != -1)
356             {
357                System.arraycopy(retBytes, 0, b, 0, retByte);
358             }
359          }
360       }
361       catch(Throwable   throwable)
362       {
363          log.error("Error reading from client stream.", throwable);
364          throw new IOException  (throwable.getMessage());
365       }
366 
367       return retByte;
368    }
369 
370    /**
371     * Reads up to <code>len</code> bytes of data from the input stream into
372     * an array of bytes.  An attempt is made to read as many as
373     * <code>len</code> bytes, but a smaller number may be read, possibly
374     * zero. The number of bytes actually read is returned as an integer.
375     * <p/>
376     * <p> This method blocks until input data is available, end of file is
377     * detected, or an exception is thrown.
378     * <p/>
379     * <p> If <code>b</code> is <code>null</code>, a
380     * <code>NullPointerException</code> is thrown.
381     * <p/>
382     * <p> If <code>off</code> is negative, or <code>len</code> is negative, or
383     * <code>off+len</code> is greater than the length of the array
384     * <code>b</code>, then an <code>IndexOutOfBoundsException</code> is
385     * thrown.
386     * <p/>
387     * <p> If <code>len</code> is zero, then no bytes are read and
388     * <code>0</code> is returned; otherwise, there is an attempt to read at
389     * least one byte. If no byte is available because the stream is at end of
390     * file, the value <code>-1</code> is returned; otherwise, at least one
391     * byte is read and stored into <code>b</code>.
392     * <p/>
393     * <p> The first byte read is stored into element <code>b[off]</code>, the
394     * next one into <code>b[off+1]</code>, and so on. The number of bytes read
395     * is, at most, equal to <code>len</code>. Let <i>k</i> be the number of
396     * bytes actually read; these bytes will be stored in elements
397     * <code>b[off]</code> through <code>b[off+</code><i>k</i><code>-1]</code>,
398     * leaving elements <code>b[off+</code><i>k</i><code>]</code> through
399     * <code>b[off+len-1]</code> unaffected.
400     * <p/>
401     * <p> In every case, elements <code>b[0]</code> through
402     * <code>b[off]</code> and elements <code>b[off+len]</code> through
403     * <code>b[b.length-1]</code> are unaffected.
404     * <p/>
405     * <p> If the first byte cannot be read for any reason other than end of
406     * file, then an <code>IOException</code> is thrown. In particular, an
407     * <code>IOException</code> is thrown if the input stream has been closed.
408     * <p/>
409     * <p> The <code>read(b,</code> <code>off,</code> <code>len)</code> method
410     * for class <code>InputStream</code> simply calls the method
411     * <code>read()</code> repeatedly. If the first such call results in an
412     * <code>IOException</code>, that exception is returned from the call to
413     * the <code>read(b,</code> <code>off,</code> <code>len)</code> method.  If
414     * any subsequent call to <code>read()</code> results in a
415     * <code>IOException</code>, the exception is caught and treated as if it
416     * were end of file; the bytes read up to that point are stored into
417     * <code>b</code> and the number of bytes read before the exception
418     * occurred is returned.  Subclasses are encouraged to provide a more
419     * efficient implementation of this method.
420     *
421     * @param b   the buffer into which the data is read.
422     * @param off the start offset in array <code>b</code>
423     *            at which the data is written.
424     * @param len the maximum number of bytes to read.
425     * @return the total number of bytes read into the buffer, or
426     *         <code>-1</code> if there is no more data because the end of
427     *         the stream has been reached.
428     * @throws java.io.IOException  if an I/O error occurs.
429     * @throws NullPointerException if <code>b</code> is <code>null</code>.
430     * @see java.io.InputStream#read()
431     */
432    public int read(byte b[], int off, int len) throws IOException  
433    {
434       if(b == null)
435       {
436          throw new NullPointerException  ("can not read for a null byte array.");
437       }
438       else
439       {
440          if(b.length == 0)
441          {
442             return 0;
443          }
444          else
445          {
446             if(off < 0 || len < 0 || off + len > b.length)
447             {
448                throw new IndexOutOfBoundsException  ("Either off or len is negative or off+len is greater than length of b.");
449             }
450             if(len == 0)
451             {
452                return 0;
453             }
454          }
455       }
456 
457       int retByte = -1;
458 
459       try
460       {
461          byte[] payloadArray = new byte[len];
462          StreamCallPayload payload = new StreamCallPayload(READBYTEARRAY);
463          payload.setParams(new Object  []{payloadArray});
464          StreamCallPayload ret = (StreamCallPayload) streamClient.invoke(payload);
465          if(ret != null)
466          {
467             Object  [] retVals = ret.getParams();
468             byte[] retBytes = (byte[]) retVals[0];
469             Integer   retInt = (Integer  ) retVals[1];
470 
471             retByte = retInt.intValue();
472 
473             System.arraycopy(retBytes, 0, b, off, retByte);
474          }
475       }
476       catch(Throwable   throwable)
477       {
478          log.error("Error reading with offset from client stream.", throwable);
479          throw new IOException  (throwable.getMessage());
480       }
481 
482       return retByte;
483    }
484 
485    /**
486     * Reads the next byte of data from the input stream. The value byte is
487     * returned as an <code>int</code> in the range <code>0</code> to
488     * <code>255</code>. If no byte is available because the end of the stream
489     * has been reached, the value <code>-1</code> is returned. This method
490     * blocks until input data is available, the end of the stream is detected,
491     * or an exception is thrown.
492     * <p/>
493     * <p> A subclass must provide an implementation of this method.
494     *
495     * @return the next byte of data, or <code>-1</code> if the end of the
496     *         stream is reached.
497     * @throws java.io.IOException if an I/O error occurs.
498     */
499    public int read() throws IOException  
500    {
501       int readInt = -1;
502 
503       try
504       {
505          Integer   retInt = (Integer  ) streamClient.invoke(new StreamCallPayload(READ));
506          if(retInt != null)
507          {
508             readInt = retInt.intValue();
509          }
510       }
511       catch(Throwable   throwable)
512       {
513          log.error("Error reading from client stream.", throwable);
514          throw new IOException  (throwable.getMessage());
515       }
516       return readInt;
517    }
518 
519 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags