WebSocket
=========
Comunicação em tempo real bidirecional.
Requer modo ASGI:
.. code-block:: bash
pip install velox-web[asgi]
---
Uso Básico
-----------
.. code-block:: python
from velox import Velox
app = Velox(__name__)
@app.websocket('/ws/chat')
async def chat(ws):
# Não precisa de ws.accept() - o Velox faz automaticamente!
while True:
msg = await ws.receive()
if msg is None:
break
await ws.send(f'echo: {msg}')
# Opções de execução (escolha UMA):
# uvicorn app:app --host localhost --port 8001 --reload
# python app.py (via app.run)
if __name__ == '__main__':
app.run(asgi=True, port=8001)
Executar (escolha UMA):
# Opção 1: Via Python
python app.py
# Opção 2: Via uvicorn (recomendado em produção)
uvicorn app:app --host localhost --port 8001 --reload
Client JavaScript:
.. code-block:: javascript
const ws = new WebSocket('ws://localhost:8001/ws/chat');
ws.onopen = () => {
// Importante: esperar um pouco antes de enviar
setTimeout(() => ws.send('Olá'), 500);
};
ws.onmessage = (e) => console.log('Recebeu:', e.data);
ws.onclose = (e) => console.log('Fechado:', e.code);
**Nota:** O ``await ws.accept()`` não é necessário - o Velox já faz isso automaticamente
antes de chamar o handler.
---
Página de Teste HTML
---------------------
Crie um arquivo ``websocket-test.html`` para testar:
.. code-block:: html
WebSocket Test
🔌 WebSocket Test
---
WebSocketManager
----------------
Gerencie conexões, rooms e broadcast:
.. code-block:: python
from velox.websocket import WebSocketManager
manager = WebSocketManager()
@app.websocket('/ws/chat')
async def chat(ws):
await manager.connect(ws, room='chat')
try:
while True:
msg = await ws.receive()
if msg is None:
break
# Broadcast para sala
await manager.broadcast({'msg': msg}, room='chat')
finally:
manager.disconnect(ws)
---
Rooms/Canais
------------
Separe conexões em salas diferentes:
.. code-block:: python
@app.websocket('/ws/')
async def room_chat(ws, room):
await manager.connect(ws, room=room)
try:
while True:
msg = await ws.receive()
if msg is None:
break
# Enviar para sala específica
await manager.send_to_room({
'type': 'message',
'data': msg,
'room': room
}, room=room)
finally:
manager.disconnect(ws)
---
Enviar para Usuário Específico
-------------------------------
.. code-block:: python
# Enviar mensagem para usuário específico
await manager.send_to_user(
{'notification': 'Nova mensagem'},
user_id='user123'
)
---
Callbacks de Eventos
--------------------
.. code-block:: python
@manager.on_connect
async def on_connected(ws, info):
print(f'Cliente conectado na sala: {info["room"]}')
@manager.on_disconnect
async def on_disconnected(ws, info):
print(f'Cliente desconectado')
@manager.on_message
async def on_message(ws, msg_type, data):
print(f'{msg_type}: {data}')
---
Broadcast Global
----------------
.. code-block:: python
# Enviar para TODOS os clientes conectados
await manager.broadcast({
'type': 'announcement',
'data': 'Manutenção programada para amanhã'
})
---
Estatísticas
------------
.. code-block:: python
stats = manager.get_stats()
print(f"Conexões: {stats['connections']}")
print(f"Rooms: {stats['rooms']}")
print(f"Detalhes: {stats['room_details']}")
---
WebSocketHandler
-----------------
Helper para processar mensagens estruturadas:
.. code-block:: python
from velox.websocket import WebSocketHandler
handler = WebSocketHandler(manager)
@handler.register('chat_message')
async def handle_chat(ws, data):
await manager.send_to_room({
'type': 'chat',
'from': ws.user_id,
'message': data
}, room=ws.room)
@handler.register('typing')
async def handle_typing(ws, data):
# Notificar outros na sala
pass
---
Configuração
------------
.. code-block:: text
# Tamanho máximo da mensagem (1MB padrão)
WS_MAX_MESSAGE_SIZE=1048576
---
Singleton
---------
Use o gerenciador padrão:
.. code-block:: python
from velox.websocket import get_manager
manager = get_manager() # retorna instância única
---
Troubleshooting
---------------
Se a conexão fechar imediatamente:
1. **Use delay no cliente** - O ``ws.send()`` deve esperar após ``onopen``:
.. code-block:: javascript
ws.onopen = () => setTimeout(() => ws.send('msg'), 500);
2. **Verifique o firewall** - Some firewalls bloqueiam WebSocket
3. **Use HTTPS em produção** - Navegadores exigem WSS (WebSocket Secure) em HTTPS
4. **Verifique logs do servidor** - O Velox mostra logs de conexão no terminal
---
Comparação com Flask e Django
-----------------------------
O Velox é mais simples que frameworks tradicionais para WebSocket:
+------------------+-------------------+-------------------+-------------------+
| | Velox | Flask-SocketIO | Django Channels |
+==================+===================+===================+===================+
| Dependências | 0 (built-in) | flask-socketio | channels + daphne |
+------------------+-------------------+-------------------+-------------------+
| Linhas de código | ~5 | ~10 | ~30+ |
+------------------+-------------------+-------------------+-------------------+
| Configuração | Nenhuma | Mínima | Extensa |
+------------------+-------------------+-------------------+-------------------+
| Rooms/Broadcast | Built-in | Built-in | Precisa config |
+------------------+-------------------+-------------------+-------------------+
**Flask com flask-socketio:**
.. code-block:: python
from flask import Flask
from flask_socketio import SocketIO, emit
app = Flask(__name__)
socketio = SocketIO(app)
@socketio.on('message')
def handle_message(data):
emit('response', {'data': f'echo: {data}'})
socketio.run(app)
**Django com Channels:**
.. code-block:: python
# consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def receive(self, text_data):
await self.send(text_data=json.dumps({'response': f'echo: {text_data}'}))
# routing.py
from django.urls import path
from . import consumers
websocket_urlpatterns = [path('ws/chat/', consumers.ChatConsumer.as_asgi())]
**Velox (mesmo resultado, muito mais simples):**
.. code-block:: python
from velox import Velox
app = Velox(__name__)
@app.websocket('/ws/chat')
async def chat(ws):
while True:
msg = await ws.receive()
if msg is None: break
await ws.send(f'echo: {msg}')
app.run(asgi=True)
**Resultado:** O Velox faz em 5 linhas o que outros frameworks fazem em 10-30+ linhas,
sem nenhuma dependência extra.