/[classpath]/classpath/gnu/CORBA/Functional_ORB.java
ViewVC logotype

Diff of /classpath/gnu/CORBA/Functional_ORB.java

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.1.2.1 by gnu_andrew, Fri May 20 18:20:49 2005 UTC revision 1.1.2.2 by gnu_andrew, Fri May 27 00:01:44 2005 UTC
# Line 51  import org.omg.CORBA.BAD_OPERATION; Line 51  import org.omg.CORBA.BAD_OPERATION;
51  import org.omg.CORBA.BAD_PARAM;  import org.omg.CORBA.BAD_PARAM;
52  import org.omg.CORBA.CompletionStatus;  import org.omg.CORBA.CompletionStatus;
53  import org.omg.CORBA.MARSHAL;  import org.omg.CORBA.MARSHAL;
54    import org.omg.CORBA.NO_RESOURCES;
55  import org.omg.CORBA.OBJECT_NOT_EXIST;  import org.omg.CORBA.OBJECT_NOT_EXIST;
56  import org.omg.CORBA.ORBPackage.InvalidName;  import org.omg.CORBA.ORBPackage.InvalidName;
57    import org.omg.CORBA.Request;
58  import org.omg.CORBA.SystemException;  import org.omg.CORBA.SystemException;
59  import org.omg.CORBA.UNKNOWN;  import org.omg.CORBA.UNKNOWN;
60  import org.omg.CORBA.portable.Delegate;  import org.omg.CORBA.portable.Delegate;
# Line 70  import java.io.OutputStream; Line 72  import java.io.OutputStream;
72  import java.net.InetAddress;  import java.net.InetAddress;
73  import java.net.ServerSocket;  import java.net.ServerSocket;
74  import java.net.Socket;  import java.net.Socket;
75    import java.net.SocketException;
76  import java.net.UnknownHostException;  import java.net.UnknownHostException;
77    
78    import java.util.ArrayList;
79  import java.util.Enumeration;  import java.util.Enumeration;
80  import java.util.Iterator;  import java.util.Iterator;
81    import java.util.LinkedList;
82  import java.util.Map;  import java.util.Map;
83  import java.util.Properties;  import java.util.Properties;
84  import java.util.StringTokenizer;  import java.util.StringTokenizer;
# Line 89  public class Functional_ORB Line 94  public class Functional_ORB
94    extends Restricted_ORB    extends Restricted_ORB
95  {  {
96    /**    /**
97       * A server, responsible for listening on requests on some
98       * local port. The ORB may listen on multiple ports and process
99       * the requests in separate threads. Normally the server takes
100       * one port per object being served.
101       */
102      class portServer
103        extends Thread
104      {
105        /**
106         * The port on that this portServer is listening for requests.
107         */
108        int s_port;
109    
110        /**
111         * The server socket of this portServer.
112         */
113        ServerSocket service;
114    
115        /**
116         * Create a new portServer, serving on specific port.
117         */
118        portServer(int _port)
119        {
120          s_port = _port;
121        }
122    
123        /**
124         * Enter the serving loop (get request/process it).
125         * All portServer normally terminate thy threads when
126         * the Functional_ORB.running is set to false.
127         */
128        public void run()
129        {
130          boolean terminated;
131    
132          try
133            {
134              service = new ServerSocket(s_port);
135            }
136          catch (IOException ex)
137            {
138              throw new BAD_OPERATION("Unable to open the server socket.");
139            }
140    
141          while (running)
142            {
143              try
144                {
145                  serve(service);
146                }
147              catch (SocketException ex)
148                {
149                  // Thrown when the service is closed by
150                  // the close_now().
151                  return;
152                }
153              catch (IOException iex)
154                {
155                  // Wait 5 seconds. Do not terminate the
156                  // service due potentially transient error.
157                  try
158                    {
159                      Thread.sleep(5000);
160                    }
161                  catch (InterruptedException ex)
162                    {
163                    }
164                }
165            }
166        }
167    
168        /**
169         * Forcibly close the server socket and mark this port as free.
170         */
171        public void close_now()
172        {
173          try
174            {
175              service.close();
176            }
177          catch (Exception ex)
178            {
179              // This may happen if the service has not been opened or
180              // cannot be closed. Return without action.
181            }
182        }
183    
184        /**
185         * If the thread is no longer in use, close the socket (if opened).
186         */
187        protected void finalize()
188        {
189          close_now();
190        }
191      }
192    
193      /**
194     * The property of port, on that this ORB is listening for requests from clients.     * The property of port, on that this ORB is listening for requests from clients.
195     * This class supports one port per ORB only.     * This class supports one port per ORB only.
196     */     */
# Line 143  public class Functional_ORB Line 245  public class Functional_ORB
245    private Map initial_references = new TreeMap();    private Map initial_references = new TreeMap();
246    
247    /**    /**
248       * The currently active portServers.
249       */
250      private ArrayList portServers = new ArrayList();
251    
252      /**
253     * The host, on that the name service is expected to be running.     * The host, on that the name service is expected to be running.
254     */     */
255    private String ns_host;    private String ns_host;
256    
257    /**    /**
258     * The port, under that the ORB is listening for remote requests.     * The port, under that the ORB is listening for remote requests.
259     * In this implementation, all objects, served by this ORB, share     * Then the new object is connected, this port is used first, then
260     * the same port, 1126.     * it is incremented by 1, etc. If the given port is not available,
261       * up to 20 subsequent values are tried and then the parameterless
262       * server socket contructor is called.
263     */     */
264    private int Port = 1126;    private static int Port = 1126;
265    
266    /**    /**
267     * The port, on that the name service is expected to be running.     * The port, on that the name service is expected to be running.
# Line 160  public class Functional_ORB Line 269  public class Functional_ORB
269    private int ns_port = 900;    private int ns_port = 900;
270    
271    /**    /**
272       * The instance, stored in this field, handles the asynchronous dynamic
273       * invocations.
274       */
275      protected Asynchron asynchron = new Asynchron();
276    
277      /**
278       * The list of the freed ports. The ORB reuses ports, when possible.
279       */
280      protected LinkedList freed_ports = new LinkedList();
281    
282      /**
283     * Create the instance of the Functional ORB.     * Create the instance of the Functional ORB.
284     */     */
285    public Functional_ORB()    public Functional_ORB()
# Line 196  public class Functional_ORB Line 316  public class Functional_ORB
316    }    }
317    
318    /**    /**
319       * Get the currently free port, starting from the initially set port
320       * and going up max 20 steps, then trying to bind into any free
321       * address.
322       *
323       * @return the currently available free port.
324       *
325       * @throws NO_RESOURCES if the server socked cannot be opened on the
326       * local host.
327       */
328      public int getFreePort()
329                      throws BAD_OPERATION
330      {
331        ServerSocket s;
332    
333        try
334          {
335            // If there are some previously freed ports, use them first.
336            if (!freed_ports.isEmpty())
337              {
338                Integer free = (Integer) freed_ports.getLast();
339                freed_ports.removeLast();
340                s = new ServerSocket(free.intValue());
341                s.close();
342                return free.intValue();
343              }
344          }
345        catch (Exception ex)
346          {
347            // This may be thrown if the request for the new port has arrived
348            // before the current service is completly shutdown.
349            // OK then, use a new port.
350          }
351    
352        for (int i = Port; i < Port + 20; i++)
353          {
354            try
355              {
356                s = new ServerSocket(i);
357                s.close();
358                Port = i + 1;
359                return s.getLocalPort();
360              }
361            catch (IOException ex)
362              {
363                // Repeat the loop if this exception has been thrown.
364              }
365          }
366    
367        try
368          {
369            // Try any port.
370            s = new ServerSocket();
371            s.close();
372            return s.getLocalPort();
373          }
374        catch (IOException ex1)
375          {
376            throw new NO_RESOURCES("Unable to open the server socket");
377          }
378      }
379    
380      /**
381     * Set the port, on that the server is listening for the client requests.     * Set the port, on that the server is listening for the client requests.
382     * In this implementation, the server is listening at only one port,     * In this implementation, the server is listening at only one port,
383     * the default value being 1126.     * the default value being 1126.
# Line 228  public class Functional_ORB Line 410  public class Functional_ORB
410     * Connect the given CORBA object to this ORB. After the object is     * Connect the given CORBA object to this ORB. After the object is
411     * connected, it starts receiving remote invocations via this ORB.     * connected, it starts receiving remote invocations via this ORB.
412     *     *
413       * The ORB tries to connect the object to the port, that has been
414       * previously set by {@link setPort(int)}. On failure, it tries
415       * 20 subsequent larger values and then calls the parameterless
416       * server socked constructor to get any free local port.
417       * If this fails, the {@link NO_RESOURCES} is thrown.
418       *
419     * @param object the object, must implement the {@link InvokeHandler})     * @param object the object, must implement the {@link InvokeHandler})
420     * interface.     * interface.
421     *     *
# Line 236  public class Functional_ORB Line 424  public class Functional_ORB
424     */     */
425    public void connect(org.omg.CORBA.Object object)    public void connect(org.omg.CORBA.Object object)
426    {    {
427      connected_objects.add(object);      int a_port = getFreePort();
428    
429      IOR ior = createIOR(object, connected_objects.getKey(object));      Connected_objects.cObject ref = connected_objects.add(object, a_port);
430        IOR ior = createIOR(ref);
431      prepareObject(object, ior);      prepareObject(object, ior);
432        if (running)
433          startService(ior);
434    }    }
435    
436    /**    /**
437     * Connect the given CORBA object to this ORB, explicitly specifying     * Connect the given CORBA object to this ORB, explicitly specifying
438     * the object key.     * the object key.
439     *     *
440       * The ORB tries to connect the object to the port, that has been
441       * previously set by {@link setPort(int)}. On failure, it tries
442       * 20 subsequent larger values and then calls the parameterless
443       * server socked constructor to get any free local port.
444       * If this fails, the {@link NO_RESOURCES} is thrown.
445       *
446     * @param object the object, must implement the {@link InvokeHandler})     * @param object the object, must implement the {@link InvokeHandler})
447     * interface.     * interface.
448     * @param key the object key, usually used to identify the object from     * @param key the object key, usually used to identify the object from
449     *  remote side.     * remote side.
450     *     *
451     * @throws BAD_PARAM if the object does not implement the     * @throws BAD_PARAM if the object does not implement the
452     * {@link InvokeHandler}).     * {@link InvokeHandler}).
453     */     */
454    public void connect(org.omg.CORBA.Object object, byte[] key)    public void connect(org.omg.CORBA.Object object, byte[] key)
455    {    {
456      prepareObject(object, createIOR(object, key));      int a_port = getFreePort();
457      connected_objects.add(key, object);  
458        Connected_objects.cObject ref = connected_objects.add(key, object, a_port);
459        IOR ior = createIOR(ref);
460        prepareObject(object, ior);
461        if (running)
462          startService(ior);
463      }
464    
465      /**
466       * Start the service on the given port of this IOR.
467       *
468       * @param ior the ior (only Internet.port is used).
469       */
470      private void startService(IOR ior)
471      {
472        portServer p = new portServer(ior.Internet.port);
473        portServers.add(p);
474        p.start();
475    }    }
476    
477    /**    /**
# Line 266  public class Functional_ORB Line 480  public class Functional_ORB
480    public void destroy()    public void destroy()
481    {    {
482      super.destroy();      super.destroy();
483    
484        portServer p;
485        for (int i = 0; i < portServers.size(); i++)
486          {
487            p = (portServer) portServers.get(i);
488            p.close_now();
489          }
490    }    }
491    
492    /**    /**
# Line 279  public class Functional_ORB Line 500  public class Functional_ORB
500     */     */
501    public void disconnect(org.omg.CORBA.Object object)    public void disconnect(org.omg.CORBA.Object object)
502    {    {
503      connected_objects.remove(object);      Connected_objects.cObject rmKey = null;
504    
505        // Handle the case when it is possible to get the object key.
506        // Handle the case when the object is known, but not local.
507        if (object instanceof ObjectImpl)
508          {
509            Delegate delegate = ((ObjectImpl) object)._get_delegate();
510            if (delegate instanceof Simple_delegate)
511            {
512              byte[] key = ((Simple_delegate) delegate).getIor().key;
513              rmKey = connected_objects.get(key);
514            }
515          }
516    
517        // Try to find and disconned the object that is not an instance of the
518        // object implementation.
519        if (rmKey == null)
520         rmKey = connected_objects.getKey(object);
521    
522        // Disconnect the object on any success.
523        if (rmKey != null)
524          {
525            // Find and stop the corresponding portServer.
526            portServer p;
527            StopService:
528            for (int i = 0; i < portServers.size(); i++)
529              {
530                p = (portServer) portServers.get(i);
531                if (p.s_port == rmKey.port)
532                  {
533                    p.close_now();
534                    freed_ports.addFirst(new Integer(rmKey.port));
535                    break StopService;
536                  }
537                connected_objects.remove(rmKey.key);
538              }
539          }
540    }    }
541    
542    /**    /**
# Line 351  public class Functional_ORB Line 608  public class Functional_ORB
608        }        }
609    
610      // Handle the case when the object is local.      // Handle the case when the object is local.
611      byte[] key = connected_objects.getKey(forObject);      Connected_objects.cObject rec = connected_objects.getKey(forObject);
612    
613      if (key == null)      if (rec == null)
614        throw new BAD_PARAM("The object " + forObject +        throw new BAD_PARAM("The object " + forObject +
615                            " has not been previously connected to this ORB"                            " has not been previously connected to this ORB"
616                           );                           );
617    
618      IOR ior = createIOR(forObject, key);      IOR ior = createIOR(rec);
619    
620      return ior.toStringifiedReference();      return ior.toStringifiedReference();
621    }    }
# Line 406  public class Functional_ORB Line 663  public class Functional_ORB
663     */     */
664    public void run()    public void run()
665    {    {
666      try      running = true;
667    
668        // Instantiate the port server for each socket.
669        Iterator iter = connected_objects.entrySet().iterator();
670        Map.Entry m;
671        Connected_objects.cObject obj;
672    
673        while (iter.hasNext())
674        {        {
675          ServerSocket service = new ServerSocket(Port);          m = (Map.Entry) iter.next();
676          running = true;          obj = (Connected_objects.cObject) m.getValue();
677          while (running)  
678            portServer subserver = new portServer(obj.port);
679            portServers.add(subserver);
680    
681            // Reuse the current thread for the last portServer.
682            if (!iter.hasNext())
683            {            {
684              serve(service);              // Discard the iterator, eliminating lock checks.
685                iter = null;
686                subserver.run();
687                return;
688            }            }
689        }          else
690      catch (IOException ex)            subserver.start();
       {  
         throw new BAD_OPERATION("Unable to open the server socket.");  
691        }        }
692    }    }
693    
# Line 425  public class Functional_ORB Line 695  public class Functional_ORB
695     * Shutdown the ORB server.     * Shutdown the ORB server.
696     *     *
697     * @param wait_for_completion if true, the current thread is     * @param wait_for_completion if true, the current thread is
698     * suspended untile the shutdown process is complete.     * suspended until the shutdown process is complete.
699     */     */
700    public void shutdown(boolean wait_for_completion)    public void shutdown(boolean wait_for_completion)
701    {    {
702      super.shutdown(wait_for_completion);      super.shutdown(wait_for_completion);
703        running = false;
704    
705        if (!wait_for_completion)
706          {
707            for (int i = 0; i < portServers.size(); i++)
708              {
709                portServer p = (portServer) portServers.get(i);
710                p.close_now();
711              }
712          }
713    }    }
714    
715    /**    /**
# Line 465  public class Functional_ORB Line 745  public class Functional_ORB
745            }            }
746    
747          object = impl;          object = impl;
748          connected_objects.add(ior.key, impl);          connected_objects.add(ior.key, impl, ior.Internet.port);
749        }        }
750      return object;      return object;
751    }    }
# Line 503  public class Functional_ORB Line 783  public class Functional_ORB
783     */     */
784    protected org.omg.CORBA.Object find_connected_object(byte[] key)    protected org.omg.CORBA.Object find_connected_object(byte[] key)
785    {    {
786      return connected_objects.get(key);      Connected_objects.cObject ref = connected_objects.get(key);
787        return ref==null?null:ref.object;
788    }    }
789    
790    /**    /**
# Line 603  public class Functional_ORB Line 884  public class Functional_ORB
884      useProperties(props);      useProperties(props);
885    }    }
886    
887    private IOR createIOR(org.omg.CORBA.Object forObject, byte[] key)    private IOR createIOR(Connected_objects.cObject ref)
888                   throws BAD_OPERATION                   throws BAD_OPERATION
889    {    {
890      IOR ior = new IOR();      IOR ior = new IOR();
891      ior.key = key;      ior.key = ref.key;
892      ior.Internet.port = Port;      ior.Internet.port = ref.port;
893    
894      if (forObject instanceof ObjectImpl)      if (ref.object instanceof ObjectImpl)
895        {        {
896          ObjectImpl imp = (ObjectImpl) forObject;          ObjectImpl imp = (ObjectImpl) ref.object;
897          if (imp._ids().length > 0)          if (imp._ids().length > 0)
898            ior.Id = imp._ids() [ 0 ];            ior.Id = imp._ids() [ 0 ];
899        }        }
900      if (ior.Id == null)      if (ior.Id == null)
901        ior.Id = forObject.getClass().getName();        ior.Id = ref.object.getClass().getName();
902    
903      try      try
904        {        {
905          ior.Internet.host = InetAddress.getLocalHost().getHostAddress();          ior.Internet.host = InetAddress.getLocalHost().getHostAddress();
906          ior.Internet.port = Port;          ior.Internet.port = ref.port;
907        }        }
908      catch (UnknownHostException ex)      catch (UnknownHostException ex)
909        {        {
# Line 691  public class Functional_ORB Line 972  public class Functional_ORB
972      ReplyHeader reply = handler.reply_header;      ReplyHeader reply = handler.reply_header;
973    
974      if (sysEx != null)      if (sysEx != null)
975        reply.reply_status = reply.SYSTEM_EXCEPTION;        reply.reply_status = ReplyHeader.SYSTEM_EXCEPTION;
976      else if (handler.isExceptionReply())      else if (handler.isExceptionReply())
977        reply.reply_status = reply.USER_EXCEPTION;        reply.reply_status = ReplyHeader.USER_EXCEPTION;
978      else      else
979        reply.reply_status = reply.NO_EXCEPTION;        reply.reply_status = ReplyHeader.NO_EXCEPTION;
980    
981      reply.request_id = rh_request.request_id;      reply.request_id = rh_request.request_id;
982    
# Line 711  public class Functional_ORB Line 992  public class Functional_ORB
992    
993      MessageHeader msh_reply = new MessageHeader();      MessageHeader msh_reply = new MessageHeader();
994      msh_reply.version = msh_request.version;      msh_reply.version = msh_request.version;
995      msh_reply.message_type = msh_reply.REPLY;      msh_reply.message_type = MessageHeader.REPLY;
996      msh_reply.message_size = out.buffer.size();      msh_reply.message_size = out.buffer.size();
997    
998      // Write the reply.      // Write the reply.
# Line 764  public class Functional_ORB Line 1045  public class Functional_ORB
1045              n = in.read(r, n, r.length - n);              n = in.read(r, n, r.length - n);
1046            }            }
1047    
1048          if (msh_request.message_type == msh_request.REQUEST)          if (msh_request.message_type == MessageHeader.REQUEST)
1049            {            {
1050              RequestHeader rh_request;              RequestHeader rh_request;
1051    
# Line 879  public class Functional_ORB Line 1160  public class Functional_ORB
1160            }            }
1161        }        }
1162    }    }
1163    
1164      /**
1165       * Get the next instance with a response being received. If all currently
1166       * sent responses not yet processed, this method pauses till at least one of
1167       * them is complete. If there are no requests currently sent, the method
1168       * pauses till some request is submitted and the response is received.
1169       * This strategy is identical to the one accepted by Suns 1.4 ORB
1170       * implementation.
1171       *
1172       * The returned response is removed from the list of the currently
1173       * submitted responses and is never returned again.
1174       *
1175       * @return the previously sent request that now contains the received
1176       * response.
1177       *
1178       * @throws WrongTransaction If the method was called from the transaction
1179       * scope different than the one, used to send the request. The exception
1180       * can be raised only if the request is implicitly associated with some
1181       * particular transaction.
1182       */
1183      public Request get_next_response()
1184                                throws org.omg.CORBA.WrongTransaction
1185      {
1186        return asynchron.get_next_response();
1187      }
1188    
1189      /**
1190       * Find if any of the requests that have been previously sent with
1191       * {@link #send_multiple_requests_deferred}, have a response yet.
1192       *
1193       * @return true if there is at least one response to the previously
1194       * sent request, false otherwise.
1195       */
1196      public boolean poll_next_response()
1197      {
1198        return asynchron.poll_next_response();
1199      }
1200    
1201      /**
1202       * Send multiple prepared requests expecting to get a reply. All requests
1203       * are send in parallel, each in its own separate thread. When the
1204       * reply arrives, it is stored in the agreed fields of the corresponing
1205       * request data structure. If this method is called repeatedly,
1206       * the new requests are added to the set of the currently sent requests,
1207       * but the old set is not discarded.
1208       *
1209       * @param requests the prepared array of requests.
1210       *
1211       * @see #poll_next_response()
1212       * @see #get_next_response()
1213       * @see Request#send_deferred()
1214       */
1215      public void send_multiple_requests_deferred(Request[] requests)
1216      {
1217        asynchron.send_multiple_requests_deferred(requests);
1218      }
1219    
1220      /**
1221       * Send multiple prepared requests one way, do not caring about the answer.
1222       * The messages, containing requests, will be marked, indicating that
1223       * the sender is not expecting to get a reply.
1224       *
1225       * @param requests the prepared array of requests.
1226       *
1227       * @see Request#send_oneway()
1228       */
1229      public void send_multiple_requests_oneway(Request[] requests)
1230      {
1231        asynchron.send_multiple_requests_oneway(requests);
1232      }
1233    
1234      /**
1235       * Set the flag, forcing all server threads to terminate.
1236       */
1237      protected void finalize()
1238                       throws java.lang.Throwable
1239      {
1240        running = false;
1241        super.finalize();
1242      }
1243  }  }
1244    
1245    

Legend:
Removed from v.1.1.2.1  
changed lines
  Added in v.1.1.2.2

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26