jerry.tian/fal-js
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reconnect token state (#31)

Browse Source
This commit is contained in:
Daniel Rochetti 2023-11-29 08:55:40 -08:00 committed by GitHub
parent 145159a12f
commit c020d97acd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
libs/client/package.json
View File

@ -1,7 +1,7 @@
{ {
"name": "@fal-ai/serverless-client", "name": "@fal-ai/serverless-client",
"description": "The fal serverless JS/TS client", "description": "The fal serverless JS/TS client",
"version": "0.6.0", "version": "0.6.1",
"license": "MIT", "license": "MIT",
"repository": { "repository": {
"type": "git", "type": "git",

29
libs/client/src/realtime.ts
View File

@ -120,6 +120,7 @@ const WebSocketErrorCodes = {
const connectionManager = (() => { const connectionManager = (() => {
const connections = new Map<string, WebSocket>(); const connections = new Map<string, WebSocket>();
const tokens = new Map<string, string>(); const tokens = new Map<string, string>();
const isAuthInProgress = new Map<string, true>();
return { return {
token(app: string) { token(app: string) {
@ -150,10 +151,23 @@ const connectionManager = (() => {
remove(connectionKey: string) { remove(connectionKey: string) {
connections.delete(connectionKey); connections.delete(connectionKey);
}, },
isAuthInProgress(app: string) {
return isAuthInProgress.has(app);
},
setAuthInProgress(app: string, inProgress: boolean) {
if (inProgress) {
isAuthInProgress.set(app, true);
} else {
isAuthInProgress.delete(app);
}
},
}; };
})(); })();
async function getConnection(app: string, key: string): Promise<WebSocket> { async function getConnection(app: string, key: string): Promise<WebSocket> {
if (connectionManager.isAuthInProgress(app)) {
throw new Error('Authentication in progress');
}
const url = buildRealtimeUrl(app); const url = buildRealtimeUrl(app);
if (connectionManager.has(key)) { if (connectionManager.has(key)) {
@ -161,7 +175,9 @@ async function getConnection(app: string, key: string): Promise<WebSocket> {
} }
let token = connectionManager.token(app); let token = connectionManager.token(app);
if (!token) { if (!token) {
connectionManager.setAuthInProgress(app, true);
token = await connectionManager.refreshToken(app); token = await connectionManager.refreshToken(app);
connectionManager.setAuthInProgress(app, false);
} }
const ws = new WebSocket(`${url}?fal_jwt_token=${token}`); const ws = new WebSocket(`${url}?fal_jwt_token=${token}`);
connectionManager.set(key, ws); connectionManager.set(key, ws);
@ -203,7 +219,7 @@ export const realtimeImpl: RealtimeClient = {
return NoOpConnection; return NoOpConnection;
} }
const enqueueMessages: Input[] = []; let pendingMessage: Input | undefined = undefined;
let reconnecting = false; let reconnecting = false;
let ws: WebSocket | null = null; let ws: WebSocket | null = null;
@ -217,7 +233,7 @@ export const realtimeImpl: RealtimeClient = {
}) })
); );
} else { } else {
enqueueMessages.push(input); pendingMessage = input;
if (!reconnecting) { if (!reconnecting) {
reconnecting = true; reconnecting = true;
reconnect(); reconnect();
@ -231,14 +247,17 @@ export const realtimeImpl: RealtimeClient = {
if (ws && ws.readyState === WebSocket.OPEN) { if (ws && ws.readyState === WebSocket.OPEN) {
return; return;
} }
if (connectionManager.isAuthInProgress(app)) {
return;
}
getConnection(app, connectionKey) getConnection(app, connectionKey)
.then((connection) => { .then((connection) => {
ws = connection; ws = connection;
ws.onopen = () => { ws.onopen = () => {
reconnecting = false; reconnecting = false;
if (enqueueMessages.length > 0) { if (pendingMessage) {
enqueueMessages.forEach((input) => send(input)); send(pendingMessage);
enqueueMessages.length = 0; pendingMessage = undefined;
} }
}; };
ws.onclose = (event) => { ws.onclose = (event) => {