0001 # Implements _both_ a connectable client, and a connectable server. 0002 # 0003 # Note that we cheat just a little - the Server in this demo is not created 0004 # via Normal COM - this means we can avoid registering the server. 0005 # However, the server _is_ accessed as a COM object - just the creation 0006 # is cheated on - so this is still working as a fully-fledged server. 0007 0008 import pythoncom 0009 import win32com.server.util 0010 import win32com.server.connect 0011 from win32com.server.exception import Exception 0012 0013 # This is the IID of the Events interface both Client and Server support. 0014 IID_IConnectDemoEvents = pythoncom.MakeIID("{A4988850-49C3-11d0-AE5D-52342E000000}") 0015 0016 # The server which implements 0017 # Create a connectable class, that has a single public method 0018 # 'DoIt', which echos to a single sink 'DoneIt' 0019 0020 class ConnectableServer(win32com.server.connect.ConnectableServer): 0021 _public_methods_ = ["DoIt"] + win32com.server.connect.ConnectableServer._public_methods_ 0022 _connect_interfaces_ = [IID_IConnectDemoEvents] 0023 # The single public method that the client can call on us 0024 # (ie, as a normal COM server, this exposes just this single method. 0025 def DoIt(self,arg): 0026 # Simply broadcast a notification. 0027 self._BroadcastNotify(self.NotifyDoneIt, (arg,)) 0028 0029 def NotifyDoneIt(self, interface, arg): 0030 interface.Invoke(1000, 0, pythoncom.DISPATCH_METHOD, 1, arg) 0031 0032 # Here is the client side of the connection world. 0033 # Define a COM object which implements the methods defined by the 0034 # IConnectDemoEvents interface. 0035 class ConnectableClient: 0036 # This is another cheat - I _know_ the server defines the "DoneIt" event 0037 # as DISPID==1000 - I also know from the implementation details of COM 0038 # that the first method in _public_methods_ gets 1000. 0039 # Normally some explicit DISPID->Method mapping is required. 0040 _public_methods_ = ["OnDoneIt"] 0041 def __init__(self): 0042 self.last_event_arg = None 0043 # A client must implement QI, and respond to a query for the Event interface. 0044 # In addition, it must provide a COM object (which server.util.wrap) does. 0045 def _query_interface_(self, iid): 0046 import win32com.server.util 0047 # Note that this seems like a necessary hack. I am responding to IID_IConnectDemoEvents 0048 # but only creating an IDispatch gateway object. 0049 if iid==IID_IConnectDemoEvents: return win32com.server.util.wrap(self) 0050 # And here is our event method which gets called. 0051 def OnDoneIt(self, arg): 0052 self.last_event_arg = arg 0053 0054 def CheckEvent(server, client, val, verbose): 0055 client.last_event_arg = None 0056 server.DoIt(val) 0057 if client.last_event_arg != val: 0058 raise RuntimeError, "Sent %r, but got back %r" % (val, client.last_event_arg) 0059 if verbose: 0060 print "Sent and received %r" % val 0061 0062 # A simple test script for all this. 0063 # In the real world, it is likely that the code controlling the server 0064 # will be in the same class as that getting the notifications. 0065 def test(verbose=0): 0066 import win32com.client.dynamic, win32com.client.connect 0067 import win32com.server.policy 0068 server = win32com.client.dynamic.Dispatch(win32com.server.util.wrap(ConnectableServer())) 0069 connection = win32com.client.connect.SimpleConnection() 0070 client = ConnectableClient() 0071 connection.Connect(server, client, IID_IConnectDemoEvents) 0072 CheckEvent(server, client, "Hello", verbose) 0073 CheckEvent(server, client, "Here is a null>"+chr(0)+"<", verbose) 0074 CheckEvent(server, client, u"Here is a null>"+unichr(0)+"<", verbose) 0075 val = unicode("test-\xe0\xf2", "latin-1") # 2 latin characters. 0076 CheckEvent(server, client, val, verbose) 0077 if verbose: 0078 print "Everything seemed to work!" 0079 # Aggressive memory leak checking (ie, do nothing!) :-) All should cleanup OK??? 0080 0081 if __name__=='__main__': 0082 test(1) 0083
Generated by PyXR 0.9.4