changed connection model
This commit is contained in:
parent
adbb1fe80a
commit
10cefaf1cd
|
|
@ -913,10 +913,22 @@ class ARIABridge:
|
|||
retry_delay = min(retry_delay * 2, 30)
|
||||
|
||||
async def _rvs_heartbeat(self) -> None:
|
||||
"""Sendet Heartbeats an den RVS damit die Verbindung offen bleibt."""
|
||||
"""Sendet Heartbeats + WebSocket Pings an den RVS damit die Verbindung offen bleibt."""
|
||||
while True:
|
||||
await asyncio.sleep(25)
|
||||
await asyncio.sleep(15)
|
||||
if self.ws_rvs:
|
||||
try:
|
||||
# WebSocket Protocol-Level Ping (haelt TCP-Verbindung am Leben)
|
||||
pong = await self.ws_rvs.ping()
|
||||
await asyncio.wait_for(pong, timeout=10)
|
||||
except Exception:
|
||||
logger.warning("[rvs] Ping fehlgeschlagen — Verbindung tot, erzwinge Reconnect")
|
||||
try:
|
||||
await self.ws_rvs.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ws_rvs = None
|
||||
break
|
||||
try:
|
||||
await self.ws_rvs.send(json.dumps({
|
||||
"type": "heartbeat",
|
||||
|
|
|
|||
|
|
@ -562,54 +562,22 @@ function sendToRVS_raw(msgObj) {
|
|||
}
|
||||
|
||||
function sendToRVS(text, isPipeline) {
|
||||
if (!RVS_HOST || !RVS_TOKEN) {
|
||||
log("error", "rvs", "Nicht konfiguriert");
|
||||
if (isPipeline) pipelineEnd(false, "RVS nicht konfiguriert");
|
||||
return false;
|
||||
}
|
||||
// Ueber Gateway senden (zuverlaessig) UND an RVS fuer App-Sichtbarkeit
|
||||
// Die Bridge empfaengt RVS-Nachrichten von der App zuverlaessig,
|
||||
// aber die Diagnostic→RVS→Bridge Route hat Zombie-Probleme.
|
||||
// Deshalb: Gateway fuer ARIA, RVS nur fuer App-Anzeige.
|
||||
|
||||
// Frische WebSocket-Verbindung fuer jede Nachricht (Zombie-Schutz)
|
||||
const proto = RVS_TLS === "true" ? "wss" : "ws";
|
||||
const url = `${proto}://${RVS_HOST}:${RVS_PORT}?token=${RVS_TOKEN}`;
|
||||
const msg = JSON.stringify({
|
||||
// 1. An Gateway senden (damit ARIA antwortet)
|
||||
const gatewayOk = sendToGateway(text, isPipeline);
|
||||
|
||||
// 2. An RVS senden (damit die App die Nachricht sieht)
|
||||
sendToRVS_raw({
|
||||
type: "chat",
|
||||
payload: { text, sender: "diagnostic" },
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
|
||||
log("info", "rvs", `Sende via frische Verbindung: ${url.split('?')[0]}`);
|
||||
|
||||
const freshWs = new WebSocket(url);
|
||||
freshWs.on("open", () => {
|
||||
freshWs.send(msg);
|
||||
log("info", "rvs", `Gesendet via RVS: "${text}"`);
|
||||
// Verbindung offen lassen fuer Antwort-Empfang, nach 5min schliessen
|
||||
setTimeout(() => { try { freshWs.close(); } catch (_) {} }, 300000);
|
||||
});
|
||||
freshWs.on("message", (raw) => {
|
||||
try {
|
||||
const resp = JSON.parse(raw.toString());
|
||||
if (resp.type === "chat" && resp.payload) {
|
||||
const sender = resp.payload.sender || "?";
|
||||
// Eigene Nachrichten und STT ignorieren (werden von persistenter Verbindung gehandelt)
|
||||
if (sender === "diagnostic" || sender === "stt") return;
|
||||
log("info", "rvs", `Chat von ${sender}: "${(resp.payload.text || "").slice(0, 100)}"`);
|
||||
if (pipelineActive && sender !== "diagnostic") {
|
||||
pipelineEnd(true, `Antwort via RVS von ${sender}: "${(resp.payload.text || "").slice(0, 120)}"`);
|
||||
}
|
||||
broadcast({ type: "rvs_chat", msg: resp });
|
||||
} else if (resp.type !== "heartbeat") {
|
||||
log("debug", "rvs", `Nachricht: ${JSON.stringify(resp).slice(0, 150)}`);
|
||||
}
|
||||
} catch {}
|
||||
});
|
||||
freshWs.on("error", (err) => {
|
||||
log("error", "rvs", `Sende-Fehler: ${err.message}`);
|
||||
if (isPipeline) pipelineEnd(false, `RVS Fehler: ${err.message}`);
|
||||
});
|
||||
|
||||
if (isPipeline) plog(`Nachricht an RVS gesendet — warte auf Antwort via RVS...`);
|
||||
return true;
|
||||
return gatewayOk;
|
||||
}
|
||||
|
||||
// ── Claude Proxy Test ────────────────────────────────────
|
||||
|
|
|
|||
Loading…
Reference in New Issue