import { noop } from "lodash" ; import ReactDOM from "react-dom" ; import { CacheConfig , ConcreteBatch , Disposable , Variables , } from "react-relay-network-modern/es" ; import { OperationOptions , SubscriptionClient , } from "subscriptions-transport-ws" ; import { ACCESS_TOKEN_PARAM , BUNDLE_CONFIG_PARAM , BUNDLE_ID_PARAM , CLIENT_ID_PARAM , } from "coral-common/constants" ; import { ERROR_CODES } from "coral-common/errors" ; /** Time after an inactive client is closed */ const STALE_CLIENT_TIMEOUT = 10 * 1000 ; const BATCH_PERIOD = 1000 ; export const enum CONNECTION_STATUS { CONNECTED = "CONNECTED" , CONNECTING = "CONNECTING" , DISCONNECTED = "DISCONNECTED" } /**
 * SubscriptionRequest contains the subscription
 * request data that comes from Relay.
 */ export interface SubscriptionRequest { operation : ConcreteBatch ; variables : Variables ; cacheConfig : CacheConfig ; observer : any ; /** batch contains response data waiting to be processed */ batch : any [ ] ; subscribe : ( ) => void ; unsubscribe : ( ( ) => void ) | null ; } export type ConnectionStatusListenerCallback = ( status : CONNECTION_STATUS ) => void ; type Unlisten = ( ) => void ; /**
 * ManagedSubscriptionClient builds on top of `SubscriptionClient`
 * and manages the websocket connection economically. A connection is
 * only establish when there is at least 1 active subscription and closes
 * when there is no more active subscriptions.
 */ export interface ManagedSubscriptionClient { /**
     * Subscribe to a GraphQL subscription, this is usually called from
     * the SubscriptionFunction provided to Relay.
     */ subscribe ( operation , variables , cacheConfig , observer ) : Disposable ; /** Pauses all active subscriptions causing websocket connection to close. */ pause ( ) : void ; /** Resume all subscriptions eventually causing websocket to start with new connection parameters */ resume ( ) : void ; /** Sets access token, you should call pause() before this and resume() afterwards. */ setAccessToken ( accessToken ? ) : void ; /** Returns current connection status */ getConnectionStatus ( ) : CONNECTION_STATUS ; /** Allows reacting to connection status changes */ on ( status , callback ) : Unlisten ; } /**
 * Batch react state updates.
 */ function batchReactUpdates ( callback ) { ReactDOM . unstable_batchedUpdates ( callback ) ; } /**
 * Handle all batched response data of given request.
 */ function runBatch ( request ) { request . batch . forEach ( ( data ) => { request . observer . onNext ( { data } ) ; } ) ; request . batch = [ ] ; } /**
 * Handle batched response data of given requests in a loop.
 */ function batchLoop ( requests , timeout ) { setTimeout ( ( ) => { batchReactUpdates ( ( ) => { requests . forEach ( ( r ) => { runBatch ( r ) ; } ) ; } ) ; batchLoop ( requests , timeout ) ; } , timeout ) ; } param url param clientID export default function createManagedSubscriptionClient ( url , clientID , bundle , bundleConfig ) { const requests = [ ] ; const connectionStatusListeners = { [ CONNECTION_STATUS . CONNECTED ] : [ ] , [ CONNECTION_STATUS . CONNECTING ] : [ ] , [ CONNECTION_STATUS . DISCONNECTED ] : [ ] , } ; let connectionStatus = CONNECTION_STATUS . DISCONNECTED ; let subscriptionClient = null ; let paused = false ; let accessToken ; let closeStaleClientTimeout = null ; batchLoop ( requests , BATCH_PERIOD ) ; const startCloseStaleClientTimeout = ( ) => { if ( subscriptionClient && ! needWebsocketConnection ( ) ) { closeStaleClientTimeout = setTimeout ( ( ) => { if ( subscriptionClient && ! needWebsocketConnection ( ) ) { closeClient ( ) ; if ( process . env . NODE_ENV !== "production" ) { console . debug ( "subscription client disconnecting, no more subscriptions being tracked" ) ; } } } , STALE_CLIENT_TIMEOUT ) ; } } ; const clearStaleClientTimeout = ( ) => { if ( closeStaleClientTimeout ) { clearTimeout ( closeStaleClientTimeout ) ; closeStaleClientTimeout = null ; } } ; const setConnectionStatus = ( status ) => { connectionStatus = status ; connectionStatusListeners [ status ] . forEach ( ( callback ) => callback ( status ) ) ; } ; const onConnectionStatus = ( status , callback ) : Unlisten => { connectionStatusListeners [ status ] . push ( callback ) ; return ( ) => { const index = connectionStatusListeners [ status ] . indexOf ( callback ) ; if ( index >= 0 ) { connectionStatusListeners [ status ] . splice ( index ) ; } } ; } ; const on = ( status , callback ) : Unlisten => { const statusArray = ! Array . isArray ( status ) ? [ status ] : status ; const unlistenArray = statusArray . map ( ( s ) => onConnectionStatus ( s , callback ) ) ; if ( needWebsocketConnection ( ) ) { clearStaleClientTimeout ( ) ; ensureClientIsCreated ( ) ; } else { startCloseStaleClientTimeout ( ) ; } return ( ) => unlistenArray . forEach ( ( unlisten ) => unlisten ( ) ) ; } ; const hasActiveSubscriptions = ( ) => requests . length > 0 && requests . some ( ( r ) => r . unsubscribe ) ; const hasConnectionStatusListeners = ( status ) => connectionStatusListeners [ status ] . length > 0 ; const needWebsocketConnection = ( ) => hasActiveSubscriptions ( ) || hasConnectionStatusListeners ( CONNECTION_STATUS . CONNECTED ) || hasConnectionStatusListeners ( CONNECTION_STATUS . CONNECTING ) || hasConnectionStatusListeners ( CONNECTION_STATUS . CONNECTED ) || hasConnectionStatusListeners ( CONNECTION_STATUS . CONNECTING ) ; const closeClient = ( ) => { if ( subscriptionClient ) { subscriptionClient . close ( ) ; ( subscriptionClient as any ) . clearMaxConnectTimeout ( ) ; ( subscriptionClient as any ) . clearTryReconnectTimeout ( ) ; subscriptionClient = null ; } } ; const ensureClientIsCreated = ( ) => { if ( subscriptionClient ) { return ; } subscriptionClient = new SubscriptionClient ( url , { reconnect : true , timeout : 60000 , connectionCallback : ( err ) => { if ( err ) { if ( ( ( err as unknown ) as Error ) . message === ERROR_CODES . LIVE_UPDATES_DISABLED && subscriptionClient ) { subscriptionClient . close ( ) ; } } } , connectionParams : { [ ACCESS_TOKEN_PARAM ] : accessToken , [ CLIENT_ID_PARAM ] : clientID , [ BUNDLE_ID_PARAM ] : bundle , [ BUNDLE_CONFIG_PARAM ] : bundleConfig , } , } ) ; subscriptionClient . onConnected ( ( payload ) => { setConnectionStatus ( CONNECTION_STATUS . CONNECTED ) ; } ) ; subscriptionClient . onConnecting ( ( payload ) => { setConnectionStatus ( CONNECTION_STATUS . CONNECTING ) ; } ) ; subscriptionClient . onReconnecting ( ( payload ) => { setConnectionStatus ( CONNECTION_STATUS . CONNECTING ) ; } ) ; subscriptionClient . onReconnected ( ( payload ) => { setConnectionStatus ( CONNECTION_STATUS . CONNECTED ) ; } ) ; subscriptionClient . onDisconnected ( ( payload ) => { setConnectionStatus ( CONNECTION_STATUS . DISCONNECTED ) ; } ) ; } ; const subscribe = ( operation , variables , cacheConfig , observer ) => { clearStaleClientTimeout ( ) ; const request = { operation , variables , cacheConfig , observer , batch : [ ] , unsubscribe : null , subscribe : noop , } ; request . subscribe = ( ) => { ensureClientIsCreated ( ) ; if ( ! operation . text && ! operation . id ) { throw Error ( "Neither subscription query nor id was provided." ) ; } const opts = { operationName : operation . name , query : operation . text || "PERSISTED_QUERY" , variables , } ; if ( ! operation . text ) { opts . id = operation . id ; } const subscription = subscriptionClient ! . request ( opts ) . subscribe ( { next ( { data } ) { request . batch . push ( data ) ; } , } ) ; request . unsubscribe = ( ) => { batchReactUpdates ( ( ) => { runBatch ( request ) ; } ) ; subscription . unsubscribe ( ) ; } ; } ; requests . push ( request ) ; if ( ! paused ) { request . subscribe ( ) ; if ( process . env . NODE_ENV !== "production" ) { console . debug ( `+1 [${ requests . length - 1 } + 1 = ${ requests . length }] subscribe called for subscription:` , request . operation ?. name , "with variables:" , JSON . stringify ( request . variables ) ) ; } } let disposed = false ; return { dispose : ( ) => { if ( disposed ) { return ; } disposed = true ; const i = requests . findIndex ( ( r ) => r === request ) ; if ( i !== - 1 ) { if ( request . unsubscribe ) { request . unsubscribe ( ) ; } requests . splice ( i , 1 ) ; startCloseStaleClientTimeout ( ) ; } if ( process . env . NODE_ENV !== "production" ) { console . debug ( `-1 [${ requests . length + 1 } - 1 = ${ requests . length }] dispose called for subscription:` , request . operation ?. name , "with variables:" , JSON . stringify ( request . variables ) ) ; } } , } ; } ; const pause = ( ) => { paused = true ; for ( const r of requests ) { if ( r . unsubscribe ) { r . unsubscribe ( ) ; r . unsubscribe = null ; } } closeClient ( ) ; } ; const resume = ( ) => { for ( const r of requests ) { if ( ! r . unsubscribe ) { r . subscribe ( ) ; } } paused = false ; } ; const setAccessToken = ( nextAccessToken ? ) => { accessToken = nextAccessToken ; } ; const getConnectionStatus = ( ) => connectionStatus ; return Object . freeze ( { subscribe , pause , resume , setAccessToken , getConnectionStatus , on , } ) ; }