import * as WebSocket from 'ws' ; import State from '../State' ; import Router from '../control/Router' ; import Manager from '../control/Manager' ; import { tryParseJSON , parseEndpoint } from '../utility' ; /** Creates an ```express-ws``` middleware function to handle new WebSocket connections. Receives messages in the form of an object whose keys represent endpoints in the format 'METHOD /path' and whose values are objects containing the arguments to be passed to the associated endpoint.
 */ const ws = ( router , callback , accept = [ ] , join = [ ] ) : Function => { const customMethods = [ 'subscribe' , 'unsubscribe' ] ; const manager = Manager . access ( ) ; const newPeer = ( socket ) => { const listener = ( paths ) => { console . log ( `${ paths } changed -- notifying peers.` ) ; socket . send ( JSON . stringify ( { UPDATE : paths } ) ) ; } ; manager . listen ( listener ) ; socket . on ( 'message' , ( msg ) => { console . log ( msg ) ; const data = tryParseJSON ( msg ) ; if ( typeof data !== 'object' ) { return ; } Object . entries ( data ) . forEach ( ( [ method , args ] ) => { method = method . toLowerCase ( ) ; if ( method === 'update' && ( typeof args === 'string' || Array . isArray ( args ) ) ) { return manager . invalidate ( args , true ) ; } return null ; } ) ; } ) ; socket . on ( 'close' , ( ) => { manager . unlisten ( listener ) ; } ) ; } ; const newClient = ( socket , req ) => { const client = ( path , state , render = true ) => { const _req = { } ; const _res = { locals : state , stream : ( data ) => socket . send ( JSON . stringify ( { [ path ] : render ? data . render ( ) : data } ) ) , } ; callback ( _req , _res ) ; } ; socket . on ( 'message' , async ( msg ) => { const data = tryParseJSON ( msg ) ; if ( typeof data !== 'object' ) { return client ( '?' , State . BAD_REQUEST ( 'Invalid Format' ) ) ; } const requests = Object . keys ( data ) ; return requests . forEach ( async ( endpoint ) => { const { method , path } = parseEndpoint ( endpoint , customMethods ) ; if ( ! method ) { return client ( endpoint , State . BAD_REQUEST ( 'Invalid Method' ) ) ; } const args = { ... req . cookies , ... data [ endpoint ] } ; if ( method === 'unsubscribe' ) { manager . unsubscribe ( client , path ) ; return client ( endpoint , State . OK ( ) , false ) ; } if ( method === 'subscribe' ) { const state = await router . request ( 'get' , path , args ) ; manager . subscribe ( client , state . $query ) ; return client ( endpoint , state , false ) ; } return client ( endpoint , await router . request ( method , path , args ) , false ) ; } ) ; } ) ; socket . on ( 'close' , ( ) => { manager . unsubscribe ( client ) ; } ) ; } ; Promise . resolve ( accept ) . then ( ( ips ) => { accept = ips ; } ) ; Promise . resolve ( join ) . then ( ( peers ) => { peers . forEach ( ( uri ) => { newPeer ( new WebSocket ( uri ) ) ; } ) ; } ) ; return ( socket , req ) => { if ( Array . isArray ( accept ) && accept . indexOf ( req . connection . remoteAddress ) !== - 1 ) { return newPeer ( socket ) ; } return newClient ( socket , req ) ; } ; } ; export default ws ;