Merge pull request #76 from y-crdt/websocket-example

Websockets Drawing Example

examples/drawing/README.md

@@ -0,0 +1,19 @@
+# Collaborative Drawing
+
+![drawing-demo](https://user-images.githubusercontent.com/5553757/189190756-21c2bc61-1816-488e-b2cd-bc910fece6d9.gif)
+
+A basic collaborative drawing application using Ypy and WebSockets. Left click on the canvas to leave a mark.
+
+## Getting Started
+
+1. Install Python dependencies:
+
+```
+pip install -r requirements.txt
+```
+
+2. Run the demo
+
+```
+python demo.py
+```

examples/drawing/client.py

@@ -0,0 +1,61 @@
+import asyncio
+from time import sleep
+import websockets
+import y_py as Y
+import queue
+import concurrent.futures
+import threading
+
+# Code based on the [`websockets` patter documentation](https://websockets.readthedocs.io/en/stable/howto/patterns.html)
+
+class YDocWSClient:
+
+    def __init__(self, uri = "ws://localhost:7654"):
+        self.send_q = queue.Queue()
+        self.recv_q = queue.Queue()
+        self.uri = uri
+        def async_loop():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+
+            loop.run_until_complete(self.start_ws_client())
+            loop.close()
+        ws_thread = threading.Thread(target=async_loop, daemon=True)
+        ws_thread.start()
+
+    def send_updates(self, txn_event: Y.AfterTransactionEvent):
+        update = txn_event.get_update()
+        # Sometimes transactions don't write, which means updates are empty.
+        # We only care about updates with meaningful mutations.
+        if update != b'\x00\x00':
+            self.send_q.put_nowait(update)
+
+    def apply_updates(self, doc: Y.YDoc):
+        while not self.recv_q.empty():
+            update = self.recv_q.get_nowait()
+            Y.apply_update(doc, update)
+
+    async def client_handler(self, websocket):
+        consumer_task = asyncio.create_task(self.consumer_handler(websocket))
+        producer_task = asyncio.create_task(self.producer_handler(websocket))
+        done, pending = await asyncio.wait(
+            [consumer_task, producer_task],
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+        for task in pending:
+            task.cancel()
+
+    async def consumer_handler(self, websocket):
+        async for message in websocket:
+            self.recv_q.put_nowait(message)
+
+    async def producer_handler(self, websocket):
+        loop = asyncio.get_running_loop()
+        while True:
+            update = await loop.run_in_executor(None,self.send_q.get)
+            await websocket.send(update)
+
+    async def start_ws_client(self):
+        async with websockets.connect(self.uri) as websocket:
+            await self.client_handler(websocket)
+

examples/drawing/demo.py

@@ -0,0 +1,33 @@
+import subprocess
+from typing import List
+
+
+def demo():
+    """
+    Spawns a server and two drawing clients.
+    """
+    processes: List[subprocess.Popen]  = []
+    # Server
+    processes.append(subprocess.Popen(["python", "server.py"]))
+
+    # Clients
+    for _ in range(2):
+        processes.append(subprocess.Popen(["python", "draw.py"]))
+
+
+    wait_until_done()
+
+    for p in processes:
+        p.kill()
+
+
+
+def wait_until_done():
+    print("waiting")
+    while input("Enter 'q' to quit: ").lower() != 'q':
+        continue
+
+
+
+if __name__ == "__main__":
+    demo()

examples/drawing/draw.py

@@ -0,0 +1,45 @@
+from turtle import position
+from p5 import *
+from y_py import YDoc, YArray, AfterTransactionEvent
+from client import YDocWSClient
+
+doc: YDoc
+strokes: YArray
+client: YDocWSClient
+
+
+def setup():
+    """
+    Initialization logic that runs before the `draw()` loop.
+    """
+    global strokes
+    global doc
+    global client
+    title("Ypy Drawing Demo")
+    size(720, 480)
+    doc = YDoc(0)
+    strokes = doc.get_array("strokes")
+    client = YDocWSClient()
+    doc.observe_after_transaction(client.send_updates)
+
+
+
+def draw():
+    """
+    Handles user input and updates the canvas.
+    """
+    global strokes
+    global doc
+    global client
+    client.apply_updates(doc)
+    rect_mode(CENTER)
+    background(255)
+    if mouse_is_pressed:
+        with doc.begin_transaction() as txn:
+            strokes.append(txn, [mouse_x, mouse_y])
+    fill(0)
+    no_stroke()
+    for x,y in strokes:
+        ellipse((x, y), 33, 33)
+
+run(frame_rate=60)

examples/drawing/requirements.txt

@@ -0,0 +1,6 @@
+y_py
+websockets
+glfw; sys_platform != 'win32'
+numpy
+vispy
+p5

examples/drawing/server.py

@@ -0,0 +1,27 @@
+import asyncio
+from turtle import update
+import websockets
+
+connected = set()
+
+async def server_handler(websocket):
+    # Register.
+    connected.add(websocket)
+    try:
+        async for message in websocket:
+            peers = {peer for peer in connected if peer is not websocket}
+            websockets.broadcast(peers, message)
+
+    except websockets.exceptions.ConnectionClosedError: 
+        pass
+    finally:
+        # Unregister.
+        connected.remove(websocket)
+
+
+async def main():
+    async with websockets.serve(server_handler, "localhost", 7654):
+        await asyncio.Future()  # run forever
+
+if __name__ == "__main__":
+    asyncio.run(main())