RmiUnMarshaller


1   /***
2    * Fractal RMI: a binder for remote method calls between Fractal components.
3    * Copyright (C) 2003 France Telecom R&D
4    *
5    * This library is free software; you can redistribute it and/or
6    * modify it under the terms of the GNU Lesser General Public
7    * License as published by the Free Software Foundation; either
8    * version 2 of the License, or (at your option) any later version.
9    *
10   * This library is distributed in the hope that it will be useful,
11   * but WITHOUT ANY WARRANTY; without even the implied warranty of
12   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13   * Lesser General Public License for more details.
14   *
15   * You should have received a copy of the GNU Lesser General Public
16   * License along with this library; if not, write to the Free Software
17   * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
18   *
19   * Contact: Eric.Bruneton@rd.francetelecom.com
20   *
21   * Author: Eric Bruneton
22   *
23   * adapted from Jonathan:
24   *   org.objectweb.jeremie.libs.presentation.StdMarshallerFactory
25   *     (authors: B. Dumant, K. Milsted)
26   * with some comments copied from:
27   *   org.objectweb.jonathan.apis.presentation.UnMarshaller
28   *     (author: B. Dumant)
29   */
30  
31  package org.objectweb.fractal.rmi.io;
32  
33  import org.objectweb.jonathan.apis.binding.NamingContext;
34  import org.objectweb.jonathan.apis.kernel.Context;
35  import org.objectweb.jonathan.apis.kernel.ContextFactory;
36  import org.objectweb.jonathan.apis.kernel.InternalException;
37  import org.objectweb.jonathan.apis.kernel.JonathanException;
38  import org.objectweb.jonathan.apis.presentation.EndOfMessageException;
39  import org.objectweb.jonathan.apis.presentation.UnMarshaller;
40  import org.objectweb.jonathan.apis.resources.Chunk;
41  import org.objectweb.jonathan.apis.resources.ChunkProvider;
42  
43  import java.io.IOException  ;
44  import java.io.InputStream  ;
45  import java.io.ObjectInputStream  ;
46  
47  /**
48   * An {@link UnMarshaller} that uses an {@link RmiObjectInputStream} to
49   * unmarshall object values.
50   */
51  
52  public class RmiUnMarshaller extends InputStream   implements UnMarshaller {
53  
54    private final static Chunk EMPTY_CHUNK = new Chunk(new byte[0], 0, 0);
55  
56    /**
57     * The domain used to instantiate {@link #is is}.
58     */
59  
60    protected NamingContext domain;
61  
62    /**
63     * The context factory used to create the context associated with this
64     * marshaller. See {@link #getContext getContext}.
65     */
66  
67    protected ContextFactory contextFactory;
68  
69    /**
70     * The object input stream used to unmarshall object values. This stream is
71     * lazily instantiated, i.e., it is instantiated at the first call to {@link
72     * #readValue readValue} or to {@link #readReference readReference}.
73     */
74  
75    protected ObjectInputStream   is;
76  
77    private ChunkProvider provider;
78  
79    private Chunk first;
80  
81    private int offset;
82  
83    private int lastOffset;
84  
85    private Context context;
86  
87    // --------------------------------------------------------------------------
88    // Constructors
89    // --------------------------------------------------------------------------
90  
91    /**
92     * Constructs a new {@link RmiUnMarshaller}.
93     *
94     * @param domain the domain to be used to instantiate {@link #is is}.
95     * @param contextFactory the context factory to be used to create the context
96     *      associated with this marshaller.
97     * @param chunk a (chain of) chunk(s).
98     * @param read the number of bytes already read from the message.
99     */
100 
101   public RmiUnMarshaller (
102     NamingContext domain,
103     ContextFactory contextFactory,
104     Chunk chunk,
105     int read)
106   {
107     super();
108     this.domain = domain;
109     this.contextFactory = contextFactory;
110     this.first = chunk;
111     offset = chunk.offset;
112     lastOffset = read - offset;
113     context = null;
114   }
115 
116   /**
117    * Constructs a new {@link RmiUnMarshaller}.
118    *
119    * @param domain the domain to be used to instantiate {@link #is is}.
120    * @param contextFactory the context factory to be used to create the context
121    *      associated with this marshaller.
122    * @param provider a chunk provider.
123    */
124 
125   public RmiUnMarshaller (
126     final NamingContext domain,
127     final ContextFactory contextFactory,
128     final ChunkProvider provider)
129   {
130     this(domain, contextFactory, EMPTY_CHUNK, 0);
131     this.provider = provider;
132   }
133 
134   // --------------------------------------------------------------------------
135   // Implementation of the UnMarshaller interface
136   // --------------------------------------------------------------------------
137 
138   /**
139    * Returns true if this unmarshaller is little-endian, false otherwise.
140    *
141    * @return true if this unmarshaller is little-endian, false otherwise.
142    */
143 
144   public boolean isLittleEndian () {
145     return false;
146   }
147 
148   /**
149    * Sets the byte order  of this unmarshaller.
150    *
151    * @param littleEndian the new byte order.
152    */
153 
154   public void setByteOrder (final boolean littleEndian) {
155     throw new RuntimeException  ("Not implemented.");
156   }
157 
158   /**
159    * Returns a {@link Context} associated with this unmarshaller.
160    *
161    * @return a {@link Context} associated with this unmarshaller.
162    */
163 
164   public final Context getContext () {
165     if (context == null) {
166       try {
167         context = contextFactory.newContext();
168       } catch (NullPointerException   e) {
169         e.printStackTrace(System.err);
170         throw new InternalException("Context factory required.");
171       }
172     }
173     return context;
174   }
175 
176   /**
177    * Returns the number of bytes read since the beginning.
178    *
179    * @return the number of bytes read since the beginning.
180    */
181 
182   public int bytesRead () {
183     return lastOffset + offset;
184   }
185 
186   /**
187    * Sets the number of bytes readable from the unmarshaller. Once this method
188    * has been called, it won't be possible to read more than the
189    * <code>size</code> specified bytes from this unmarshaller. Knowing the exact
190    * number of readable bytes lets the unmarshaller free the resources (such as
191    * a chunk provider) that won't be used. This method may block until the
192    * expected number of bytes is readable.
193    *
194    * @param size the expected number of readable bytes.
195    * @throws JonathanException if something goes wrong.
196    */
197 
198   public void setSize (int size) throws JonathanException {
199     boolean sz0 = (size == 0);
200     Chunk result = null, current = null;
201     int lastOff = lastOffset + offset;
202     int available = first.top - offset;
203     first.offset = offset;
204     while (size > available) {
205       if (available > 0) {
206         if (current == null) {
207           result = first.duplicate();
208           current = result;
209         } else {
210           current.next = first.duplicate();
211           current = current.next;
212         }
213         size -= available;
214         offset = first.top;
215       }
216       prepare();
217       available = first.top - offset;
218     }
219     if (size > 0) {
220       if (current == null) {
221         result = first.duplicate(offset,offset + size);
222       } else {
223         current.next = first.duplicate(offset,offset + size);
224       }
225       offset += size;
226     }
227     Context thisContext = context;
228     if (context != null) {
229       context.acquire();
230     }
231     close();
232     context = thisContext;
233     if (sz0) {
234       first = EMPTY_CHUNK;
235     } else {
236       first = result;
237     }
238     offset = first.offset;
239     lastOffset = lastOff - offset;
240   }
241 
242   /**
243    * Returns an input stream to read data from the unmarshaller. Closing the
244    * returned input stream has the same effect as closing the actual
245    * unmarshaller.
246    *
247    * @return an input stream to read from the unmarshaller.
248    */
249 
250   public InputStream   inputStream () {
251     return this;
252   }
253 
254   /**
255    * Reads a byte.
256    *
257    * @return a byte.
258    * @throws JonathanException if the format of the data is incompatible with
259    *      the request.
260    */
261 
262   public byte readByte () throws JonathanException {
263     if (first.top == offset) {
264       prepare();
265     }
266     return first.data[offset++];
267   }
268 
269   /**
270    * Reads a boolean.
271    *
272    * @return a boolean.
273    * @throws JonathanException if the format of the data is incompatible with
274    *      the request.
275    */
276 
277   public boolean readBoolean () throws JonathanException {
278     if (first.top == offset) {
279       prepare();
280     }
281     return first.data[offset++] != 0;
282   }
283 
284   /**
285    * Reads a 8 bits char.
286    *
287    * @return a char.
288    * @throws JonathanException if the format of the data is incompatible with
289    *      the request.
290    */
291 
292   public char readChar8 () throws JonathanException {
293     if (first.top == offset) {
294       prepare();
295     }
296     return (char) (first.data[offset++] & 0xFF);
297   }
298 
299   /**
300    * Reads a 16 bits char.
301    *
302    * @return a char.
303    * @throws JonathanException if the format of the data is incompatible with
304    *      the request.
305    */
306 
307   public char readChar16 () throws JonathanException {
308     byte[] data = first.data;
309     int val = 0;
310     if (first.top - offset >= 2) {
311       val = ((data[offset] & 0xFF) << 8) + (data[offset + 1] & 0xFF);
312       offset += 2;
313     } else {
314       int m = 1, p;
315       while (m >= 0) {
316         p = Math.min(m + 1, first.top-offset);
317         if (p == 0) {
318           prepare();
319           data = first.data;
320         } else {
321           while (--p >= 0) {
322             val += (data[offset++] & 0xFF) << (m-- * 8);
323           }
324         }
325       }
326     }
327     return (char) val;
328   }
329 
330   /**
331    * Reads a short.
332    *
333    * @return a short.
334    * @throws JonathanException if the format of the data is incompatible with
335    *      the request.
336    */
337 
338   public short readShort () throws JonathanException {
339     byte[] data = first.data;
340     int val = 0;
341     if (first.top - offset >= 2) {
342       val = ((data[offset] & 0xFF) << 8) + (data[offset + 1] & 0xFF);
343       offset += 2;
344     } else {
345       int m = 1, p;
346       while (m >= 0) {
347         p = Math.min(m + 1, first.top-offset);
348         if (p == 0) {
349           prepare();
350           data = first.data;
351         } else {
352           while (--p >= 0) {
353             val += (data[offset++] & 0xFF) << (m-- * 8);
354           }
355         }
356       }
357     }
358     return (short) val;
359   }
360   /**
361    * Reads an int.
362    *
363    * @return an int.
364    * @throws JonathanException if the format of the data is incompatible with
365    *      the request.
366    */
367 
368   public int readInt () throws JonathanException {
369     byte[] data = first.data;
370     int val = 0;
371     if (first.top - offset >= 4) {
372       val =
373         ((data[offset] & 0xFF) << 24) +
374         ((data[offset + 1] & 0xFF) << 16) +
375         ((data[offset + 2] & 0xFF) << 8) +
376         (data[offset + 3] & 0xFF);
377       offset += 4;
378     } else {
379       int m = 3, p;
380       while (m >= 0) {
381         p = Math.min(m + 1, first.top-offset);
382         if (p == 0) {
383           prepare();
384           data = first.data;
385         } else {
386           while (--p >= 0) {
387             val += (data[offset++] & 0xFF) << (m-- * 8);
388           }
389         }
390       }
391     }
392     return val;
393   }
394 
395   /**
396    * Reads a float.
397    *
398    * @return a float.
399    * @throws JonathanException if the format of the data is incompatible with
400    *      the request.
401    */
402 
403   public float readFloat () throws JonathanException {
404     byte[] data = first.data;
405     int val = 0;
406     if (first.top - offset >= 4) {
407       val =
408         ((data[offset] & 0xFF) << 24) +
409         ((data[offset + 1] & 0xFF) << 16) +
410         ((data[offset + 2] & 0xFF) << 8) +
411         (data[offset + 3] & 0xFF);
412       offset += 4;
413     } else {
414       int m = 3, p;
415       while (m >= 0) {
416         p = Math.min(m + 1, first.top-offset);
417         if (p == 0) {
418           prepare();
419           data = first.data;
420         } else {
421           while (--p >= 0) {
422             val += (data[offset++] & 0xFF) << (m-- * 8);
423           }
424         }
425       }
426     }
427     return Float.intBitsToFloat(val);
428   }
429 
430   /**
431    * Reads a long.
432    *
433    * @return a long.
434    * @throws JonathanException if the format of the data is incompatible with
435    *      the request.
436    */
437 
438   public long readLong () throws JonathanException {
439     byte[] data = first.data;
440     long val = 0;
441     if (first.top - offset >= 8) {
442       val =
443         (((long)data[offset] & 0xFF) << 56) +
444         (((long)data[offset + 1] & 0xFF) << 48) +
445         (((long)data[offset + 2] & 0xFF) << 40) +
446         (((long)data[offset + 3] & 0xFF) << 32) +
447         (((long)data[offset + 4] & 0xFF) << 24) +
448         (((long)data[offset + 5] & 0xFF) << 16) +
449         (((long)data[offset + 6] & 0xFF) << 8) +
450         ((long)data[offset + 7] & 0xFF);
451       offset += 8;
452     } else {
453       int m = 7, p;
454       while (m >= 0) {
455         p = Math.min(m + 1,first.top-offset);
456         if (p == 0) {
457           prepare();
458           data = first.data;
459         } else {
460           while (--p >= 0) {
461             val += (((long) data[offset++]) & 0xFF) << (m-- * 8);
462           }
463 
464         }
465       }
466     }
467     return val;
468   }
469 
470   /**
471    * Reads a double.
472    *
473    * @return a double.
474    * @throws JonathanException if the format of the data is incompatible with
475    *      the request.
476    */
477 
478   public double readDouble () throws JonathanException {
479     byte[] data = first.data;
480     long val = 0;
481     if (first.top - offset >= 8) {
482       val =
483         (((long)data[offset] & 0xFF) << 56) +
484         (((long)data[offset + 1] & 0xFF) << 48) +
485         (((long)data[offset + 2] & 0xFF) << 40) +
486         (((long)data[offset + 3] & 0xFF) << 32) +
487         (((long)data[offset + 4] & 0xFF) << 24) +
488         (((long)data[offset + 5] & 0xFF) << 16) +
489         (((long)data[offset + 6] & 0xFF) << 8) +
490         ((long)data[offset + 7] & 0xFF);
491       offset += 8;
492       return Double.longBitsToDouble(val);
493     } else {
494       int m = 7, p;
495       while (m >= 0) {
496         p = Math.min(m + 1,first.top-offset);
497         if (p == 0) {
498           prepare();
499           data = first.data;
500         } else {
501           while (--p >= 0) {
502             val += (((long) data[offset++]) & 0xFF) << (m-- * 8);
503           }
504         }
505       }
506       return Double.longBitsToDouble(val);
507     }
508   }
509 
510   /**
511    * Reads a string composed of 8 bits chars.
512    *
513    * @return a string.
514    * @throws JonathanException if the format of the data is incompatible with
515    *      the request.
516    */
517 
518   public String   readString8 () throws JonathanException {
519     int len = readInt() - 1;
520     char[] tab = new char[len];
521     byte[] data = first.data;
522     int off = 0;
523     int max = first.top - offset;
524     // read the characters
525     while (max < len) {
526       while (off < max) {
527         tab[off++] = (char) (data[offset++] & 0xFF);
528       }
529       prepare();
530       data = first.data;
531       max = first.top - offset + off;
532     }
533     while (off < len) {
534       tab[off++] = (char) (data[offset++] & 0xFF);
535     }
536     if (first.top == offset) {
537       prepare();
538       offset++;
539     } else {
540       offset++;
541     }
542     return new String  (tab);
543   }
544 
545   /**
546    * Reads a string composed of 16 bits chars.
547    *
548    * @return a string.
549    * @throws JonathanException if the format of the data is incompatible with
550    *      the request.
551    */
552 
553   public String   readString16 () throws JonathanException {
554     throw new InternalException("Not implemented.");
555   }
556 
557   /**
558    * Reads an array of bytes.
559    *
560    * @param array a byte array (of size >= offset + len).
561    * @param off the position (in array) of the first byte to write.
562    * @param len the total number of bytes to read.
563    * @throws JonathanException if the format of the data is incompatible with
564    *      the request.
565    */
566 
567   public void readByteArray (final byte[] array, int off, int len)
568     throws JonathanException
569   {
570     byte[] data = first.data;
571     int max = first.top - offset;
572     while (max < len) {
573       System.arraycopy(data,offset,array,off,max);
574       offset += max;
575       off += max;
576       len -= max;
577       prepare();
578       data = first.data;
579       max = first.top - offset;
580     }
581     System.arraycopy(data,offset,array,off,len);
582     offset += len;
583   }
584 
585   /**
586    * Reads a reference to an object.
587    *
588    * @return a reference to an object.
589    * @throws JonathanException if the format of the data is incompatible with
590    *      the request.
591    */
592 
593   public Object   readReference () throws JonathanException {
594     try {
595       initObjectStream();
596       return is.readObject();
597     } catch (Exception   e) {
598       System.err.println(Thread.currentThread() + " Exception " + e);
599       throw new JonathanException(e);
600     }
601   }
602 
603   /**
604    * Reads a value.
605    *
606    * @return an object representing the read value.
607    * @throws JonathanException if the format of the data is incompatible with
608    *      the request.
609    */
610 
611   public Object   readValue () throws JonathanException {
612     try {
613       initObjectStream();
614       return is.readObject();
615     } catch (Exception   e) {
616       System.err.println(Thread.currentThread() + " Exception " + e);
617       throw new JonathanException(e);
618     }
619   }
620 
621   // --------------------------------------------------------------------------
622   // Overriden methods
623   // --------------------------------------------------------------------------
624 
625   public int available () throws IOException   {
626     Chunk chunk = first;
627     int size = chunk.top - offset;
628     chunk = chunk.next;
629     while (chunk != null) {
630       size += chunk.top - chunk.offset;
631       chunk = chunk.next;
632     }
633     return size;
634   }
635 
636   public int read () throws IOException   {
637     try {
638       if (first.top == offset) {
639         prepare();
640       }
641       return first.data[offset++] & 0xFF;
642     } catch (JonathanException e) {
643       Throwable   f = e.represents();
644       if (f instanceof EndOfMessageException) {
645         return -1;
646       } else if (f instanceof IOException  ) {
647         throw (IOException  ) f;
648       } else {
649         throw new IOException  (f.getMessage());
650       }
651     }
652   }
653 
654   public int read (final byte[] array, int off, int length) throws IOException   {
655     try {
656       if (array == null) {
657         return (int) skip(length);
658       } else {
659         int len = length, toCopy, available;
660         while(len > 0) {
661           available = first.top - offset;
662           if (available == 0) {
663             prepare();
664             available = first.top - offset;
665           }
666           toCopy = Math.min(available,len);
667           System.arraycopy(first.data,offset,array,off,toCopy);
668           offset += toCopy;
669           off += toCopy;
670           len -= toCopy;
671         }
672         return length;
673       }
674     } catch (JonathanException e) {
675       Throwable   f = e.represents();
676       if (f instanceof EndOfMessageException) {
677         return -1;
678       } else if (f instanceof IOException  ) {
679         throw (IOException  ) f;
680       } else {
681         throw new IOException  (f.getMessage());
682       }
683     }
684   }
685 
686   public long skip (final long n) throws IOException   {
687     try {
688       long len = n, toSkip;
689       int available;
690       while(len > 0) {
691         available = first.top - offset;
692         if (available == 0) {
693           prepare();
694           available = first.top - offset;
695         }
696         toSkip = Math.min(available,len);
697         offset += toSkip;
698         len -= toSkip;
699       }
700       return n;
701     } catch (JonathanException e) {
702       Throwable   f = e.represents();
703       if (f instanceof EndOfMessageException) {
704         return -1;
705       } else if (f instanceof IOException  ) {
706         throw (IOException  ) f;
707       } else {
708         throw new IOException  (f.getMessage());
709       }
710     }
711   }
712 
713   /**
714    * Closes this input stream and releases any system resources associated
715    * with the stream.
716    */
717 
718   public void close () {
719     if (first != null) {
720       first.offset = offset;
721     }
722     Chunk next;
723     while (first != null) {
724       next = first.next;
725       first.release();
726       first = next;
727     }
728     if (provider != null) {
729       provider.close();
730       provider = null;
731     }
732     if (context != null) {
733       context.release();
734       context = null;
735     }
736   }
737 
738   // --------------------------------------------------------------------------
739   // Utility methods
740   // --------------------------------------------------------------------------
741 
742   /**
743    * Instantiates the delegate object input stream {@link #is is}. This
744    * method creates {@link RmiObjectInputStream} stream.
745    *
746    * @throws IOException if the delegate object input stream cannot be created.
747    */
748 
749   protected void initObjectStream () throws IOException   {
750     if (is == null) {
751       is = new RmiObjectInputStream(this, domain, contextFactory);
752     }
753   }
754 
755   private void prepare () throws JonathanException {
756     lastOffset += offset;
757     first.offset = offset;
758     Chunk prev;
759     while (first != null && first.top == first.offset) {
760       prev = first;
761       first = first.next;
762       prev.release();
763     }
764     if (first != null) {
765       offset = first.offset;
766       lastOffset -= offset;
767     } else if (provider != null) {
768       first = provider.prepare();
769       offset = first.offset;
770       lastOffset -= offset;
771     } else {
772       throw new EndOfMessageException();
773     }
774   }
775 }
776
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags