ypy-websocket

import asyncio import uvicorn from y_py import YDoc from ypy_websocket.server import YpyWebSocketServer from ypy_websocket.yroom import YRoom from ypy_websocket.ystore import YStore async def main(): # Initialize a YDoc ydoc = YDoc() # Apply an initial update, e.g., create a text type with ydoc.begin_transaction() as txn: ydoc.get_text('my_text').insert(txn, 0, 'Hello Y-CRDT!') # Create a YStore and YRoom to manage the YDoc ystore = YStore() yroom = YRoom(ydoc, ystore=ystore) # The server will handle rooms implicitly; here we just show direct room creation # Create the WebSocket server # In a real app, 'get_room' would likely fetch a room based on path async def get_room(path: str) -> YRoom: if path == '/ws/doc': return yroom raise ValueError('Room not found') websocket_server = YpyWebSocketServer(get_room=get_room) # This part typically runs in a separate process (e.g., via `uvicorn`) # but we demonstrate the ASGI app creation. # For a quick runnable demo, we'll start it directly if possible (not ideal for actual deployment) # To run this, you would typically save it as `app.py` and run `uvicorn app:websocket_server` print("Ypy WebSocket server initialized. Connect to ws://127.0.0.1:8000/ws/doc") # A full quickstart would involve running uvicorn # For a runnable quickstart here, we'll demonstrate just the setup. # To make it runnable via `python your_script.py`, it would require more extensive setup: # from uvicorn.config import Config # from uvicorn.server import Server # config = Config(websocket_server, host="0.0.0.0", port=8000, ws="wsproto") # server = Server(config) # await server.serve() # For a simple runnable example that demonstrates the setup without blocking: # In a real scenario, this would be exposed by an ASGI server like Uvicorn. # Let's mock a client connection setup without actually serving. print("To run the server, save the above 'websocket_server' variable as 'app.py' and execute:") print("`uvicorn app:websocket_server --host 127.0.0.1 --port 8000`") print("Then connect with a WebSocket client to `ws://127.0.0.1:8000/ws/doc`") if __name__ == '__main__': asyncio.run(main())