import Logger from "@supercollider/logger" ; import { packBundle , packMessage , unpackMessage } from "@supercollider/osc" ; import { spawn } from "child_process" ; import * as dgram from "dgram" ; import { EventEmitter } from "events" ; import _ from "lodash" ; import path from "path" ; import { IDisposable , Observable , Subject } from "rx" ; import SendOSC from "./internals/SendOSC" ; import Store from "./internals/Store" ; import { defaults , resolveOptions , ServerArgs , ServerOptions } from "./options" ; import { MsgType , OscType } from "./osc-types" ; import { CallAndResponse , notify } from "./osc/msg" ; import { parseMessage } from "./osc/utils" ; import ServerState from "./ServerState" ; interface ServerObservers { [ name ] : Observable < any > | IDisposable ; } /**
 * Server - starts a SuperCollider synthesis server (scsynth)
 * as a child process. Enables OSC communication, subscribe to process events,
 * send call and response OSC messages.
 *
 * SuperCollider comes with an executable called scsynth
 * which can be communicated with via OSC.
 *
 * To send raw OSC messages:
 * ```js
 * server.send.msg('/s_new', ['defName', 440])
 * ```
 *
 * Raw OSC responses can be subscribed to:
 * ```js
 * server.receive.subscribe(function(msg) {
 *   console.log(msg);
 * });
 * ```
 */ export default class Server extends EventEmitter { options : ServerOptions ; address : string ; /**
     * The process id that nodejs spawn() returns
     */ process : any ; isRunning : boolean ; /**
     * Supports `server.send.msg()` and `server.send.bundle()`
     *
     * You can also subscribe to it and get the OSC messages
     * and bundles that are being sent echoed to you for
     * debugging purposes.
     */ send : SendOSC ; /**
     * A subscribeable stream of OSC events received.
     */ receive : Subject < MsgType > ; /**
     * A subscribeable stream of STDOUT printed by the scsynth process.
     */ stdout : Subject < string > ; /**
     * A subscribeable stream of events related to the scsynth process.
     * Used internally.
     */ processEvents : Subject < string | Error > ; /**
     * Holds the mutable server state
     * including allocators and the node state watcher.
     * If a parent stateStore is supplied then it will store within that.
     */ state : ServerState ; /**
     * The logger used to print messages to the console.
     */ log : Logger ; private osc ? : dgram . Socket ; private _serverObservers : ServerObservers ; param stateStore constructor ( options = { } , stateStore ? ) { super ( ) ; this . options = resolveOptions ( options ) ; this . address = this . options . host + ":" + this . options . serverPort ; this . process = null ; this . isRunning = false ; this . send = new SendOSC ( ) ; this . receive = new Subject ( ) ; this . stdout = new Subject ( ) ; this . processEvents = new Subject ( ) ; this . log = this . _initLogger ( ) ; this . _initEmitter ( ) ; this . _initSender ( ) ; this . _serverObservers = { } ; this . state = new ServerState ( this , stateStore ) ; } private _initLogger ( ) { const log = new Logger ( this . options . debug , this . options . echo , this . options . log ) ; this . send . subscribe ( event => { let out = JSON . stringify ( event . payload || event , ( k , v ) : any => { if ( k === "data" && _ . isArray ( v ) ) { return _ . reduce ( v , ( memo , n ) : string => memo + n . toString ( 16 ) , "" ) ; } return v ; } , 2 ) ; if ( ! this . osc ) { out = "[NOT CONNECTED] " + out ; } log . sendosc ( out ) ; } ) ; this . receive . subscribe ( o => { log . rcvosc ( o ) ; if ( o [ 0 ] === "/fail" ) { log . err ( o ) ; } } , ( err ) => log . err ( err ) ) ; this . stdout . subscribe ( o => { if ( o . match ( /ERROR|FAILURE/ ) ) { log . stderr ( o ) ; } else { log . stdout ( o ) ; } } , ( err ) => log . stderr ( err ) ) ; this . processEvents . subscribe ( o => log . dbug ( o ) , ( err ) => log . err ( err ) ) ; return log ; } deprecated private _initEmitter ( ) { this . receive . subscribe ( msg => { this . emit ( "OSC" , msg ) ; } ) ; this . processEvents . subscribe ( ( ) => { } , err => this . emit ( "exit" , err ) ) ; this . stdout . subscribe ( out => this . emit ( "out" , out ) , out => this . emit ( "stderr" , out ) ) ; } private _initSender ( ) { this . send . on ( "msg" , msg => { if ( this . osc ) { const buf = packMessage ( msg ) ; this . osc . send ( buf , 0 , buf . length , parseInt ( this . options . serverPort ) , this . options . host ) ; } } ) ; this . send . on ( "bundle" , bundle => { if ( this . osc ) { const buf = packBundle ( bundle ) ; this . osc . send ( buf , 0 , buf . length , parseInt ( this . options . serverPort ) , this . options . host ) ; } } ) ; } return args ( ) { const flagMap = { numAudioBusChannels : "-a" , numControlBusChannels : "-c" , numInputBusChannels : "-i" , numOutputBusChannels : "-o" , numBuffers : "-b" , maxNodes : "-n" , maxSynthDefs : "-d" , blockSize : "-z" , hardwareBufferSize : "-Z" , memSize : "-m" , numRGens : "-r" , numWireBufs : "-w" , sampleRate : "-S" , loadDefs : "-D" , inputStreamsEnabled : "-I" , outputStreamsEnabled : "-O" , device : "-H" , verbosity : "-V" , zeroConf : "-R" , restrictedPath : "-P" , ugenPluginsPath : "-U" , password : "-p" , maxLogins : "-l" , } ; const { serverPort , protocol , commandLineOptions } = this . options ; const opts = [ "-u" , serverPort ] ; if ( protocol === "tcp" ) { throw new Error ( "Only udp sockets are supported at this time." ) ; } _ . forEach ( this . options , ( option , argName ) => { const flag = flagMap [ argName ] ; if ( flag ) { if ( option !== defaults [ argName ] ) { opts . push ( flag ) ; if ( _ . isArray ( option ) ) { opts . push ( ... option ) ; } else if ( _ . isString ( option ) ) { opts . push ( option ) ; } else { this . log . err ( `Bad type in server options: ${ argName } ${ option } ${ typeof option }` ) ; } } } } ) ; if ( _ . isArray ( commandLineOptions ) ) { opts . push ( ... commandLineOptions ) ; } return opts . map ( String ) ; } returns { Promise } boot ( ) { return new Promise ( ( resolve , reject ) => { this . isRunning = false ; try { this . _spawnProcess ( ) ; } catch ( e ) { reject ( e ) ; } this . _serverObservers . stdout = Observable . fromEvent ( this . process . stdout , "data" , data => String ( data ) ) ; this . _serverObservers . stdout . subscribe ( e => this . stdout . onNext ( e ) ) ; this . _serverObservers . stderr = Observable . fromEvent ( this . process . stderr , "data" ) . subscribe ( out => { this . stdout . onError ( out ) ; } ) ; let stdoutBuffer = "" ; this . _serverObservers . stdout . takeWhile ( ( text ) : boolean => { stdoutBuffer += text ; return ! stdoutBuffer . match ( /SuperCollider 3 server ready/ ) ; } ) . subscribe ( ( ) => { } , this . log . err , ( ) => { stdoutBuffer = "" ; this . isRunning = true ; resolve ( this ) ; } ) ; setTimeout ( ( ) => { if ( ! this . isRunning ) { reject ( new Error ( "Server failed to start in 3000ms" ) ) ; } } , 3000 ) ; } ) ; } _spawnProcess ( ) { const execPath = this . options . scsynth , args = this . args ( ) ; if ( ! execPath ) { throw new Error ( `Missing options.scsynth executable path` ) ; } const logMsg = "Start process: " + execPath + " " + args . join ( " " ) ; this . processEvents . onNext ( logMsg ) ; const options = { cwd : this . options . cwd || path . dirname ( execPath ) , detached : false , env : this . options . env ? ( this . options . env as NodeJS . ProcessEnv ) : undefined , } ; this . log . dbug ( { execPath , args , options } ) ; this . process = spawn ( execPath , args , options ) ; if ( ! this . process . pid ) { const error = `Failed to boot ${ execPath }` ; this . processEvents . onError ( error ) ; throw new Error ( error ) ; } this . processEvents . onNext ( "pid: " + this . process . pid ) ; const killChild = ( ) : void => { if ( this . process ) { this . process . kill ( "SIGTERM" ) ; this . process = null ; } } ; process . on ( "exit" , killChild ) ; this . process . on ( "error" , ( err ) => { this . processEvents . onError ( err ) ; this . isRunning = false ; } ) ; this . process . on ( "close" , ( code , signal ) => { this . processEvents . onError ( "Server closed. Exit code: " + code + " signal: " + signal ) ; this . isRunning = false ; } ) ; this . process . on ( "exit" , ( code , signal ) => { this . processEvents . onError ( "Server exited. Exit code: " + code + " signal: " + signal ) ; this . isRunning = false ; } ) ; } /**
     * quit
     *
     * kill scsynth process
     * TODO: should send /quit first for shutting files
     */ quit ( ) { if ( this . process ) { this . disconnect ( ) ; this . process . kill ( "SIGTERM" ) ; this . process = null ; } } returns { Promise } connect ( ) { return new Promise ( ( resolve , reject ) => { const udpListening = "udp is listening" ; this . osc = dgram . createSocket ( "udp4" ) ; this . osc . on ( "listening" , ( ) => { this . processEvents . onNext ( udpListening ) ; } ) ; this . osc . on ( "close" , e => { this . processEvents . onNext ( "udp closed: " + e ) ; this . disconnect ( ) ; } ) ; this . _serverObservers . oscMessage = Observable . fromEvent ( this . osc , "message" , msgbuf => unpackMessage ( msgbuf ) ) ; this . _serverObservers . oscMessage . subscribe ( e => this . receive . onNext ( parseMessage ( e ) ) ) ; this . _serverObservers . oscError = Observable . fromEvent ( this . osc , "error" ) ; this . _serverObservers . oscError . subscribe ( e => { this . receive . onError ( e ) ; reject ( e ) ; } ) ; this . callAndResponse ( notify ( ) ) . then ( ( ) => { resolve ( this ) ; } ) ; } ) ; } private disconnect ( ) { if ( this . osc ) { this . osc . close ( ) ; delete this . osc ; } this . _serverObservers = { } ; } deprecated param { String } address param { Array } args sendMsg ( address , args ) { this . send . msg ( [ address , ... args ] ) ; } param { Array } matchArgs param { int } timeout returns { Promise } oscOnce ( matchArgs , timeout = 4000 ) { return new Promise ( ( resolve , reject ) => { const subscription = this . receive . subscribe ( msg => { const command = msg . slice ( 0 , matchArgs . length ) ; if ( _ . isEqual ( command , matchArgs ) ) { const payload = msg . slice ( matchArgs . length ) ; resolve ( payload ) ; dispose ( ) ; } } ) ; const tid = setTimeout ( ( ) => { dispose ( ) ; reject ( new Error ( `Timed out waiting for OSC response: ${ JSON . stringify ( matchArgs ) }` ) ) ; } , timeout ) ; function dispose ( ) { subscription . dispose ( ) ; clearTimeout ( tid ) ; } } ) ; } param { int } timeout returns { Promise } callAndResponse ( callAndResponse , timeout = 4000 ) { const promise = this . oscOnce ( callAndResponse . response , timeout ) ; this . send . msg ( callAndResponse . call ) ; return promise ; } } param { object } options param { Store } store returns { Promise } export async function boot ( options = { } , store = null ) { const s = new Server ( options , store ) ; await s . boot ( ) ; await s . connect ( ) ; return s ; }