MuxServer


1   /*
2    * The Apache Software License, Version 1.1
3    *
4    * Copyright (c) 2001-2004 Caucho Technology, Inc.  All rights reserved.
5    *
6    * Redistribution and use in source and binary forms, with or without
7    * modification, are permitted provided that the following conditions
8    * are met:
9    *
10   * 1. Redistributions of source code must retain the above copyright
11   *    notice, this list of conditions and the following disclaimer.
12   *
13   * 2. Redistributions in binary form must reproduce the above copyright
14   *    notice, this list of conditions and the following disclaimer in
15   *    the documentation and/or other materials provided with the
16   *    distribution.
17   *
18   * 3. The end-user documentation included with the redistribution, if
19   *    any, must include the following acknowlegement:
20   *       "This product includes software developed by the
21   *        Caucho Technology (http://www.caucho.com/)."
22   *    Alternately, this acknowlegement may appear in the software itself,
23   *    if and wherever such third-party acknowlegements normally appear.
24   *
25   * 4. The names "Hessian", "Resin", and "Caucho" must not be used to
26   *    endorse or promote products derived from this software without prior
27   *    written permission. For written permission, please contact
28   *    info@caucho.com.
29   *
30   * 5. Products derived from this software may not be called "Resin"
31   *    nor may "Resin" appear in their names without prior written
32   *    permission of Caucho Technology.
33   *
34   * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
35   * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
36   * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
37   * DISCLAIMED.  IN NO EVENT SHALL CAUCHO TECHNOLOGY OR ITS CONTRIBUTORS
38   * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
39   * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
40   * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
41   * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
42   * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
43   * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
44   * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
45   *
46   * @author Scott Ferguson
47   */
48  
49  package com.caucho.hessian.mux;
50  
51  import java.io.IOException  ;
52  import java.io.InputStream  ;
53  import java.io.OutputStream  ;
54  
55  /**
56   * Hessian Mux, a peer-to-peer protocol.
57   */
58  public class MuxServer {
59    private Object   READ_LOCK = new Object  ();
60    private Object   WRITE_LOCK = new Object  ();
61    
62    private InputStream   is;
63    private OutputStream   os;
64    private boolean isClient;
65  
66    private transient boolean isClosed;
67    
68    // channels that have data ready.
69    private boolean inputReady[] = new boolean[4];
70  
71    // true if there's a thread already reading
72    private boolean isReadLocked;
73    // true if there's a thread already writing
74    private boolean isWriteLocked;
75  
76    /**
77     * Null argument constructor.
78     */
79    public MuxServer()
80    {
81    }
82  
83    /**
84     * Create a new multiplexor with input and output streams.
85     *
86     * @param is the underlying input stream
87     * @param os the underlying output stream
88     * @param isClient true if this is the connection client.
89     */
90    public MuxServer(InputStream   is, OutputStream   os, boolean isClient)
91    {
92      init(is, os, isClient);
93    }
94  
95    /**
96     * Initialize the multiplexor with input and output streams.
97     *
98     * @param is the underlying input stream
99     * @param os the underlying output stream
100    * @param isClient true if this is the connection client.
101    */
102   public void init(InputStream   is, OutputStream   os, boolean isClient)
103   {
104     this.is = is;
105     this.os = os;
106     this.isClient = isClient;
107   }
108 
109   /**
110    * Gets the raw input stream.  Clients will normally not call
111    * this.
112    */
113   public InputStream   getInputStream()
114   {
115     return is;
116   }
117 
118   /**
119    * Gets the raw output stream.  Clients will normally not call
120    * this.
121    */
122   public OutputStream   getOutputStream()
123   {
124     return os;
125   }
126 
127   /**
128    * Starts a client call.
129    */
130   public boolean startCall(MuxInputStream in, MuxOutputStream out)
131     throws IOException  
132   {
133     int channel = isClient ? 2 : 3;
134 
135     return startCall(channel, in, out);
136   }
137 
138   /**
139    * Starts a client call.
140    */
141   public boolean startCall(int channel, MuxInputStream in, MuxOutputStream out)
142     throws IOException  
143   {
144     // XXX: Eventually need to check to see if the channel is used.
145     // It's not clear whether this should cause a wait or an exception.
146     
147     in.init(this, channel);
148     out.init(this, channel);
149 
150     return true;
151   }
152   
153   /**
154    * Reads a server request.
155    */
156   public boolean readRequest(MuxInputStream in, MuxOutputStream out)
157     throws IOException  
158   {
159     int channel = isClient ? 3 : 2;
160     
161     in.init(this, channel);
162     out.init(this, channel);
163 
164     if (readChannel(channel) != null) {
165       in.setInputStream(is);
166       in.readToData(false);
167       return true;
168     }
169     else
170       return false;
171   }
172 
173   /**
174    * Grabs the channel for writing.
175    *
176    * @param channel the channel
177    *
178    * @return true if the channel has permission to write.
179    */
180   OutputStream   writeChannel(int channel)
181     throws IOException  
182   {
183     while (os != null) {
184       boolean canWrite = false;
185       synchronized (WRITE_LOCK) {
186         if (! isWriteLocked) {
187           isWriteLocked = true;
188           canWrite = true;
189         }
190         else {
191           try {
192             WRITE_LOCK.wait(5000);
193           } catch (Exception   e) {
194           }
195         }
196       }
197 
198       if (canWrite) {
199         os.write('C');
200         os.write(channel >> 8);
201         os.write(channel);
202         
203         return os;
204       }
205     }
206 
207     return null;
208   }
209 
210   void yield(int channel)
211     throws IOException  
212   {
213     os.write('Y');
214     freeWriteLock();
215   }
216 
217   void flush(int channel)
218     throws IOException  
219   {
220     os.write('Y');
221     os.flush();
222     freeWriteLock();
223   }
224 
225   void close(int channel)
226     throws IOException  
227   {
228     if (os != null) {
229       os.write('Q');
230       os.flush();
231       freeWriteLock();
232     }
233   }
234   
235   /**
236    * Frees the channel for writing.
237    */
238   void freeWriteLock()
239   {
240     synchronized (WRITE_LOCK) {
241       isWriteLocked = false;
242       WRITE_LOCK.notifyAll();
243     }
244   }
245 
246   /**
247    * Reads data from a channel.
248    *
249    * @param channel the channel
250    *
251    * @return true if the channel is valid.
252    */
253   InputStream   readChannel(int channel)
254     throws IOException  
255   {
256     while (! isClosed) {
257       if (inputReady[channel]) {
258         inputReady[channel] = false;
259         return is;
260       }
261 
262       boolean canRead = false;
263       synchronized (READ_LOCK) {
264         if (! isReadLocked) {
265           isReadLocked = true;
266           canRead = true;
267         }
268         else {
269           try {
270             READ_LOCK.wait(5000);
271           } catch (Exception   e) {
272           }
273         }
274       }
275       
276       if (canRead) {
277         try {
278           readData();
279         } catch (IOException   e) {
280           close();
281         }
282       }
283     }
284 
285     return null;
286   }
287 
288   boolean getReadLock()
289   {
290     synchronized (READ_LOCK) {
291       if (! isReadLocked) {
292         isReadLocked = true;
293         return true;
294       }
295       else {
296         try {
297           READ_LOCK.wait(5000);
298         } catch (Exception   e) {
299         }
300       }
301     }
302 
303     return false;
304   }
305 
306   /**
307    * Frees the channel for reading.
308    */
309   void freeReadLock()
310   {
311     synchronized (READ_LOCK) {
312       isReadLocked = false;
313       READ_LOCK.notifyAll();
314     }
315   }
316 
317   /**
318    * Reads data until a channel packet 'C' or error 'E' is received.
319    */
320   private void readData()
321     throws IOException  
322   {
323     while (! isClosed) {
324       int code = is.read();
325 
326       switch (code) {
327       case ' ':
328       case '\t':
329       case '\n':
330       case '\r':
331         break;
332 
333       case 'C': {
334         int channel = (is.read() << 8) + is.read();
335 
336         inputReady[channel] = true;
337         return;
338       }
339 
340       case 'E': {
341         int channel = (is.read() << 8) + is.read();
342         int status = (is.read() << 8) + is.read();
343 
344         inputReady[channel] = true;
345 
346         return;
347       }
348 
349       case -1:
350         close();
351         return;
352         
353       default:
354         // An error in the protocol.  Kill the connection.
355         close();
356         return;
357       }
358     }
359 
360     return;
361   }
362 
363   /**
364    * Close the mux
365    */
366   public void close()
367     throws IOException  
368   {
369     isClosed = true;
370     
371     OutputStream   os = this.os;
372     this.os = null;
373     
374     InputStream   is = this.is;
375     this.is = null;
376 
377     if (os != null)
378       os.close();
379 
380     if (is != null)
381       is.close();
382   }
383 }
384
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags