Skip to content

Allow custom app logic to trigger when a websocket connects/reconnects #75

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 21, 2023
Merged

Allow custom app logic to trigger when a websocket connects/reconnects #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b5c4fc4
Allow custom app logic to trigger when a websocket connects/reconnects
peterbroadhurst Aug 10, 2023
029e343
Add option to wait for complete cleanup of the websocket on close
peterbroadhurst Aug 11, 2023
e396f5d
Resolve the close promise
peterbroadhurst Aug 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions lib/firefly.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ import {
FireFlyDeleteOptions,
FireFlyTokenApprovalFilter,
FireFlyTokenApprovalResponse,
FireFlyWebSocketConnectCallback,
} from './interfaces';
import { FireFlyWebSocket, FireFlyWebSocketCallback } from './websocket';
import HttpBase, { mapConfig } from './http';
Expand Down Expand Up @@ -598,6 +599,7 @@ export default class FireFly extends HttpBase {
subscriptions: string | string[] | FireFlySubscriptionBase,
callback: FireFlyWebSocketCallback,
socketOptions?: WebSocket.ClientOptions | http.ClientRequestArgs,
afterConnect?: FireFlyWebSocketConnectCallback,
): FireFlyWebSocket {
const options: FireFlyWebSocketOptions = {
host: this.options.websocket.host,
Expand All @@ -609,6 +611,7 @@ export default class FireFly extends HttpBase {
reconnectDelay: this.options.websocket.reconnectDelay,
heartbeatInterval: this.options.websocket.heartbeatInterval,
socketOptions: socketOptions,
afterConnect: afterConnect,
};

const handler: FireFlyWebSocketCallback = (socket, event) => {
Expand Down
9 changes: 9 additions & 0 deletions lib/interfaces.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,14 @@ export interface FireFlyOptions extends FireFlyOptionsInput {
};
}

export interface FireFlyWebSocketSender {
send: (json: JSON) => void;
}

export interface FireFlyWebSocketConnectCallback {
(sender: FireFlyWebSocketSender): void | Promise<void>;
}

export interface FireFlyWebSocketOptions {
host: string;
namespace: string;
Expand All @@ -72,6 +80,7 @@ export interface FireFlyWebSocketOptions {
reconnectDelay: number;
heartbeatInterval: number;
socketOptions?: WebSocket.ClientOptions | http.ClientRequestArgs;
afterConnect?: FireFlyWebSocketConnectCallback;
}

// Namespace
Expand Down
21 changes: 17 additions & 4 deletions lib/websocket.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ export class FireFlyWebSocket {
private readonly logger = new Logger(FireFlyWebSocket.name);

private socket?: WebSocket;
private closed = false;
private closed? = () => {};
private pingTimer?: NodeJS.Timeout;
private disconnectTimer?: NodeJS.Timeout;
private reconnectTimer?: NodeJS.Timeout;
Expand Down Expand Up @@ -61,7 +61,7 @@ export class FireFlyWebSocket {
auth,
handshakeTimeout: this.options.heartbeatInterval,
}));
this.closed = false;
this.closed = undefined;

socket
.on('open', () => {
Expand All @@ -83,13 +83,17 @@ export class FireFlyWebSocket {
);
this.logger.log(`Started listening on subscription ${this.options.namespace}:${name}`);
}
if (this.options?.afterConnect !== undefined) {
this.options.afterConnect(this);
}
})
.on('error', (err) => {
this.logger.error('Error', err.stack);
})
.on('close', () => {
if (this.closed) {
this.logger.log('Closed');
this.closed(); // do this after all logging
} else {
this.disconnectDetected = true;
this.reconnect('Closed by peer');
Expand Down Expand Up @@ -156,6 +160,12 @@ export class FireFlyWebSocket {
}
}

send(json: JSON) {
if (this.socket !== undefined) {
this.socket.send(JSON.stringify(json));
}
}

ack(event: FireFlyEventDelivery) {
if (this.socket !== undefined && event.id !== undefined) {
this.socket.send(
Expand All @@ -168,15 +178,18 @@ export class FireFlyWebSocket {
}
}

close() {
this.closed = true;
async close(wait?: boolean): Promise<void> {
const closedPromise = new Promise<void>(resolve => {
this.closed = resolve;
});
this.clearPingTimers();
if (this.socket) {
Copy link
Contributor

@awrichar awrichar Aug 14, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rather than relying on the other "close" handler and dealing with callbacks back and forth, would it be possible to just add another handler here?

const closedPromise = new Promise<void>(resolve => this.socket.on('close', resolve));

Copy link
Contributor Author

@peterbroadhurst peterbroadhurst Aug 21, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure I can for the use case I'm fighting. Jest cannot handle any logging after a test completes, so what I need to know is that the existing callback that does logging has done it.

If I add another callback as you propose, then I'll be relying on the order of invocation of event handlers. Happy to do that if you'd prefer (I have written up the code for it ready to push, but I'll need to test it).

Copy link
Contributor Author

@peterbroadhurst peterbroadhurst Aug 21, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Discussed in a message directly with @awrichar and deciding to merge.

try {
this.socket.close();
} catch (e: any) {
this.logger.warn(`Failed to clean up websocket: ${e.message}`);
}
if (wait) await closedPromise;
this.socket = undefined;
}
}
Expand Down