documented the example and checked functionality

Author: jryannel

README.md

@@ -33,5 +33,5 @@ pip3 install uvicorn
 and run using
 
 ```
-uvicorn demo:app --port=8080
+uvicorn demo_server:app --port=8080
 ```

examples/README.md

@@ -0,0 +1,35 @@
+# ObjectLink Core Python Protocol Example
+
+To get a running client and server we need to hook up the protocol with a websocket connection and pass messages into the protocol and send messages emitted from the protocol.
+
+The app example shows the client side of the protocol with an ObjectLink client.
+
+The server example show the server side of the example with a ObjectLink service.
+
+# App
+
+The app is a websocket client which sends periodically an increment call to the remote endpoint.
+It uses the ObjectLink client to send the increment call to the remote endpoint.
+A websocket adapter adapts the client logic to the websocket protocol.
+The adapter uses the ObjectLink protocol to handle all the message parsing and flow.
+
+- ClientWebSocketAdapter - connects to server and handles the websocket messages using the ObjectLink protocol.
+- CounterSink - is the client implementation as ObjectSink
+
+TODO: might be better to split the sink logic from the client logic. or to define a clean client interface which can be exposed to the outside world
+
+# Server
+
+The server is a websocket async server which handles the websocket connection and the ObjectLink protocol.
+The service handles all the messages from the client and sends back the messages to the client using the ObjectLink protocol.
+A websocket adapter adapts the service logic to the websocket protocol.
+A remote endpoint handles the websocket connection and the communication with the adapter
+
+```
+WebSocketServer -> WebSocketEndpoint -> WebSocketAdapter(ObjectLinkProtocol) -> Service
+```
+
+- `WebSocketServer` handles all incoming connection. It creates a WebSocketEndpoint for each connection.
+- `WebSocketEndpoint` handles the websocket connection and the communication with the adapter.
+- `WebSocketAdapter` handles the ObjectLink protocol and wraps the service implementation.
+- `Service` is the service implementation.

examples/app.py

@@ -5,92 +5,114 @@
 import asyncio
 import websockets
 
-class Counter(IObjectSink):
+
+class CounterSink(IObjectSink):
+    # this is the sink for the counter
     count = 0
     client = None
+
     def __init__(self):
+        # register sink with client node
         self.client = ClientNode.register_sink(self)
         print('client', self.client)
 
     def increment(self):
+        # remote call the increment method
         if self.client:
             self.client.invoke_remote('demo.Counter/increment', [], None)
 
     def olink_object_name(self):
+        # return the name of the sink
         return 'demo.Counter'
 
     def olink_on_signal(self, name: str, args: list[Any]):
+        # handle the incoming signal from the remote source
         path = Name.path_from_name(name)
         print('on signal: %s: %s' % (path, args))
 
     def olink_on_property_changed(self, name: str, value: Any) -> None:
+        # handle the property change from the remote source
         path = Name.path_from_name(name)
         print('on property changed: %s: %s' % (path, value))
 
     def olink_on_init(self, name: str, props: object, node: ClientNode):
+        # handle the initialization of the sink,
+        # called when the sink is linked to remote source
         print('on init: %s: %s' % (name, props))
         self.client = node
-        
+
     def olink_on_release(self):
+        # handle the release of the sink,
+        # called when the sink is unlinked from remote source
         print('on release')
-    
-
-async def sender(ws, q):
-    while True:
-        data = await q.get()
-        await ws.send_text(data)
-        q.task_done()
 
 
-async def connect(address: str):
-    node = ClientNode()
-    queue = Queue()
-    async with websockets.connect(address) as ws:
-        def writer(msg: str):
-            queue.put_nowait(msg)
-        node.on_write(writer)
-        while True:
-            data = await ws.recv()
-            node.handle_message(data)
-
-class Client:
+class ClientWebsocketAdapter:
+    # adapts the websocket communication to the client node
     queue = Queue()
     node = None
+
     def __init__(self, node):
         self.node = node
+        # register a write function
         self.node.on_write(self.writer)
 
     def writer(self, data):
-        print('writer, data')
+        # don't send directly, first write to queue
+        print('write to queue')
         self.queue.put_nowait(data)
 
     async def _reader(self, ws):
+        # handle incoming ws messages
         while True:
-            data = await ws.recv()
-            self.node.handle_message(data)
-    
+            msg = await ws.recv()
+            self.node.handle_message(msg)
+
     async def _sender(self, ws):
+        # send messages from queue
         while True:
             data = await self.queue.get()
             await ws.send(data)
             self.queue.task_done()
 
     async def connect(self, address: str):
-        async with websockets.connect(address) as ws:
-            print('connected')
-            sender_task = asyncio.create_task(self._sender(ws))
-            reader_task = asyncio.create_task(self._reader(ws))
-            await asyncio.gather(sender_task, reader_task)
-            await self.queue.join()
-
+        # connect to websocket server
+        async for ws in websockets.connect(address):
+            # connect might fail so loop continuously, for a retry-connection
+            # see https://websockets.readthedocs.io/en/stable/reference/client.html#opening-a-connection
+            try:
+                sender_task = asyncio.create_task(self._sender(ws))
+                reader_task = asyncio.create_task(self._reader(ws))
+                await asyncio.gather(sender_task, reader_task)
+                await self.queue.join()
+            except Exception as e:
+                print('exception while connecting: ', e)
 
 
 address = 'ws://localhost:8282/ws'
+# create a client node for ObjectLink registry and protocol
 node = ClientNode()
+# link the node to the service name
 node.link_node('demo.Counter')
-client = Client(node)
 
-counter = Counter()
+# create a ws client which handles the ws adapter
+client = ClientWebsocketAdapter(node)
+
+counter = CounterSink()
 node.link_remote('demo.Counter')
 counter.increment()
-asyncio.get_event_loop().run_until_complete(client.connect(address))
+
+
+async def countForever():
+    # every send increment the counter
+    while True:
+        await asyncio.sleep(1)
+        counter.increment()
+
+
+# await both tasks to complete
+future = asyncio.gather(client.connect(address), countForever())
+# get the event loop
+loop = asyncio.get_event_loop()
+# run the event loop until future completes
+loop.run_until_complete(future)

examples/requirements.txt

@@ -0,0 +1,3 @@
+websockets
+starlette
+uvicorn

examples/server.py

@@ -10,81 +10,115 @@
 from olink.remotenode import IObjectSource, RemoteNode
 
 
-class Counter:
+class CounterService:
     count = 0
     _node: RemoteNode
+
     def increment(self):
         self.count += 1
         self._node.notify_property_change('demo.Counter/count', self.count)
 
-class CounterAdapter(IObjectSource):
+
+class CounterWebsocketAdapter(IObjectSource):
+    # adapts the websocket communication to the remote source
     node: RemoteNode = None
+
     def __init__(self, impl):
         self.impl = impl
+        # register the source with the node registry
         RemoteNode.register_source(self)
 
     def olink_object_name(self):
+        # return service name
         return 'demo.Counter'
 
     def olink_invoke(self, name: str, args: list[Any]) -> Any:
+        # handle the remote call from client node
         path = Name.path_from_name(name)
+        # get the function from the implementation
         func = getattr(self.impl, path)
-        func()
+        try:
+            # call function with arguments from the implementation
+            result = func(**args)
+        except Exception as e:
+            # need to have proper exception handling here
+            print('error: %s' % e)
+            result = None
+        # results will be send back to calling client node
+        return result
 
     def olink_set_property(self, name: str, value: Any):
+        # set property value on implementation
         path = Name.path_from_name(name)
         setattr(self, self.impl, value)
 
     def olink_linked(self, name: str, node: "RemoteNode"):
+        # called when the source is linked to a client node
         self.impl._node = node
 
     def olink_collect_properties(self) -> object:
+        # collect properties from implementation to send back to client node initially
         return {k: getattr(self.impl, k) for k in ['count']}
 
-counter = Counter()
-adapter = CounterAdapter(counter)
 
+# create the service implementation
+counter = CounterService()
 
+# create the adapter for the implementation
+adapter = CounterWebsocketAdapter(counter)
 
 
 class RemoteEndpoint(WebSocketEndpoint):
+    # endpoint to handle a client connection
     encoding = "text"
     node = RemoteNode()
+    # message queue
     queue = Queue()
 
     async def sender(self, ws):
+        # sender coroutine, messages from queue are send to client
         print('start sender')
         while True:
-            print('001')
             msg = await self.queue.get()
             print('send', msg)
             await ws.send_text(msg)
-            self.queue.task_done()        
+            self.queue.task_done()
 
     async def on_connect(self, ws: WebSocket):
+        # handle a socket connection
         print('on_connect')
+        # register a sender to the connection
         asyncio.create_task(self.sender(ws))
 
+        # a writer function to queue messages
         def writer(msg: str):
-            print('writer', msg)
+            print('write to queue:', msg)
             self.queue.put_nowait(msg)
+        # register the writer function to the node
         self.node.on_write(writer)
+        # call the super connection handler
         await super().on_connect(ws)
 
-
     async def on_receive(self, ws: WebSocket, data: Any) -> None:
+        # handle a message from a client socket
         print('on_receive', data)
         self.node.handle_message(data)
 
     async def on_disconnect(self, websocket: WebSocket, close_code: int) -> None:
+        # handle a socket disconnect
         await super().on_disconnect(websocket, close_code)
+        # remove the writer from the node
         self.node.on_write(None)
+        # wait for all messages to be send, before closing the connection
         await self.queue.join()
 
 
-
+# see https://www.starlette.io/routing/
 routes = [
     WebSocketRoute("/ws", RemoteEndpoint)
 ]
 
-app = Starlette(routes=routes)
+
+# call with `uvicorn server:app --port 8282`
+# see https://www.starlette.io
+app = Starlette(routes=routes)