| | import uvicorn |
| | from fastapi import FastAPI, WebSocket, WebSocketDisconnect |
| | from fastapi.responses import HTMLResponse |
| | from typing import List |
| |
|
| | app = FastAPI() |
| |
|
| | |
| | class ConnectionManager: |
| | def __init__(self): |
| | |
| | self.macos_connections: List[WebSocket] = [] |
| | self.linux_connections: List[WebSocket] = [] |
| |
|
| | async def connect_macos(self, websocket: WebSocket): |
| | await websocket.accept() |
| | self.macos_connections.append(websocket) |
| | print(f"Client connected to macOS. Total: {len(self.macos_connections)}") |
| |
|
| | async def connect_linux(self, websocket: WebSocket): |
| | await websocket.accept() |
| | self.linux_connections.append(websocket) |
| | print(f"Client connected to Linux. Total: {len(self.linux_connections)}") |
| |
|
| | def disconnect_macos(self, websocket: WebSocket): |
| | if websocket in self.macos_connections: |
| | self.macos_connections.remove(websocket) |
| | print("Client disconnected from macOS") |
| |
|
| | def disconnect_linux(self, websocket: WebSocket): |
| | if websocket in self.linux_connections: |
| | self.linux_connections.remove(websocket) |
| | print("Client disconnected from Linux") |
| |
|
| | async def forward_to_linux_clients(self, message: str): |
| | |
| | |
| | for connection in self.linux_connections[:]: |
| | try: |
| | await connection.send_text(message) |
| | except Exception as e: |
| | print(f"Error sending to linux client: {e}") |
| |
|
| | async def forward_to_macos_clients(self, message: str): |
| | |
| | for connection in self.macos_connections[:]: |
| | try: |
| | await connection.send_text(message) |
| | except Exception as e: |
| | print(f"Error sending to macos client: {e}") |
| |
|
| | manager = ConnectionManager() |
| |
|
| | |
| |
|
| | @app.websocket("/macos") |
| | async def websocket_macos(websocket: WebSocket): |
| | await manager.connect_macos(websocket) |
| | try: |
| | while True: |
| | |
| | data = await websocket.receive_text() |
| | print(f"Received from macOS: {data}") |
| | |
| | await manager.forward_to_linux_clients(data) |
| | except WebSocketDisconnect: |
| | manager.disconnect_macos(websocket) |
| | except Exception as e: |
| | print(f"Error in macOS socket: {e}") |
| | manager.disconnect_macos(websocket) |
| |
|
| | @app.websocket("/linux") |
| | async def websocket_linux(websocket: WebSocket): |
| | await manager.connect_linux(websocket) |
| | try: |
| | while True: |
| | |
| | data = await websocket.receive_text() |
| | print(f"Received from Linux: {data}") |
| | |
| | await manager.forward_to_macos_clients(data) |
| | except WebSocketDisconnect: |
| | manager.disconnect_linux(websocket) |
| | except Exception as e: |
| | print(f"Error in Linux socket: {e}") |
| | manager.disconnect_linux(websocket) |
| |
|
| | |
| | @app.get("/") |
| | async def get(): |
| | return HTMLResponse(""" |
| | <!DOCTYPE html> |
| | <html> |
| | <head> |
| | <title>Bridge Tester</title> |
| | </head> |
| | <body> |
| | <h1>OS Bridge Tester</h1> |
| | <div style="display:flex; gap: 20px;"> |
| | <div style="border:1px solid #ccc; padding:10px;"> |
| | <h2>Pretend to be macOS</h2> |
| | <button onclick="connect('macos')">Connect to /macos</button> |
| | <input type="text" id="msgMac" placeholder="Message to Linux" /> |
| | <button onclick="send('macos')">Send</button> |
| | <ul id="logMac"></ul> |
| | </div> |
| | |
| | <div style="border:1px solid #ccc; padding:10px;"> |
| | <h2>Pretend to be Linux</h2> |
| | <button onclick="connect('linux')">Connect to /linux</button> |
| | <input type="text" id="msgLin" placeholder="Message to macOS" /> |
| | <button onclick="send('linux')">Send</button> |
| | <ul id="logLin"></ul> |
| | </div> |
| | </div> |
| | |
| | <script> |
| | var wsMac = null; |
| | var wsLin = null; |
| | |
| | function connect(os) { |
| | var ws = new WebSocket(((window.location.protocol === "https:") ? "wss://" : "ws://") + window.location.host + "/" + os); |
| | |
| | ws.onopen = function() { appendLog(os, "Connected!"); }; |
| | ws.onmessage = function(event) { appendLog(os, "Received: " + event.data); }; |
| | |
| | if(os === 'macos') wsMac = ws; |
| | else wsLin = ws; |
| | } |
| | |
| | function send(os) { |
| | var input = document.getElementById(os === 'macos' ? 'msgMac' : 'msgLin'); |
| | var ws = os === 'macos' ? wsMac : wsLin; |
| | if(ws) ws.send(input.value); |
| | input.value = ''; |
| | } |
| | |
| | function appendLog(os, msg) { |
| | var ul = document.getElementById(os === 'macos' ? 'logMac' : 'logLin'); |
| | var li = document.createElement("li"); |
| | li.appendChild(document.createTextNode(msg)); |
| | ul.appendChild(li); |
| | } |
| | </script> |
| | </body> |
| | </html> |
| | """) |
| |
|
| | |
| | if __name__ == "__main__": |
| | uvicorn.run(app, host="0.0.0.0", port=7860) |
