Kieran/snort
5
14
Fork 14
Code Issues 37 Pull Requests 1 Packages Projects Releases 24 Wiki Activity

fix: connection race

Browse Source
This commit is contained in:
kieran 2024-04-25 13:50:24 +01:00
parent 5af182c1bd
commit 38af05edb8
Signed by: Kieran
GPG Key ID: DE71CEB3925BE941
2 changed files with 14 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
packages/system/src/connection-pool.ts
View File

@ -21,7 +21,7 @@ export interface ConnectionTypeEvents {
unknownMessage: (obj: Array<any>) => void;
}
export interface ConnectionSubscription {}
export interface ConnectionSubscription { }
/**
* Basic relay connection
@ -93,15 +93,14 @@ export type ConnectionBuilder<T extends ConnectionType> = (
address: string,
options: RelaySettings,
ephemeral: boolean,
) => Promise<T>;
) => Promise<T> | T;
/**
* Simple connection pool containing connections to multiple nostr relays
*/
export class DefaultConnectionPool<T extends ConnectionType = Connection>
extends EventEmitter<ConnectionPoolEvents>
implements ConnectionPool
{
implements ConnectionPool {
#system: SystemInterface;
#log = debug("ConnectionPool");
@ -122,7 +121,7 @@ export class DefaultConnectionPool<T extends ConnectionType = Connection>
this.#connectionBuilder = builder;
} else {
this.#connectionBuilder = (addr, options, ephemeral) => {
return Promise.resolve<T>(new Connection(addr, options, ephemeral) as unknown as T);
return new Connection(addr, options, ephemeral) as unknown as T;
};
}
}

11
packages/system/src/connection.ts
View File

@ -43,6 +43,7 @@ export class Connection extends EventEmitter<ConnectionTypeEvents> implements Co
#closing = false;
#downCount = 0;
#activeRequests = new Set<string>();
#connectStarted = false;
id: string;
readonly address: string;
@ -85,6 +86,10 @@ export class Connection extends EventEmitter<ConnectionTypeEvents> implements Co
return this.Socket?.readyState === WebSocket.OPEN;
}
get isConnecting() {
return this.Socket?.readyState === WebSocket.CONNECTING;
}
get isDown() {
return this.#downCount > 0;
}
@ -95,9 +100,12 @@ export class Connection extends EventEmitter<ConnectionTypeEvents> implements Co
async connect() {
// already connected
if (this.isOpen) return;
if (this.isOpen || this.isConnecting) return;
// wait for re-connect timer
if (this.ReconnectTimer) return;
// prevent race condition
if (this.#connectStarted) return;
this.#connectStarted = true;
try {
if (this.info === undefined) {
@ -138,6 +146,7 @@ export class Connection extends EventEmitter<ConnectionTypeEvents> implements Co
this.Socket.onmessage = e => this.#onMessage(e);
this.Socket.onerror = e => this.#onError(e);
this.Socket.onclose = e => this.#onClose(e);
this.#connectStarted = false;
}
close() {