129 lines
3.7 KiB
Python
129 lines
3.7 KiB
Python
import uvicorn
|
|
import sys
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
|
from fastapi.responses import HTMLResponse
|
|
from ObjectManager import TaskManager, Task
|
|
from typing import List
|
|
|
|
|
|
'''
|
|
1. 创建一个任务, 则分发任务给worker节点
|
|
2. 客户端建立一个websocket连接, 提供所有的任务的状态更新
|
|
'''
|
|
|
|
|
|
html = """
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<title>Chat</title>
|
|
</head>
|
|
<body>
|
|
<h1>WebSocket Chat</h1>
|
|
<h2>Your ID: <span id="ws-id"></span></h2>
|
|
<form action="" onsubmit="sendMessage(event)">
|
|
<input type="text" id="messageText" autocomplete="off"/>
|
|
<button>Send</button>
|
|
</form>
|
|
<ul id='messages'>
|
|
</ul>
|
|
<script type="module">
|
|
var task = await fetch('/tasks', {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json'
|
|
},
|
|
body: JSON.stringify({
|
|
"name": "test",
|
|
"description": "test",
|
|
"status": "running",
|
|
"created_at": "2020-01-01 00:00:00",
|
|
"updated_at": "2020-01-01 00:00:00"
|
|
})
|
|
}).then(response => response.json())
|
|
console.log(task)
|
|
document.querySelector("#ws-id").textContent = task.id;
|
|
var ws = new WebSocket(`ws://localhost:8000/tasks/${task.id}`);
|
|
ws.onmessage = function(event) {
|
|
var messages = document.getElementById('messages')
|
|
var message = document.createElement('li')
|
|
var content = document.createTextNode(event.data)
|
|
message.appendChild(content)
|
|
messages.appendChild(message)
|
|
};
|
|
ws.onclose = function(event) {
|
|
console.log('Socket is closed. Reconnect will be attempted in 1 second.', event.reason);
|
|
};
|
|
function sendMessage(event) {
|
|
var input = document.getElementById("messageText")
|
|
ws.send(input.value)
|
|
input.value = ''
|
|
event.preventDefault()
|
|
}
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
@app.get("/")
|
|
async def get():
|
|
return HTMLResponse(html)
|
|
|
|
task_manager = TaskManager()
|
|
|
|
|
|
# 通知所有的ws客户端
|
|
@app.post("/notify")
|
|
async def notify():
|
|
pass
|
|
|
|
|
|
@app.get("/tasks", response_model=List[Task])
|
|
async def get_tasks():
|
|
return task_manager.query_all()
|
|
|
|
|
|
@app.post("/tasks", response_model=Task)
|
|
async def create_task(task: Task):
|
|
return task_manager.add(task)
|
|
|
|
|
|
'''
|
|
监听任务进度
|
|
可能有多个客户端监听同一个任务(向任务的观察者列表中添加websocket连接)
|
|
可能有多个任务被同一客户端监听(向客户端的观察目标列表中添加任务)
|
|
应检查目标任务是否存在
|
|
'''
|
|
|
|
|
|
@app.websocket("/tasks/{task_id}")
|
|
async def task_endpoint(websocket: WebSocket, task_id: str):
|
|
await websocket.accept()
|
|
if not task_manager.has_task(task_id):
|
|
await websocket.close()
|
|
print(f"close websocket: {task_id}")
|
|
return
|
|
task_manager.add_observer(task_id, websocket)
|
|
try:
|
|
while True:
|
|
data = await websocket.receive_text()
|
|
print(f"Client #says: {data}")
|
|
except WebSocketDisconnect:
|
|
task_manager.remove_observer(task_id, websocket)
|
|
print(f"close websocket: {task_id}")
|
|
|
|
|
|
'''
|
|
维护一个任务队列, 任务队列中的任务会被分发给worker节点
|
|
任务状态变化时通知对应的客户端
|
|
'''
|
|
|
|
# 启动服务
|
|
if __name__ == '__main__':
|
|
port = 8000 if len(sys.argv) < 2 else int(sys.argv[1])
|
|
uvicorn.run(app='main:app', host='0.0.0.0', port=port, reload=True, workers=1)
|