Channel


1   /**
2    * Redistribution and use of this software and associated documentation
3    * ("Software"), with or without modification, are permitted provided
4    * that the following conditions are met:
5    *
6    * 1. Redistributions of source code must retain copyright
7    *    statements and notices.  Redistributions must also contain a
8    *    copy of this document.
9    *
10   * 2. Redistributions in binary form must reproduce the
11   *    above copyright notice, this list of conditions and the
12   *    following disclaimer in the documentation and/or other
13   *    materials provided with the distribution.
14   *
15   * 3. The name "Exolab" must not be used to endorse or promote
16   *    products derived from this Software without prior written
17   *    permission of Exoffice Technologies.  For written permission,
18   *    please contact info@exolab.org.
19   *
20   * 4. Products derived from this Software may not be called "Exolab"
21   *    nor may "Exolab" appear in their names without prior written
22   *    permission of Exoffice Technologies. Exolab is a registered
23   *    trademark of Exoffice Technologies.
24   *
25   * 5. Due credit should be given to the Exolab Project
26   *    (http://www.exolab.org/).
27   *
28   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
29   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
30   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
31   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
32   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
33   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
34   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
35   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
36   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
37   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
38   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
39   * OF THE POSSIBILITY OF SUCH DAMAGE.
40   *
41   * Copyright 2003-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
42   *
43   * $Id: Channel.java,v 1.3 2005/06/04 14:28:53 tanderson Exp $
44   */
45  package org.exolab.jms.net.multiplexer;
46  
47  import java.io.EOFException  ;
48  import java.io.IOException  ;
49  import java.io.ObjectInputStream  ;
50  import java.io.ObjectOutputStream  ;
51  import java.rmi.MarshalException  ;
52  import java.rmi.RemoteException  ;
53  import java.rmi.UnmarshalException  ;
54  
55  import org.apache.commons.logging.Log;
56  import org.apache.commons.logging.LogFactory;
57  
58  import org.exolab.jms.net.connector.Request;
59  import org.exolab.jms.net.connector.Response;
60  
61  
62  /**
63   * A <code>Channel</code> represents a single-threaded virtual connection over a
64   * physical connection.
65   *
66   * @author <a HREF="mailto:tma@netspace.net.au">Tim Anderson</a>
67   * @version $Revision: 1.3 $ $Date: 2005/06/04 14:28:53 $
68   */
69  class Channel implements Constants {
70  
71      /**
72       * The channel identifier
73       */
74      private int _id;
75  
76      /**
77       * The multiplexer
78       */
79      private Multiplexer _multiplexer;
80  
81      /**
82       * The input stream
83       */
84      private MultiplexInputStream _in;
85  
86      /**
87       * The output stream
88       */
89      private MultiplexOutputStream _out;
90  
91      /**
92       * Synchronization helper used when pinging the connection
93       */
94      private final Object   _lock = new Object  ();
95  
96      /**
97       * Determines if this channel is closed
98       */
99      private volatile boolean _closed = false;
100 
101     /**
102      * The logger
103      */
104     private static final Log _log = LogFactory.getLog(Channel.class);
105 
106 
107     /**
108      * Construct a new <code>Channel</code>
109      *
110      * @param id          the identifier for this channel
111      * @param multiplexer the multiplexer
112      * @param in          the stream to receive data on
113      * @param out         the stream to send data on
114      */
115     public Channel(int id, Multiplexer multiplexer,
116                    MultiplexInputStream in, MultiplexOutputStream out) {
117         _id = id;
118         _multiplexer = multiplexer;
119         _in = in;
120         _out = out;
121     }
122 
123     /**
124      * Returns the channel identifier
125      *
126      * @return the channel identifier
127      */
128     public int getId() {
129         return _id;
130     }
131 
132     /**
133      * Invoke a method on a remote object
134      *
135      * @param request    the request
136      * @return the result of the invocation
137      * @throws RemoteException if the distributed call cannot be made
138      */
139     public Response invoke(Request request) throws RemoteException   {
140         if (_log.isDebugEnabled()) {
141             _log.debug("invoke() [channel=" + _id + "]");
142         }
143         Response response;
144         ObjectOutputStream   out = null;
145         try {
146             // set the packet type
147             _out.setType(REQUEST);
148 
149             // write the request
150             out = new ObjectOutputStream  (_out);
151             request.write(out);
152         } catch (IOException   exception) {
153             throw new MarshalException  ("Failed to marshal call", exception);
154         } catch (Exception   exception) {
155             throw new MarshalException  ("Failed to marshal call", exception);
156         } finally {
157             if (out != null) {
158                 try {
159                     out.close();
160                 } catch (IOException   ignore) {
161                 }
162             }
163         }
164 
165         // read the response
166         ObjectInputStream   in = null;
167         try {
168             in = new ObjectInputStream  (_in);
169             response = Response.read(in, request.getMethod());
170         } catch (ClassNotFoundException   exception) {
171             throw new UnmarshalException  ("Failed to unmarshal response",
172                                          exception);
173         } catch (IOException   exception) {
174             throw new UnmarshalException  ("Failed to unmarshal response",
175                                          exception);
176         } finally {
177             try {
178                 if (in != null) {
179                     in.close();
180                 }
181             } catch (IOException   ignore) {
182             }
183         }
184         if (_log.isDebugEnabled()) {
185             _log.debug("invoke() [channel=" + _id + "] - end");
186         }
187         return response;
188     }
189 
190     /**
191      * Ping the connection
192      *
193      * @throws IOException if the ping fails
194      */
195     public void ping() throws IOException   {
196         if (_log.isDebugEnabled()) {
197             _log.debug("ping [channel=" + _id + "]");
198         }
199         synchronized (_lock) {
200             _multiplexer.send(PING_REQUEST, _id);
201             checkDisconnection();
202             try {
203                 _lock.wait();
204             } catch (InterruptedException   exception) {
205                 throw new IOException  (exception.getMessage());
206             }
207             checkDisconnection();
208         }
209         if (_log.isDebugEnabled()) {
210             _log.debug("ping [channel=" + _id + "] - end");
211         }
212     }
213 
214     /**
215      * Read a request from the channel
216      *
217      * @return the request
218      * @throws IOException if the request can't be read
219      * @todo synchronization required due to scheduling in Multiplexer?
220      */
221     public synchronized Request readRequest() throws IOException   {
222         Request request;
223         ObjectInputStream   in = new ObjectInputStream  (_in);
224         request = Request.read(in);
225         return request;
226     }
227 
228     /**
229      * Write a response to a request
230      *
231      * @param response the response to write
232      * @throws IOException for any I/O error
233      * @todo synchronization required due to scheduling in Multiplexer?
234      */
235     public synchronized void writeResponse(Response response)
236             throws IOException   {
237         // set the packet type
238         _out.setType(RESPONSE);
239 
240         // write the response
241         ObjectOutputStream   out = new ObjectOutputStream  (_out);
242         try {
243             response.write(out);
244         } finally {
245             out.close();
246         }
247     }
248 
249     /**
250      * Handle a ping request
251      *
252      * @throws IOException for any I/O error
253      */
254     public void handlePingRequest() throws IOException   {
255         _multiplexer.send(PING_RESPONSE, _id);
256     }
257 
258     /**
259      * Handle a ping repsonse
260      */
261     public void handlePingResponse() {
262         notifyLock();
263     }
264 
265     /**
266      * Invoked when the underlying physical connection is closed
267      */
268     public void disconnected() {
269         if (_log.isDebugEnabled()) {
270             _log.debug("disconnected [channel=" + _id + "]");
271         }
272         _closed = true;
273         _in.disconnected();
274         _out.disconnected();
275         notifyLock();  // unblock ping
276     }
277 
278     /**
279      * Returns the underlying input stream
280      *
281      * @return the underlying input stream
282      */
283     public MultiplexInputStream getMultiplexInputStream() {
284         return _in;
285     }
286 
287     /**
288      * Returns the underlying output stream
289      *
290      * @return the underlying output stream
291      */
292     public MultiplexOutputStream getMultiplexOutputStream() {
293         return _out;
294     }
295 
296     /**
297      * Releases this channel for re-use
298      */
299     public void release() {
300         notifyLock();
301         _multiplexer.release(this);
302     }
303 
304     /**
305      * Closes this channel
306      *
307      * @throws IOException for any I/O error
308      */
309     public void close() throws IOException   {
310         if (_multiplexer != null) {
311             _closed = true;
312             notifyLock();
313             try {
314                 _multiplexer.close(this);
315             } finally {
316                 _multiplexer = null;
317 
318                 try {
319                     _in.destroy();
320                 } catch (IOException   ignore) {
321                     // @todo
322                 }
323                 try {
324                     _out.close();
325                 } catch (IOException   ignore) {
326                     // @todo
327                 }
328             }
329         }
330     }
331 
332     /**
333      * Destroy this channel
334      */
335     public void destroy() {
336         try {
337             close();
338         } catch (IOException   exception) {
339             // @todo
340         }
341     }
342 
343     /**
344      * Returns a string representation of this.
345      *
346      * @return a string representation of this
347      */
348     public String   toString() {
349         return "Channel[id=" + _id + ", out=" + _out + ", in=" + _in + " ]";
350     }
351 
352     /**
353      * Notifies all threads blocking on the 'ping' lock
354      */
355     private void notifyLock() {
356         synchronized (_lock) {
357             _lock.notifyAll();
358         }
359     }
360 
361     /**
362      * Helper to checks for disconnectiong, and raise an EOFException
363      *
364      * @throws EOFException if disconnection has occurred
365      */
366     private void checkDisconnection() throws EOFException   {
367         if (_closed) {
368             // channel was disconnected
369             throw new EOFException  ("Channel disconnected");
370         }
371     }
372 }
373
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags