1# Implements _both_ a connectable client, and a connectable server.
2#
3# Note that we cheat just a little - the Server in this demo is not created
4# via Normal COM - this means we can avoid registering the server.
5# However, the server _is_ accessed as a COM object - just the creation
6# is cheated on - so this is still working as a fully-fledged server.
7
8import pythoncom
9import win32com.server.util
10import win32com.server.connect
11from win32com.server.exception import Exception
12from pywin32_testutil import str2bytes
13
14# This is the IID of the Events interface both Client and Server support.
15IID_IConnectDemoEvents = pythoncom.MakeIID("{A4988850-49C3-11d0-AE5D-52342E000000}")
16
17# The server which implements
18# Create a connectable class, that has a single public method
19# 'DoIt', which echos to a single sink 'DoneIt'
20
21
22class ConnectableServer(win32com.server.connect.ConnectableServer):
23 _public_methods_ = [
24 "DoIt"
25 ] + win32com.server.connect.ConnectableServer._public_methods_
26 _connect_interfaces_ = [IID_IConnectDemoEvents]
27 # The single public method that the client can call on us
28 # (ie, as a normal COM server, this exposes just this single method.
29 def DoIt(self, arg):
30 # Simply broadcast a notification.
31 self._BroadcastNotify(self.NotifyDoneIt, (arg,))
32
33 def NotifyDoneIt(self, interface, arg):
34 interface.Invoke(1000, 0, pythoncom.DISPATCH_METHOD, 1, arg)
35
36
37# Here is the client side of the connection world.
38# Define a COM object which implements the methods defined by the
39# IConnectDemoEvents interface.
40class ConnectableClient:
41 # This is another cheat - I _know_ the server defines the "DoneIt" event
42 # as DISPID==1000 - I also know from the implementation details of COM
43 # that the first method in _public_methods_ gets 1000.
44 # Normally some explicit DISPID->Method mapping is required.
45 _public_methods_ = ["OnDoneIt"]
46
47 def __init__(self):
48 self.last_event_arg = None
49
50 # A client must implement QI, and respond to a query for the Event interface.
51 # In addition, it must provide a COM object (which server.util.wrap) does.
52 def _query_interface_(self, iid):
53 import win32com.server.util
54
55 # Note that this seems like a necessary hack. I am responding to IID_IConnectDemoEvents
56 # but only creating an IDispatch gateway object.
57 if iid == IID_IConnectDemoEvents:
58 return win32com.server.util.wrap(self)
59
60 # And here is our event method which gets called.
61 def OnDoneIt(self, arg):
62 self.last_event_arg = arg
63
64
65def CheckEvent(server, client, val, verbose):
66 client.last_event_arg = None
67 server.DoIt(val)
68 if client.last_event_arg != val:
69 raise RuntimeError("Sent %r, but got back %r" % (val, client.last_event_arg))
70 if verbose:
71 print("Sent and received %r" % val)
72
73
74# A simple test script for all this.
75# In the real world, it is likely that the code controlling the server
76# will be in the same class as that getting the notifications.
77def test(verbose=0):
78 import win32com.client.dynamic, win32com.client.connect
79 import win32com.server.policy
80
81 server = win32com.client.dynamic.Dispatch(
82 win32com.server.util.wrap(ConnectableServer())
83 )
84 connection = win32com.client.connect.SimpleConnection()
85 client = ConnectableClient()
86 connection.Connect(server, client, IID_IConnectDemoEvents)
87 CheckEvent(server, client, "Hello", verbose)
88 CheckEvent(server, client, str2bytes("Here is a null>\x00<"), verbose)
89 CheckEvent(server, client, "Here is a null>\x00<", verbose)
90 val = "test-\xe0\xf2" # 2 extended characters.
91 CheckEvent(server, client, val, verbose)
92 if verbose:
93 print("Everything seemed to work!")
94 # Aggressive memory leak checking (ie, do nothing!) :-) All should cleanup OK???
95
96
97if __name__ == "__main__":
98 test(1)