- refactored
This commit is contained in:
jens
@@ -0,0 +1,132 @@
|
||||
import asyncio
|
||||
import websockets
|
||||
from threading import Thread
|
||||
from queue import Queue
|
||||
from time import sleep
|
||||
import json
|
||||
from ws.connection import IConnection
|
||||
|
||||
|
||||
class WsClient:
|
||||
def __init__(self, listener: IConnection):
|
||||
self.listener = listener
|
||||
self.loop = asyncio.new_event_loop()
|
||||
self.stop = None
|
||||
self.bg_thread = None
|
||||
|
||||
def bg_run(self, uri):
|
||||
print("bg_run: started")
|
||||
asyncio.set_event_loop(self.loop)
|
||||
fut = asyncio.ensure_future(self.run_client(uri), loop=self.loop)
|
||||
self.loop.run_until_complete(fut)
|
||||
self.stop = None
|
||||
print("bg_run: terminated")
|
||||
|
||||
def connect(self, uri):
|
||||
if self.bg_thread is not None and self.bg_thread.is_alive():
|
||||
return
|
||||
|
||||
self.stop: asyncio.Future[None] = self.loop.create_future()
|
||||
self.bg_thread = Thread(target=self.bg_run, args=(uri,))
|
||||
self.bg_thread.setDaemon(True)
|
||||
self.bg_thread.start()
|
||||
|
||||
def disconnect(self):
|
||||
if self.stop is not None:
|
||||
self.loop.call_soon_threadsafe(self.stop.set_result, None)
|
||||
self.bg_thread.join()
|
||||
|
||||
async def run_client(self, uri):
|
||||
try:
|
||||
websocket = await websockets.connect(uri, loop=self.loop)
|
||||
except Exception as exc:
|
||||
print(f"Failed to connect to {uri}: {exc}.")
|
||||
return
|
||||
|
||||
print("handler: got connection to {}".format(websocket.remote_address))
|
||||
self.listener.on_connect()
|
||||
try:
|
||||
path = "/"
|
||||
consumer_task = asyncio.ensure_future(self.handler_recv(websocket, path), loop=self.loop)
|
||||
producer_task = asyncio.ensure_future(self.handler_send(websocket, path), loop=self.loop)
|
||||
done, pending = await asyncio.wait([consumer_task, producer_task, self.stop], return_when=asyncio.FIRST_COMPLETED,)
|
||||
for task in pending:
|
||||
task.cancel()
|
||||
finally:
|
||||
print("handler: lost connection from {}".format(websocket.remote_address))
|
||||
await websocket.close()
|
||||
self.listener.on_disconnect()
|
||||
|
||||
async def handler_recv(self, websocket, path):
|
||||
while True:
|
||||
try:
|
||||
data = await websocket.recv()
|
||||
await self.listener.on_recv(data)
|
||||
except:
|
||||
print("handler_recv: lost connection from {}".format(websocket.remote_address))
|
||||
return
|
||||
|
||||
async def handler_send(self, websocket, path):
|
||||
while True:
|
||||
data = await self.listener.on_send()
|
||||
if data is not None:
|
||||
try:
|
||||
await websocket.send(json.dumps(data))
|
||||
except:
|
||||
print("handler_send: lost connection from {}".format(websocket.remote_address))
|
||||
break
|
||||
|
||||
|
||||
class PeriodicUpdater(IConnection):
|
||||
def __init__(self):
|
||||
self.ws_client = WsClient(self)
|
||||
|
||||
self.state = Queue()
|
||||
self.thread_cancel = True
|
||||
self.thread = None
|
||||
|
||||
def connect(self, uri):
|
||||
self.ws_client.connect(uri)
|
||||
|
||||
def disconnect(self):
|
||||
self.ws_client.disconnect()
|
||||
|
||||
def on_connect(self):
|
||||
self.thread_cancel = False
|
||||
self.thread = Thread(target=self.run)
|
||||
self.thread.setDaemon(True)
|
||||
self.thread.start()
|
||||
|
||||
def on_disconnect(self):
|
||||
self.thread_cancel = True
|
||||
self.thread.join(1)
|
||||
|
||||
async def on_send(self):
|
||||
return await asyncio.get_event_loop().run_in_executor(None, self.state.get)
|
||||
|
||||
async def on_recv(self, data):
|
||||
print("R: {}".format(data))
|
||||
|
||||
def send(self, data):
|
||||
self.state.put_nowait(data)
|
||||
|
||||
def run(self):
|
||||
count = 0
|
||||
while not self.thread_cancel:
|
||||
self.send({"Value": count})
|
||||
count += 1
|
||||
sleep(0.5)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
ws = PeriodicUpdater()
|
||||
ws.connect(uri="ws://localhost:8765")
|
||||
|
||||
sleep(5)
|
||||
ws.disconnect()
|
||||
sleep(1)
|
||||
ws.connect(uri="ws://localhost:8765")
|
||||
|
||||
while True:
|
||||
sleep(1)
|
||||
@@ -0,0 +1,19 @@
|
||||
import abc
|
||||
|
||||
|
||||
class IConnection:
|
||||
@abc.abstractmethod
|
||||
def on_connect(self):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_disconnect(self):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_recv(self, data):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_send(self):
|
||||
return None
|
||||
@@ -0,0 +1,77 @@
|
||||
import asyncio
|
||||
from ws.connection import IConnection
|
||||
import json
|
||||
|
||||
|
||||
class MsgIo:
|
||||
def __init__(self, key, send):
|
||||
self.key = key
|
||||
self.sender = send
|
||||
self.receiver = None
|
||||
|
||||
def set_recv_handler(self, handler):
|
||||
self.receiver = handler
|
||||
|
||||
async def on_recv(self, data):
|
||||
print("MessageHandler {}".format(data))
|
||||
if self.receiver is not None:
|
||||
await self.receiver(data[self.key])
|
||||
|
||||
async def send(self, data):
|
||||
return await self.sender({self.key: data})
|
||||
|
||||
def can_key(self, key):
|
||||
return self.key == key
|
||||
|
||||
def get_key(self):
|
||||
return self.key
|
||||
|
||||
|
||||
class Value:
|
||||
def __init__(self, initial=None):
|
||||
self.value = initial
|
||||
self.has_changed = True
|
||||
|
||||
def set(self, value):
|
||||
self.has_changed = self.value != value
|
||||
self.value = value
|
||||
|
||||
def is_changed(self):
|
||||
return self.has_changed
|
||||
|
||||
def get(self):
|
||||
self.has_changed = False
|
||||
return self.value
|
||||
|
||||
|
||||
class MessageDispatcher(IConnection):
|
||||
def __init__(self):
|
||||
self.msg_handlers: MsgIo = []
|
||||
self.state = asyncio.Queue()
|
||||
|
||||
def msgio_get(self, key):
|
||||
obj = MsgIo(key, self.send)
|
||||
if key not in self.msg_handlers:
|
||||
self.msg_handlers.append(obj)
|
||||
return obj
|
||||
return None
|
||||
|
||||
def on_connect(self):
|
||||
pass
|
||||
|
||||
def on_disconnect(self):
|
||||
pass
|
||||
|
||||
async def on_recv(self, data):
|
||||
d = json.loads(data)
|
||||
for key_req in d.keys():
|
||||
for h in self.msg_handlers:
|
||||
if h.can_key(key_req):
|
||||
await h.on_recv(d)
|
||||
|
||||
async def on_send(self):
|
||||
return await self.state.get()
|
||||
|
||||
async def send(self, data):
|
||||
await self.state.put(data)
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
import asyncio
|
||||
import websockets
|
||||
import abc
|
||||
|
||||
|
||||
class WsServer:
|
||||
def __init__(self, loop=None):
|
||||
if loop is None:
|
||||
self.loop = asyncio.get_event_loop()
|
||||
else:
|
||||
self.loop = loop
|
||||
|
||||
async def listen(self, host, port):
|
||||
return await websockets.serve(self.handler, host, port, loop=self.loop)
|
||||
|
||||
def disconnect(self):
|
||||
pass
|
||||
|
||||
async def handler(self, websocket, path):
|
||||
print("handler: got connection from {}".format(websocket.remote_address))
|
||||
await self.register(websocket)
|
||||
try:
|
||||
consumer_task = asyncio.ensure_future(self.handler_recv(websocket, path), loop=self.loop)
|
||||
producer_task = asyncio.ensure_future(self.handler_send(websocket, path), loop=self.loop)
|
||||
done, pending = await asyncio.wait([consumer_task, producer_task], return_when=asyncio.FIRST_COMPLETED,)
|
||||
for task in pending:
|
||||
task.cancel()
|
||||
finally:
|
||||
print("handler: lost connection from {}".format(websocket.remote_address))
|
||||
await self.unregister(websocket)
|
||||
|
||||
@abc.abstractmethod
|
||||
async def handler_recv(self, websocket, path):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
async def handler_send(self, websocket, path):
|
||||
pass
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
import asyncio
|
||||
from ws.server.ws_server import WsServer
|
||||
from ws.user import User, UserSet, update
|
||||
import abc
|
||||
from ws.connection import IConnection
|
||||
|
||||
|
||||
class IWsServer:
|
||||
@abc.abstractmethod
|
||||
def on_recv(self, data, loop):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def on_send(self, loop):
|
||||
pass
|
||||
|
||||
|
||||
class WsServerMultiUser(WsServer):
|
||||
def __init__(self, loop=None, listener: IConnection = None):
|
||||
WsServer.__init__(self, loop)
|
||||
self.listener = listener
|
||||
self.USERS = UserSet()
|
||||
self.global_state = {}
|
||||
|
||||
async def notify_state(self, data):
|
||||
if self.USERS: # asyncio.wait doesn't accept an empty list
|
||||
await asyncio.wait([user.send(data) for user in self.USERS])
|
||||
|
||||
async def notify_users(self):
|
||||
if self.USERS: # asyncio.wait doesn't accept an empty list
|
||||
message = {"info": {"type": "users", "count": len(self.USERS)}}
|
||||
await asyncio.wait([user.send(message) for user in self.USERS])
|
||||
|
||||
async def register(self, websocket):
|
||||
usr = User(websocket)
|
||||
usr.path_add("info")
|
||||
self.USERS.add(websocket, usr)
|
||||
await self.notify_users()
|
||||
|
||||
async def unregister(self, websocket):
|
||||
self.USERS.remove(websocket)
|
||||
await self.notify_users()
|
||||
|
||||
async def handler_recv(self, websocket, path):
|
||||
while True:
|
||||
try:
|
||||
data = await websocket.recv()
|
||||
usr = self.USERS.get(websocket)
|
||||
processed = await usr.process(data)
|
||||
if processed:
|
||||
await usr.send(self.global_state)
|
||||
else:
|
||||
await self.listener.on_recv(data)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
break
|
||||
|
||||
async def handler_send(self, websocket, path):
|
||||
while True:
|
||||
data = await self.listener.on_send()
|
||||
try:
|
||||
await self.notify_state(data)
|
||||
# Update global state
|
||||
self.global_state = update(self.global_state, data)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
break
|
||||
+132
@@ -0,0 +1,132 @@
|
||||
import dpath.util as dp
|
||||
import json
|
||||
|
||||
|
||||
def update(d, u, only_existing=True):
|
||||
for k, v in u.items():
|
||||
if isinstance(v, dict):
|
||||
if only_existing:
|
||||
r = update(d.get(k, {}), v)
|
||||
d[k] = r
|
||||
else:
|
||||
d[k] = v
|
||||
else:
|
||||
d[k] = u[k]
|
||||
return d
|
||||
|
||||
|
||||
class User:
|
||||
def __init__(self, socket):
|
||||
self.socket = socket
|
||||
self.paths = []
|
||||
|
||||
async def send(self, data):
|
||||
result = {}
|
||||
for path in self.paths:
|
||||
msg = dp.search(data, path)
|
||||
result.update(msg)
|
||||
if len(result) != 0:
|
||||
await self.socket.send(json.dumps(result))
|
||||
|
||||
async def process(self, data):
|
||||
handled = True
|
||||
try:
|
||||
# Handle message subscription
|
||||
d = json.loads(data)
|
||||
if "+" in d:
|
||||
path = d['+']
|
||||
print("Received subscribe {}".format(path))
|
||||
self.path_add(path)
|
||||
elif "-" in d:
|
||||
path = d['-']
|
||||
print("Received unsubscribe {}".format(path))
|
||||
self.path_remove(path)
|
||||
elif "?" in d:
|
||||
print("Subscriptions: {}".format(self.paths))
|
||||
await self.socket.send(json.dumps({"?": self.paths}))
|
||||
else:
|
||||
handled = False
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
return handled
|
||||
|
||||
def path_add(self, path):
|
||||
if path not in self.paths:
|
||||
self.paths.append(path)
|
||||
|
||||
def path_remove(self, path):
|
||||
if path in self.paths:
|
||||
self.paths.remove(path)
|
||||
|
||||
|
||||
class UserSet:
|
||||
def __init__(self):
|
||||
self.users = {}
|
||||
self.keys = None
|
||||
|
||||
def __len__(self):
|
||||
return len(self.users)
|
||||
|
||||
def clear(self):
|
||||
self.users = {}
|
||||
self.keys = None
|
||||
|
||||
def add(self, key, user):
|
||||
self.users[key] = user
|
||||
|
||||
def remove(self, key):
|
||||
self.users.pop(key, None)
|
||||
|
||||
def get(self, key):
|
||||
return self.users[key]
|
||||
|
||||
def __iter__(self):
|
||||
self.keys = iter(self.users.keys())
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
return self.users[next(self.keys)]
|
||||
|
||||
|
||||
def check_fill(u):
|
||||
if u:
|
||||
print("Is occupied")
|
||||
else:
|
||||
print("Is empty")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
userset = UserSet()
|
||||
check_fill(userset)
|
||||
|
||||
userset.add(1, "A")
|
||||
userset.add(2, "B")
|
||||
userset.add(3, "C")
|
||||
check_fill(userset)
|
||||
|
||||
print("Len(userset) = {}".format(userset))
|
||||
for i in userset:
|
||||
print(i)
|
||||
|
||||
userset.remove(3)
|
||||
for i in userset:
|
||||
print(i)
|
||||
|
||||
userset.remove(2)
|
||||
for i in userset:
|
||||
print(i)
|
||||
|
||||
userset.remove(1)
|
||||
for i in userset:
|
||||
print(i)
|
||||
|
||||
a = {'Pot': {'Temp': 15.5}}
|
||||
b = {'Pot': {'Power': 2400}}
|
||||
|
||||
print(a)
|
||||
print(b)
|
||||
|
||||
a = update(a, b)
|
||||
|
||||
print("a | b = {}".format(a))
|
||||
Reference in New Issue
Block a user