ClientConnection


1   /*
2    *        JacORB - a free Java ORB
3    *
4    *   Copyright (C) 1997-2004 Gerald Brose.
5    *
6    *   This library is free software; you can redistribute it and/or
7    *   modify it under the terms of the GNU Library General Public
8    *   License as published by the Free Software Foundation; either
9    *   version 2 of the License, or (at your option) any later version.
10   *
11   *   This library is distributed in the hope that it will be useful,
12   *   but WITHOUT ANY WARRANTY; without even the implied warranty of
13   *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14   *   Library General Public License for more details.
15   *
16   *   You should have received a copy of the GNU Library General Public
17   *   License along with this library; if not, write to the Free
18   *   Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
19   */
20  
21  package org.jacorb.orb.giop;
22  
23  import java.util.*;
24  
25  import org.apache.avalon.framework.logger.Logger;
26  
27  import org.jacorb.orb.CDROutputStream;
28  import org.jacorb.orb.ParsedIOR;
29  
30  import org.omg.IOP.*;
31  import org.omg.CONV_FRAME.*;
32  
33  /**
34   * ClientConnection.java
35   *
36   *
37   * Created: Sat Aug 18 18:37:56 2002
38   *
39   * @author Nicolas Noffke
40   * @version $Id: ClientConnection.java,v 1.55 2005/04/23 13:09:04 phil.mesnier Exp $
41   */
42  
43  public class ClientConnection
44      implements ReplyListener, ConnectionListener
45  {
46      private GIOPConnection connection = null;
47      private org.omg.CORBA.ORB   orb = null;
48  
49      private HashMap replies;
50  
51      // support for SAS Stateful contexts
52      private HashMap sasContexts;
53      private static long last_client_context_id = 0;
54  
55      /* how many clients use this connection? */
56      private int client_count = 0;
57  
58      //to generate request ids
59      private int id_count = 0;
60  
61      private ClientConnectionManager conn_mg = null;
62  
63      private boolean client_initiated = true;
64  
65      private String   info = null;
66  
67      // indicates if the stream has been closed gracefully, i.e. by a
68      // CloseConnection message. This will trigger a remarshaling of
69      // all pending messages.
70      private boolean gracefulStreamClose = false;
71  
72      //The profile that was used for registering with the
73      //ClientConnectionManager. In case of BiDirIIOP it is NOT equal to
74      //the transports profile.
75      private org.omg.ETF.Profile registeredProfile = null;
76  
77      private Logger logger = null;
78  
79      public ClientConnection( GIOPConnection connection,
80                               org.omg.CORBA.ORB   orb,
81                               ClientConnectionManager conn_mg,
82                               org.omg.ETF.Profile registeredProfile,
83                               boolean client_initiated )
84      {
85          this.connection = connection;
86          this.orb = orb;
87          this.conn_mg = conn_mg;
88          this.registeredProfile = registeredProfile;
89          this.info = registeredProfile.toString();
90          this.client_initiated = client_initiated;
91  
92          logger =
93              ((org.jacorb.orb.ORB)orb).getConfiguration().getNamedLogger("jacorb.giop.conn");
94  
95          //For BiDirGIOP, the connection initiator may only generate
96          //even valued request ids, and the other side odd valued
97          //request ids. Therefore, we always step the counter by 2, so
98          //we always get only odd or even ids depending on the counters
99          //initial value.
100         if( ! client_initiated )
101         {
102             id_count = 1;
103         }
104 
105         connection.setReplyListener( this );
106         connection.setConnectionListener( this );
107 
108         replies = new HashMap();
109         sasContexts = new HashMap();
110     }
111 
112     public final GIOPConnection getGIOPConnection()
113     {
114         return connection;
115     }
116 
117     /**
118      * Get the profile that was used for registering with the
119      * ClientConnectionManager. In case of BiDirIIOP it is NOT equal
120      * to the transports profile.
121      */
122     public org.omg.ETF.Profile getRegisteredProfile()
123     {
124         return registeredProfile;
125     }
126 
127     public void setCodeSet( ParsedIOR pior )
128     {
129         if( isTCSNegotiated() )
130         {
131             //if already negotiated, do nothing
132             return;
133         }
134 
135         //if the other side only talks GIOP 1.0, don't send a codeset
136         //context and don't try again
137         if( pior.getEffectiveProfile().version().minor == 0 )
138         {
139             connection.markTCSNegotiated();
140             return;
141         }
142 
143         int tcs = -1;
144         int tcsw = -1;
145 
146         CodeSetComponentInfo info = pior.getCodeSetComponentInfo();
147         if( info != null )
148         {
149             tcs = CodeSet.selectTCS( info );
150             tcsw = CodeSet.selectTCSW( info );
151         }
152         else
153         {
154             if (logger.isDebugEnabled())
155                 logger.debug("No CodeSetComponentInfo in IOR. Will use default CodeSets" );
156 
157             //If we can't find matching codesets, we still mark the
158             //GIOPConnection as negotiated, so the following requests
159             //will not always try to select a codeset again.
160 
161             /* ******
162                until the ETF spec is ammended to include components within
163                the base Profile type, then this is going to be problem. So
164                rather than not setting the codeset component, we should
165                pick reasonable default values and send those.
166             */
167 
168             //connection.markTCSNegotiated();
169             //return null;
170 
171             tcs = CodeSet.getTCSDefault();
172             tcsw = CodeSet.getTCSWDefault();
173         }
174 
175         if( tcs == -1 || tcsw == -1 )
176         {
177             //if no matching codesets can be found, an exception is
178             //thrown
179             throw new org.omg.CORBA.CODESET_INCOMPATIBLE  (
180                 "WARNING: CodeSet negotiation failed! No matching " +
181                 (( tcs == -1 )? "normal" : "wide") +
182                 " CodeSet found");
183         }
184 
185         connection.setCodeSets( tcs, tcsw );
186 
187         if (logger.isDebugEnabled())
188         {
189             logger.debug( "Successfully negotiated Codesets. Using " +
190                           CodeSet.csName( tcs ) + " as TCS and " +
191                           CodeSet.csName( tcsw ) + " as TCSW" );
192         }
193 
194     }
195 
196     public boolean isTCSNegotiated()
197     {
198         return connection.isTCSNegotiated();
199     }
200 
201     public int getTCS()
202     {
203         return connection.getTCS();
204     }
205 
206     public int getTCSW()
207     {
208         return connection.getTCSW();
209     }
210 
211     public String   getInfo()
212     {
213         return info;
214     }
215 
216     public synchronized int getId()
217     {
218         int id = id_count;
219 
220         //if odd or even is determined by the starting value of
221         //id_count
222         id_count += 2;
223 
224         return id;
225     }
226 
227     public synchronized void incClients()
228     {
229         client_count++;
230     }
231 
232     /**
233      * This method decrements the number of clients. If the number reaches
234      * zero it also calls close.
235      *
236      * @return a <code>boolean</code> value, true if client_count is zero.
237      */
238     public synchronized boolean decClients()
239     {
240         boolean result = false;
241 
242         client_count--;
243 
244         if (client_count == 0 )
245         {
246             result = true;
247         }
248         return result;
249     }
250 
251 
252     public boolean isClientInitiated()
253     {
254         return client_initiated;
255     }
256 
257     /**
258      * The request_id parameter is only used, if response_expected.
259      */
260     public void sendRequest( MessageOutputStream os,
261                              ReplyPlaceholder placeholder,
262                              int request_id,
263                              boolean response_expected )
264     {
265         Integer   key = new Integer  ( request_id );
266 
267         synchronized( replies )
268         {
269             replies.put( key, placeholder );
270         }
271 
272         try
273         {
274             sendRequest( os, response_expected );
275         }
276         catch( org.omg.CORBA.SystemException   e )
277         {
278            //remove reply receiver from list
279            //because there will be no response to this request
280            synchronized( replies )
281            {
282               replies.remove( key );
283            }
284            throw e;
285         }
286 
287     }
288 
289     public void sendRequest( MessageOutputStream os,
290                              boolean response_expected )
291     {
292         try
293         {
294             connection.sendRequest( os, response_expected );
295         }
296         catch (java.io.IOException   e)
297         {
298             if (logger.isDebugEnabled())
299                 logger.debug("IOException", e);
300 
301             throw new org.omg.CORBA.COMM_FAILURE  
302                 (0, org.omg.CORBA.CompletionStatus.COMPLETED_MAYBE);
303         }
304     }
305 
306     /**
307      * called from Delegate/ConnectionManagement etc.
308      */
309 
310     public void close()
311     {
312         connection.close();
313     }
314 
315     /**
316      * Operations from ReplyListener
317      */
318 
319     public void replyReceived( byte[] reply,
320                                GIOPConnection connection )
321     {
322         connection.decPendingMessages();
323 
324         Integer   key = new Integer  ( Messages.getRequestId( reply ));
325 
326         ReplyPlaceholder placeholder = null;
327 
328         synchronized( replies )
329         {
330             placeholder =
331                 (ReplyPlaceholder) replies.remove( key );
332         }
333 
334         if( placeholder != null )
335         {
336             ReplyInputStream ris = new ReplyInputStream (orb, reply);
337             ris.setCodeSet (this.getTCS(),this.getTCSW());
338             //this will unblock the waiting thread
339             placeholder.replyReceived(ris);
340         }
341         else
342         {
343             if (logger.isWarnEnabled())
344             {
345                 logger.warn("Received reply for unknown request id: " +
346                     key);
347             }
348         }
349     }
350 
351 
352     public void locateReplyReceived( byte[] reply,
353                                      GIOPConnection connection )
354     {
355         connection.decPendingMessages();
356 
357         Integer   key = new Integer  ( Messages.getRequestId( reply ));
358 
359         ReplyPlaceholder placeholder = null;
360 
361         synchronized( replies )
362         {
363             placeholder =
364                 (ReplyPlaceholder) replies.remove( key );
365         }
366 
367         if( placeholder != null )
368         {
369             //this will unblock the waiting thread
370             placeholder.replyReceived( new LocateReplyInputStream( orb,
371                                                                    reply ));
372         }
373         else
374         {
375             if (logger.isWarnEnabled())
376             {
377                 logger.warn("Received reply for unknown request id: " +
378                     key);
379             }
380         }
381     }
382 
383     /**
384      * Received a CloseConnection message. Remarshal all pending
385      * messages. The close mechanism will be invoked separately by the
386      * actual closing of the Transport and will trigger the
387      * remarshaling.
388      */
389     public void closeConnectionReceived( byte[] close_conn,
390                                          GIOPConnection connection )
391     {
392         if (logger.isInfoEnabled())
393         {
394             logger.info("Received CloseConnection message");
395         }
396 
397         if( client_initiated )
398         {
399             gracefulStreamClose = true;
400             ((ClientGIOPConnection) connection).closeAllowReopen();
401 
402             //since this is run on the message receptor thread itself, it
403             //will not try to read again after returning, because it just
404             //closed the transport itself. Therefore, no exception goes
405             //back up into the GIOPConnection, where streamClosed() will
406             //be called. Ergo, we need to call streamClosed() ourselves.
407             streamClosed();
408         }
409     }
410 
411 
412     /**
413      * Operations from ConnectionListener
414      * used for upcalls from GIOPConnection
415      */
416 
417     public void connectionClosed()
418     {
419         if( ! client_initiated )
420         {
421             //if this is a server side BiDir connection, it will stay
422             //pooled in the ClientConnectionManager even if no Delegate is
423             //associated with it. Therefore, it has to be removed when
424             //the underlying connection closed.
425 
426             conn_mg.removeConnection( this );
427         }
428 
429         streamClosed();
430     }
431 
432     /**
433      * the transport has been
434      * removed underneath the GIOP layer
435      */
436 
437     public void streamClosed()
438     {
439         synchronized( replies )
440         {
441             if( replies.size() > 0 )
442             {
443                 if( gracefulStreamClose )
444                 {
445                     if (logger.isDebugEnabled())
446                     {
447                         logger.debug("Stream closed. Will remarshal " +
448                                      replies.size() + " messages" );
449                     }
450                 }
451                 else
452                 {
453                     if (logger.isWarnEnabled())
454                     {
455                         logger.warn("Abnormal connection termination. Lost " +
456                                      replies.size() + " outstanding replie(s)!");
457                     }
458                 }
459 
460                 Iterator entries = replies.values().iterator();
461                 ReplyPlaceholder placeholder;
462 
463                 while( entries.hasNext() )
464                 {
465                     placeholder = (ReplyPlaceholder)entries.next();
466 
467                     if( gracefulStreamClose )
468                     {
469                         placeholder.retry();
470                     }
471                     else
472                     {
473                         placeholder.cancel();
474                     }
475                     entries.remove();
476                 }
477             }
478         }
479 
480         gracefulStreamClose = false;
481     }
482 
483     public org.omg.ETF.Profile get_server_profile()
484     {
485         return connection.getTransport().get_server_profile();
486     }
487 
488     public long cacheSASContext(byte[] client_authentication_token)
489     {
490         long client_context_id = 0;
491         String   key = new String  (client_authentication_token);
492         synchronized ( sasContexts )
493         {
494             if (!sasContexts.containsKey(key))
495             {
496                 // new context
497                 client_context_id = ++last_client_context_id;
498                 sasContexts.put(key, new Long  (client_context_id));
499                 client_context_id = -client_context_id;
500             }
501             else
502             {
503                 // reuse cached context
504                 client_context_id = ((Long  )sasContexts.get(key)).longValue();
505             }
506         }
507         return client_context_id;
508     }
509 
510     public long purgeSASContext(long client_context_id)
511     {
512         synchronized ( sasContexts )
513         {
514             Iterator entries = sasContexts.keySet().iterator();
515             while( entries.hasNext() )
516             {
517                 Object   key = entries.next();
518                 if (((Long  )sasContexts.get(key)).longValue() != client_context_id)
519                 {
520                     continue;
521                 }
522                 entries.remove();
523                 break;
524             }
525         }
526         return client_context_id;
527     }
528 }// ClientConnection
529
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags