override subscription query once .subscription initiated

[luisweir]

Hi,

I have a question which is somehow related related to #292.

Is there a way to override the actual payload of the subscription query AFTER initialising the client? let me explain:

I initialise the client via constructor. The client on listener has a timer such as every 59 mins the connection is refreshed with a new token. So far so good.

public constructor(env: any, wsUrl: string, oauthUrl: string, oauthOptions: any, call: Call) {
        this.offset = env.OFFSET;
        let activeSocket: WebSocket, timedOut: any;
        this.client = createWSClient({
            webSocketImpl: WebSocket,
            url: wsUrl,
            connectionParams: async() => {
                // want to override query here...
                return {
                    'Authorization': `Bearer ${(await call.fetchToken(oauthUrl, oauthOptions)).access_token}`,
                    'x-app-key': `${env.APP_KEY}`
                };
            },
            shouldRetry: () => true,
            lazy: true,
            keepAlive: (env.PING), // frequency to ping server
            on: {
                connected: (socket: any) => {
                    activeSocket = socket;
                    log.info(`Connected to socket ${wsUrl}`);
                    setTimeout(() => {
                        log.info('Refreshing connection with new token');
                        activeSocket.close(4408, 'Token Expired');
                    }, 30000 );
                // }, env.TOKEN_EXPIRY );
                },
                closed: (event: any) => {
                    log.error(`Socket closed with event ${event.code} ${event.reason}`);
                },
                ping: (received) => {
                    if (!received) // sent
                        timedOut = setTimeout(() => {
                            if (activeSocket.readyState === WebSocket.OPEN)
                                activeSocket.close(4408, 'Request Timeout');
                        }, env.PING / 2); // if pong not received within this timeframe then recreate connection
                },
                pong: (received) => {
                    if (received) clearTimeout(timedOut); // pong is received, clear connection close timeout
                },
                error: (error) => {
                    log.error(error);
                }
            }
        });
    }

However there is a portion of the GraphQL query (an offset) which will be different whenever the connection is refreshed. I want to use the new offset however don't know how to override the already passed query from the connectionParams.

    public async execute<T>(env: any) {
        const query = this.createQuery(env.CHAIN,this.offset,env.HOTELID,env.DELTA);
        const payload: SubscribePayload = {query};
        return new Promise<T>((resolve, reject) => {
            let result: any;
            log.info(`Subscribing to ${query}`);
            this.client.subscribe<T>(payload, {
                next: (data) => {
                    result = data;
                    this.offset = result.data.newEvent.metadata.offset;
                    log.info(`Processed ${result.data.newEvent.eventName} event with: offset ${result.data.newEvent.metadata.offset}, primaryKey ${result.data.newEvent.primaryKey}, HotelId ${(result.data.newEvent.hotelId) ? result.data.newEvent.hotelId : null}`);
                },
                error: (error) => {
                    log.error(error);
                    reject();
                },
                complete: () => resolve(result)
            });
        });
    }

I could of course add a timer around the execute method and just drop the entire connection and restart it but seems inefficient when the timer approach on the on listener works well if I could only just be able to override the query.

Any help would be appreciated.

[enisdenjo]

Is there a way to override the actual payload of the subscription query AFTER initialising the client?

Once a subscription query establishes a subscription, it cannot change. You have to complete the subscription and create a new one with new arguments.

The connectionParams are, and can only be, issued once per connection. If you want to change them, you have to reconnect and provide new params during the connect.

On the other hand, if you want to access the connectionParams when subscribing, you can simply create a global variable currentConnectionParams, assign it always on connect and access it when you need it.

...
connectionParams: async() => {
    const params = 
+   this.currentConnectionParams = {
+       'Authorization': `Bearer ${(await call.fetchToken(oauthUrl, oauthOptions)).access_token}`,
+       'x-app-key': `${env.APP_KEY}`
+   };
    return params;
},
...

[luisweir]

ok thanks.

Was able to solve the problem by disposing the connection entirely and creating a new one every 60 mins which is the duration of the token (once token expires server ends the connection....).

public async start(): Promise<void> {
        let client: Client = this.getClient();
        const initiate = async() => {
            log.debug('Initiating a new connection');
            client.dispose();
            this.activeSocket?.terminate();
            if (this.env.STATS) {
                this.printAndClearStats();
            }
            client = this.getClient();
            try {
                await this.subscribe(client);
            } catch (error) {
                log.error(error);
            }
        };
        setImmediate(() => initiate());
        setInterval(() => {initiate();}, this.env.TOKEN_EXPIRY);
    }
}