import { connect, StringCodec } from "nats.ws"; import { writable } from "svelte/store"; import msghandlerCSR from './msghandler-csr.js'; export const connectionStatus = writable(null); let config = null; export function setConfig(cfg) { config = cfg; } connectionStatus.subscribe(status => { console.log('[NATS] Status updated:', status); }); let nc = null; let sc = StringCodec(); let subs = new Map(); let reconnectTimer = null; let isDisconnecting = false; export async function getNats(opts = {}) { const servers = opts.servers || (import.meta.env.VITE_NATS_URL) || (config?.nats_server_info?.url) || "wss://natsws.yiem.cc"; console.log('[NATS] Connecting to:', servers); clearTimeout(reconnectTimer); reconnectTimer = null; isDisconnecting = false; try { connectionStatus.set({ state: "connecting", message: `connecting to ${servers}` }); console.log('[NATS] Status: connecting'); nc = await connect({ servers }); console.log('[NATS] Connected successfully!'); connectionStatus.set({ state: "connected", message: `connected to ${servers}` }); isDisconnecting = false; // Track connection state changes for status dot // nats.ws emits status events for disconnect/reconnect const statusSub = nc.status(); (async () => { for await (const status of statusSub) { console.log('[NATS] Status event:', status.type, status.data); if (status.type === 'disconnect') { connectionStatus.set({ state: 'disconnected', message: 'connection lost' }); } else if (status.type === 'reconnect') { connectionStatus.set({ state: 'connected', message: 'reconnected' }); } else if (status.type === 'connect') { connectionStatus.set({ state: 'connected', message: `connected to ${servers}` }); } } })(); // Also monitor permanent closure nc.closed().then((err) => { if (isDisconnecting) return; if (err) { console.log('[NATS] Connection permanently closed with error:', err); connectionStatus.set({ state: "error", message: String(err) }); } else { console.log('[NATS] Connection permanently closed normally'); connectionStatus.set({ state: "disconnected", message: "connection closed" }); } }); return nc; } catch (e) { console.error('[NATS] Connection failed:', e); connectionStatus.set({ state: "error", message: String(e) }); // nats.ws will automatically attempt reconnect when we call getNats() again throw e; } } export async function subscribe(subject, onMessage) { console.log('[NATS] Subscribing to:', subject); if (!nc) await getNats(); const sub = nc.subscribe(subject); subs.set(subject, sub); console.log('[NATS] Subscription created for:', subject); (async () => { for await (const m of sub) { let text = ""; try { text = sc.decode(m.data); let parsed = {}; try { parsed = JSON.parse(text); } catch (e) { parsed = { text }; } console.log('[NATS] Received message on', subject, ':', parsed); onMessage(parsed); } catch (e) { console.warn("nats message parse error", e); } } })(); return { unsubscribe: async () => { console.log('[NATS] Unsubscribing from:', subject); try { sub.unsubscribe(); } catch (e) {} subs.delete(subject); } }; } export async function publish(subject, payload = {}, opts = {}) { console.log('[NATS] Publishing to subject:', subject); console.log('[NATS] Payload type:', typeof payload); console.log('[NATS] Payload (as JSON):', typeof payload === 'string' ? payload : JSON.stringify(payload, null, 2)); if (!nc) { console.log('[NATS] No connection, attempting to connect...'); await getNats(opts); } const msg = typeof payload === "string" ? payload : JSON.stringify(payload); console.log('[NATS] Final message to publish (first 300 chars):', msg.substring(0, 300) + (msg.length > 300 ? '...' : '')); console.log('[NATS] Message length:', msg.length); console.log('[NATS] NATS connection exists:', !!nc); console.log('[NATS] About to publish...'); nc.publish(subject, sc.encode(msg)); console.log('[NATS] Message published successfully to:', subject); } export async function closeNats() { console.log('[NATS] Closing connection...'); clearTimeout(reconnectTimer); reconnectTimer = null; isDisconnecting = true; connectionStatus.set({ state: "disconnected", message: "closing" }); try { for (const s of subs.values()) { try { s.unsubscribe(); } catch (e) {} } subs.clear(); if (nc) { await nc.close(); nc = null; } console.log('[NATS] Connection closed'); connectionStatus.set({ state: "disconnected", message: "closed" }); } catch (e) { console.error('[NATS] Error closing connection:', e); connectionStatus.set({ state: "error", message: String(e) }); } } export async function connectNATS(brokerUrl, cfg) { console.log('[NATS] Connecting with broker URL:', brokerUrl); let serverUrl = brokerUrl; if (!serverUrl && cfg) { serverUrl = cfg?.nats_server_info?.url || 'wss://natsws.yiem.cc'; } console.log('[NATS] Using server URL:', serverUrl); setConfig(cfg); await getNats({ servers: serverUrl }); } export async function request(subject, payload, opts = {}) { console.log('[NATS] Requesting from:', subject, 'payload:', payload); const { timeout, ...getNatsOpts } = opts; if (!nc) await getNats(getNatsOpts); const msg = typeof payload === "string" ? payload : JSON.stringify(payload); const requestOpts = timeout ? { timeout } : {}; const response = await nc.request(subject, sc.encode(msg), requestOpts); const responseText = sc.decode(response.data); let responseJson; try { responseJson = JSON.parse(responseText); } catch (e) { responseJson = { text: responseText }; } console.log('[NATS] Request response:', responseJson); return responseJson; } export async function reply(subject, handler, opts = {}) { console.log('[NATS] Setting up reply handler on:', subject); if (!nc) await getNats(opts); const sub = nc.subscribe(subject); subs.set(subject, sub); (async () => { for await (const m of sub) { let text = ""; try { text = sc.decode(m.data); let parsed = {}; try { parsed = JSON.parse(text); } catch (e) { parsed = { text }; } console.log('[NATS] Received request on', subject, ':', parsed); const response = handler(parsed, m); const responseMsg = typeof response === "string" ? response : JSON.stringify(response); nc.publish(m.subject, sc.encode(responseMsg)); console.log('[NATS] Reply sent to:', m.subject); } catch (e) { console.warn("nats reply handler error", e); } } })(); return { unsubscribe: async () => { console.log('[NATS] Unsubscribing from:', subject); try { sub.unsubscribe(); } catch (e) {} subs.delete(subject); } }; } export async function sendMsghandlerRequest(subject, functioncall, args, config) { console.log('[NATS] Sending msghandler request:', functioncall); const payload = { functioncall: functioncall, args: args }; const brokerUrl = config?.nats_server_info?.url || 'wss://natsws.yiem.cc'; const fileserverUrl = config?.file_server_url || 'https://fileserver.yiem.cc'; const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack( subject, [["db_request", payload, "dictionary"]], { broker_url: brokerUrl, fileserver_url: fileserverUrl } ); console.log('[NATS] Request envelope created, sending...'); const natsReply = await request(subject, envelopeJsonStr); console.log('[NATS] Response received, unpacking...'); const unpackedEnvelope = await msghandlerCSR.smartunpack(natsReply); console.log('-----'); console.log(unpackedEnvelope); console.log('===== 1'); for (const [dataname, data, type] of unpackedEnvelope.payloads) { console.log('[NATS] Extracted payload:', dataname, type); if (type === "jsontable" || type === "dictionary") { console.log('[NATS] Request successful:', data); return data; } if (type === "text") { console.error('[NATS] Error response:', data); throw new Error(data); } } throw new Error('No valid response found'); }