Introducción

Tecnología de comunicación en tiempo real en la web: se trata de un mecanismo que permite al servidor enviar actualizaciones de datos al cliente de forma inmediata. Debido a las limitaciones del navegador web tradicional (basado en solicitudes HTTP unidireccionales), se han desarrollado varias técnicas para implementar esta funcionalidad. Estos métodos se clasifican en dos grupos principales: aquellos basados en HTTP (encuesta, encuesta larga, SSE) y aquellos que utilizan protocolos independientes de HTTP (como WebSocket).

Cuando se desarrolla un sistema de encuestas en tiempo real o una aplicación de comunicación instantánea, es necesario mantener la información del front end actualizada sin recargar manualmente la página. Esto implica el uso de técnicas como la encuesta periódica, la encuesta prolongada o la conexión WebSocket.

Encuesta

La encuesta básica consiste en realizar solicitudes al servidor a intervalos regulares (por ejemplo, cada 2 segudnos). Aunque esta técnica es sencilla de implementar, presenta el problema de la latencia, ya que las actualizaciones solo ocurren cuando se completa el ciclo de solicitud.

Ejemplo de implementación:

from flask import Flask, render_template, request, jsonify

app = Flask(__name__)

CANDIDATOS = {
    101: {"nombre": "Shin", "votos": 0},
    102: {"nombre": "Jin", "votos": 0},
    103: {"nombre": "Min", "votos": 0}
}

@app.route("/")
def inicio():
    return render_template("pagina_principal.html", candidatos=CANDIDATOS)

@app.route("/registrar_voto", methods=["POST"])
def registrar_voto():
    identificador = request.json.get("id_candidato")
    CANDIDATOS[identificador]["votos"] += 1
    return "Voto registrado"

@app.route("/obtener_resultados")
def obtener_resultados():
    return jsonify(CANDIDATOS)

if __name__ == '__main__':
    app.run()

servidor_encuesta.py```

{% for clave, info in candidatos.items() %} - {{info.nombre}} ({{info.votos}}) {% endfor %}

pagina_principal.htmlEncuesta prolongada

Para reducir la latencia de la encuesta básica, se puede implementar la encuesta prolongada. En este enfoque, el servidor mantiene la conexión abierta hasta que hayan nuevos datos disponibles o se alcance un tiempo máximo de espera.

Implementación con cola de mensajes:

import queue
from flask import Flask

app = Flask(__name__)
cola_mensajes = queue.Queue()

@app.route("/recibir_datos")
def recibir_datos():
    try:
        mensaje = cola_mensajes.get(timeout=20)
    except queue.Empty:
        mensaje = "Espera finalizada"
    return mensaje

@app.route("/actualizar_datos")
def actualizar_datos():
    cola_mensajes.put("Nuevo valor 10")
    return "Datos actualizados"

ejemplo_cola.pyPara gestionar múltiples conexiones concurrentes:

from flask import Flask, render_template, session, jsonify
import queue
import uuid

app = Flask(__name__)
app.secret_key = "clave_segura"

CANDIDATOS = {
    101: {"nombre": "Shin", "votos": 0},
    102: {"nombre": "Jin", "votos": 0},
    103: {"nombre": "Min", "votos": 0}
}

# Diccionario de colas por sesión
diccionario_colas = {}

@app.route("/")
def inicio():
    identificador_sesion = str(uuid.uuid4())
    session["id_sesion"] = identificador_sesion
    diccionario_colas[identificador_sesion] = queue.Queue()
    return render_template("pagina_prolongada.html", candidatos=CANDIDATOS)

@app.route("/registrar_voto", methods=["POST"])
def registrar_voto():
    identificador = request.json.get("id_candidato")
    CANDIDATOS[identificador]["votos"] += 1
    
    # Notificar a todos los clienets
    for cola in diccionario_colas.values():
        cola.put(CANDIDATOS)
    
    return "Voto registrado"

@app.route("/obtener_datos")
def obtener_datos():
    cola = diccionario_colas[session["id_sesion"]]
    try:
        datos = cola.get(timeout=30)
    except queue.Empty:
        datos = {}
    return jsonify(datos)

servidor_prolongado.py```

{% for clave, info in candidatos.items() %} - {{info.nombre}} ({{info.votos}}) {% endfor %}

pagina_prolongada.htmlWebSocket

WebSocket es un protocolo de comunicación bidireccional que mantiene la conexión abierta entre cliente y servidor. Requiere un handshake inicial y utiliza un canal persistente para el intercambio de datos.

Implementación con Flask y gevent-websocket:

from flask import Flask, render_template, request
from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer
import json

app = Flask(__name__)

CANDIDATOS = {
    101: {"nombre": "Shin", "votos": 0},
    102: {"nombre": "Jin", "votos": 0},
    103: {"nombre": "Min", "votos": 0}
}

clientes_conectados = []

@app.route("/")
def inicio():
    return render_template("pagina_websocket.html", candidatos=CANDIDATOS)

@app.route("/conexion")
def conexion():
    socket = request.environ.get("wsgi.websocket")
    if not socket:
        return "Solicitud HTTP inválida"
    
    clientes_conectados.append(socket)
    
    while True:
        mensaje = socket.receive()
        if not mensaje:
            clientes_conectados.remove(socket)
            break
            
        identificador = int(mensaje)
        CANDIDATOS[identificador]["votos"] += 1
        
        # Notificar a todos los clientes
        datos_actualizados = {
            "id": identificador,
            "nombre": CANDIDATOS[identificador]["nombre"],
            "total": CANDIDATOS[identificador]["votos"]
        }
        
        for cliente in clientes_conectados:
            cliente.send(json.dumps(datos_actualizados))

servidor_websocket.py```

{% for clave, info in candidatos.items() %} - {{info.nombre}} ({{info.votos}}) {% endfor %}

pagina_websocket.html