home *** CD-ROM | disk | FTP | other *** search
/ OS/2 Shareware BBS: 10 Tools / 10-Tools.zip / fnb101.zip / Lib / site-packages / Fnorb / orb / GIOPClientWorkerReactive.py < prev    next >
Text File  |  1999-06-28  |  10KB  |  328 lines
  1. #!/usr/bin/env python
  2. #############################################################################
  3. # Copyright (C) DSTC Pty Ltd (ACN 052 372 577) 1997, 1998, 1999
  4. # All Rights Reserved.
  5. #
  6. # The software contained on this media is the property of the DSTC Pty
  7. # Ltd.  Use of this software is strictly in accordance with the
  8. # license agreement in the accompanying LICENSE.HTML file.  If your
  9. # distribution of this software does not contain a LICENSE.HTML file
  10. # then you have no rights to use this software in any manner and
  11. # should contact DSTC at the address below to determine an appropriate
  12. # licensing arrangement.
  13. # 
  14. #      DSTC Pty Ltd
  15. #      Level 7, GP South
  16. #      Staff House Road
  17. #      University of Queensland
  18. #      St Lucia, 4072
  19. #      Australia
  20. #      Tel: +61 7 3365 4310
  21. #      Fax: +61 7 3365 4311
  22. #      Email: enquiries@dstc.edu.au
  23. # 
  24. # This software is being provided "AS IS" without warranty of any
  25. # kind.  In no event shall DSTC Pty Ltd be liable for damage of any
  26. # kind arising out of or in connection with the use or performance of
  27. # this software.
  28. #
  29. # Project:      Fnorb
  30. # File:         $Source: /units/arch/src/Fnorb/orb/RCS/GIOPClientWorkerReactive.py,v $
  31. # Version:      @(#)$RCSfile: GIOPClientWorkerReactive.py,v $ $Revision: 1.6 $
  32. #
  33. #############################################################################
  34. """ GIOPClientWorkerReactive class. """
  35.  
  36.  
  37. # Fnorb modules.
  38. import CORBA, EventHandler, GIOPClientWorker, GIOPConnectionHandler, Reactor
  39.  
  40.  
  41. class GIOPClientWorkerReactive(GIOPClientWorker.GIOPClientWorker,
  42.                    EventHandler.EventHandler):
  43.     """ GIOPClientWorkerReactive class.
  44.  
  45.     A concrete 'EventHandler' in the 'Reactor' pattern.
  46.  
  47.     This class is *not* thread safe, but since the whole point of the 'Reactor'
  48.     pattern is to work in a single-threaded environment, I don't see this as
  49.     much of a problem ;^)
  50.  
  51.     """
  52.  
  53.     def __init__(self, protocol, address):
  54.     """ Provide an interface to a remote object!
  55.  
  56.     'protocol' is the protocol used by this worker.
  57.     'address'  is the address of the remote object.
  58.  
  59.     """
  60.     # fixme: We don't need to keep these as instance variables... (maybe
  61.     # for unpickling?
  62.     self.__protocol = protocol
  63.     self.__address = address
  64.  
  65.     # This flag is set when we get a CloseConnection message, or an
  66.     # exception event.
  67.     self.__closed = 0
  68.  
  69.     # Each request has a unique id. associated with it.
  70.     self.__request_id = 0
  71.  
  72.     # Queue of outgoing messages.
  73.     self.__outgoing = [] # [(RequestId, Message)]
  74.  
  75.     # Dictionary of complete replies.
  76.     self.__replies = {} # {RequestId: (ReplyHeader, Cursor)}
  77.  
  78.      # Ask the protocol to create a connection and actually connect to the
  79.      # remote object.
  80.      self.__connection = protocol.create_connection()
  81.      self.__connection.connect(self.__address)
  82.  
  83.     # Create a handler to look after the connection.
  84.     self.__handler = GIOPConnectionHandler.GIOPConnectionHandler \
  85.              (self, self.__connection)
  86.  
  87.     # Get a reference to the active Reactor.
  88.     #
  89.     # fixme: Should the reactor be obtained from the protocol?
  90.     self.__reactor = Reactor.Reactor_init()
  91.  
  92.     # Register our interest in read events with the Reactor.
  93.     self.__reactor.register_handler(self, Reactor.READ)
  94.  
  95.     return
  96.  
  97.     def pseudo__del__(self):
  98.     """ Pseudo destructor to remove circular references.
  99.  
  100.     This method is called from the 'pseudo__del__' of the GIOPClient that
  101.     we belong to.
  102.  
  103.     """
  104.     # Close down the worker.
  105.     self.__close()
  106.  
  107.     # The handler holds a (circular) reference to this instance so we
  108.     # have to explicitly clean it up.
  109.     self.__handler.pseudo__del__()
  110.  
  111.     # Clean up *our* reference to *it*!
  112.     del self.__handler
  113.  
  114.     return
  115.  
  116.     #########################################################################
  117.     # GIOPClientWorker interface.
  118.     #########################################################################
  119.  
  120.     def send(self, request_id, message):
  121.     """ Send an operation request to the remote object.
  122.  
  123.     This method WAITS for the message to be sent.
  124.  
  125.     """
  126.     # Add the outgoing message to the queue.
  127.     self.__outgoing.append((request_id, message))
  128.  
  129.     # Register my interest in write events with the Reactor.
  130.     self.__reactor.register_handler(self, Reactor.WRITE)
  131.  
  132.     # Get the reactor to process events until the message has been sent.
  133.     while 1:
  134.         for (id, message) in self.__outgoing:
  135.         if request_id == id:
  136.             break
  137.  
  138.         # Else, the request has been sent!
  139.         else:
  140.         break
  141.  
  142.         # Get the reactor to wait for and process a single event.
  143.         self.__reactor.do_one_event()
  144.  
  145.     return
  146.  
  147.     def recv(self, request_id):
  148.     """ Wait for a specific reply. """
  149.  
  150.     # Get the reactor to process events until we either get a close event
  151.     # or we get a reply.
  152.     while 1:
  153.         # Have we got our reply yet?
  154.         try:
  155.         reply = self.__replies[request_id]
  156.         del self.__replies[request_id]
  157.         break
  158.  
  159.         # Err, nope!
  160.         except KeyError:
  161.         pass
  162.  
  163.         # Get the reactor to wait for and process a single event.
  164.         self.__reactor.do_one_event()
  165.  
  166.     return reply
  167.  
  168.     def poll(self, request_id):
  169.     """ Poll for a reply to a specific request. """
  170.  
  171.     # Get the reactor to poll for a single event.
  172.     self.__reactor.do_one_event(0)
  173.  
  174.     # Have we got our reply yet?
  175.     return self.__replies.has_key(request_id)
  176.  
  177.     def peek(self, request_id):
  178.     """ Peek at the reply for the specified request.
  179.  
  180.     This method does *not* delete the reply from the client's queue.
  181.  
  182.     """
  183.     # Get the reply.
  184.     return self.__replies[request_id]
  185.  
  186.     def delete_reply(self, request_id):
  187.     """ Delete the reply with the specified request id. """
  188.  
  189.     # Delete the reply.
  190.     del self.__replies[request_id]
  191.  
  192.     return
  193.  
  194.     def next_request_id(self):
  195.     """ Return the next request id. """
  196.  
  197.     # fixme: This will get stupidly big oneday!!!!
  198.     self.__request_id = self.__request_id + 1
  199.  
  200.     return self.__request_id
  201.  
  202.     def is_closed(self):
  203.     """ Has the worker received a close event? """
  204.  
  205.     return self.__closed
  206.  
  207.     def close_connection(self):
  208.     """ Orderly shutdown, in repsonse to a 'CloseConnection' message. """
  209.  
  210.     if not self.__closed:
  211.         # Close down the worker.
  212.         self.__close()
  213.         
  214.         # We notify our 'GIOPClient' that a 'CloseConnection' message
  215.         # was received by raising a 'CORBA.TRANSIENT' system exception.
  216.         raise CORBA.TRANSIENT()
  217.  
  218.     return
  219.  
  220.     #########################################################################
  221.     # EventHandler interface.
  222.     #########################################################################
  223.  
  224.     def handle_event(self, mask):
  225.     """ Callback method to handle all events except close events. """
  226.  
  227.     # Read event.
  228.     if mask & Reactor.READ:
  229.         self.__handler.recv()
  230.  
  231.     # Write event.
  232.     if mask & Reactor.WRITE:
  233.         # Get the message at the head of the outgoing queue.
  234.         (request_id, message) = self.__outgoing[0]
  235.  
  236.         # Send it!
  237.         self.__handler.send(message)
  238.  
  239.     # Exception event.
  240.     if mask & Reactor.EXCEPTION:
  241.         # Close down the worker.
  242.         self.__close()
  243.  
  244.         # Barf!
  245.         raise CORBA.COMM_FAILURE() # System exception.
  246.  
  247.     return
  248.  
  249.     def handle_close(self):
  250.     """ Callback method to handle close events. """
  251.  
  252.     # Close down the worker.
  253.     self.__close()
  254.  
  255.     return
  256.  
  257.     def handle(self):
  258.     """ Return my underlying I/O handle.
  259.  
  260.     In this case, my I/O handle is the file descriptor of my socket. 
  261.  
  262.     """
  263.     return self.__connection.handle()
  264.  
  265.     #########################################################################
  266.     # GIOPConnectionHandlerListener interface.
  267.     #########################################################################
  268.  
  269.     def message_received(self, message):
  270.     """ Called when a complete GIOP message has been received. """
  271.  
  272.     # Code re-use! The protected method '_message_received' is implemented
  273.     # in the 'GIOPClientWorker' class.
  274.     #
  275.     # If the message is a GIOP Reply message, then this method will call
  276.     # our '_reply_received' method.
  277.     self._message_received(message)
  278.  
  279.     return
  280.  
  281.     def message_sent(self):
  282.     """ Called when a complete GIOP message has been sent. """
  283.  
  284.     # Get the details of the message that has just been sent.
  285.     (request_id, message) = self.__outgoing[0]
  286.  
  287.     # Remove the message from the outgoing queue.
  288.     del self.__outgoing[0]
  289.  
  290.     # If there are no other messages left to send then tell the Reactor
  291.     # that I am no longer interested in write events.
  292.     if len(self.__outgoing) == 0:
  293.         self.__reactor.unregister_handler(self, Reactor.WRITE)
  294.  
  295.     return
  296.  
  297.     #########################################################################
  298.     # Protected interface.
  299.     #########################################################################
  300.  
  301.     def _reply_received(self, reply_header, cursor):
  302.     """ Called by the '_message_received' method of GIOPClientWorker. """
  303.  
  304.     # Add the reply to our dictionary of complete replies.
  305.     self.__replies[reply_header.request_id] = (reply_header, cursor)
  306.  
  307.     return
  308.  
  309.     #########################################################################
  310.     # Private interface.
  311.     #########################################################################
  312.     
  313.     def __close(self):
  314.     """ Close down the worker. """
  315.  
  316.     # Withdraw all of my Reactor registrations.
  317.     self.__reactor.unregister_handler(self, Reactor.ALL)
  318.  
  319.     # Close our connection.
  320.     self.__connection.disconnect()
  321.  
  322.     # All done!
  323.     self.__closed = 1
  324.  
  325.     return
  326.  
  327. #############################################################################
  328.