import { ChildProcess } from "child_process" ; import { actions , AppState , ContentRef , KernelRecord , KernelRef , LocalKernelProps , selectors } from "@nteract/core" ; import { ipcRenderer as ipc } from "electron" ; import { createMainChannel , JupyterConnectionInfo } from "enchannel-zmq-backend" ; import * as jmp from "jmp" ; import sample from "lodash.sample" ; import { ofType , StateObservable } from "redux-observable" ; import { empty , merge , Observable , of , Subscriber } from "rxjs" ; import { catchError , concatMap , first , map , mergeMap , switchMap , tap , timeout } from "rxjs/operators" ; import { launchSpec } from "spawnteract" ; import { KernelspecInfo , Kernelspecs , KernelStatus } from "@nteract/types" ; import { Channels , childOf , ofMessageType , shutdownRequest } from "@nteract/messaging" ; import { Actions } from "../actions" ; param { KernelInfo } kernelSpec param { String } cwd export function launchKernelObservable ( kernelSpec , cwd , kernelRef , contentRef , state ) { const spec = kernelSpec . spec ; return new Observable ( observer => { launchSpec ( spec , { cwd , stdio : [ "ignore" , "pipe" , "pipe" ] } ) . then ( async ( c ) => { const { config , spawn , connectionFile } = c ; const logColor = sample ( [ "#404040" , "#704040" , "#407040" , "#404070" , "#704070" , "#707040" , "#407070" , "#707070" ] ) ; const logStd = ( text ) => { console . log ( `%c${ text }` , `color: ${ logColor }; font-family: Source Code Pro, courier;` ) ; } ; console . log ( `\n>>>> %cLogging kernel ${ kernelSpec . name } (ref ${ kernelRef }) stdout and stderr to javascript console in %cthis color %c  %c <<<<\n` , "font-weight: bold;" , `color: ${ logColor }; font-weight: bold;` , `background-color: ${ logColor }; padding: 2px;` , "color: black" ) ; spawn . stdout ?. on ( "data" , data => { const text = data . toString ( ) ; logStd ( text ) ; observer . next ( actions . kernelRawStdout ( { text , kernelRef } ) ) ; } ) ; spawn . stderr ?. on ( "data" , data => { const text = data . toString ( ) ; logStd ( text ) ; observer . next ( actions . kernelRawStderr ( { text , kernelRef } ) ) ; } ) ; const channels = await createMainChannel ( config , undefined , undefined , undefined , jmp ) ; const kernelInfo = selectors . kernelspecByName ( state , { name : kernelSpec . name } ) ; if ( kernelInfo ) { observer . next ( actions . setKernelMetadata ( { contentRef , kernelInfo } ) ) ; } const kernel = { info : null , type : "zeromq" , hostRef : null , channels , connectionFile , spawn , cwd , kernelSpecName : kernelSpec . name , lastActivity : null , status : KernelStatus . Launched } ; observer . next ( actions . launchKernelSuccessful ( { kernel , kernelRef , contentRef , selectNextKernel : true } ) ) ; observer . next ( actions . setExecutionState ( { kernelStatus : KernelStatus . Launched , kernelRef } ) ) ; observer . complete ( ) ; } ) . catch ( error => { observer . error ( { type : "ERROR" , payload : error , err : true } ) ; } ) ; } ) ; } returns { Observable } export const kernelSpecsObservable = new Observable < Kernelspecs > ( observer => { ipc . on ( "kernel_specs_reply" , ( event , specs ) => { observer . next ( specs ) ; observer . complete ( ) ; } ) ; ipc . send ( "kernel_specs_request" ) ; } ) ; export const launchKernelByNameEpic = ( action$ ) : Observable < Actions > => action$ . pipe ( ofType ( actions . LAUNCH_KERNEL_BY_NAME ) , tap ( ( action ) => { if ( ! action . payload . kernelSpecName ) { throw new Error ( "launchKernelByNameEpic requires a kernel name" ) ; } } ) , mergeMap ( ( action ) => kernelSpecsObservable . pipe ( map ( specs => { const kernelSpec = specs [ action . payload . kernelSpecName ! ] ; if ( kernelSpec ) { return actions . launchKernel ( { contentRef : action . payload . contentRef , cwd : action . payload . cwd , kernelRef : action . payload . kernelRef , kernelSpec , selectNextKernel : action . payload . selectNextKernel } ) ; } else { return actions . launchKernelFailed ( { contentRef : action . payload . contentRef , error : new Error ( `A kernel named ${ action . payload . kernelSpecName } does not appear to be available. Try installing the ${ action . payload . kernelSpecName } kernelspec or selecting a different kernel.` ) , kernelRef : action . payload . kernelRef } ) ; } } ) ) ) ) ; type LaunchKernelResponseActions = actions . KillKernelAction | actions . ErrorAction < "ERROR" > ; param { ActionObservable } action$ export const launchKernelEpic = ( action$ , state$ ) : Observable < Actions > => { const response$ = action$ . pipe ( ofType ( actions . LAUNCH_KERNEL ) , switchMap ( ( action ) => { if ( ! action . payload || ! action . payload . kernelSpec || ! action . payload . kernelRef ) { return of ( actions . launchKernelFailed ( { error : new Error ( "launchKernel needs a kernelSpec and a kernelRef" ) , kernelRef : action . payload && action . payload . kernelRef , contentRef : action . payload . contentRef } ) ) ; } ipc . send ( "nteract:ping:kernel" , action . payload . kernelSpec ) ; const oldKernelRef = selectors . kernelRefByContentRef ( state$ . value , { contentRef : action . payload . contentRef } ) ; let cleanupOldKernel$ = empty ( ) ; if ( oldKernelRef && oldKernelRef !== action . payload . kernelRef ) { cleanupOldKernel$ = of ( actions . killKernel ( { restarting : false , kernelRef : oldKernelRef } ) ) ; } return merge ( launchKernelObservable ( action . payload . kernelSpec , action . payload . cwd , action . payload . kernelRef , action . payload . contentRef , state$ . value ) , cleanupOldKernel$ ) . pipe ( catchError ( ( error ) => of ( actions . launchKernelFailed ( { error , kernelRef : action . payload . kernelRef , contentRef : action . payload . contentRef } ) ) ) ) ; } ) , catchError ( ( error ) => { return of ( { type : "ERROR" , payload : error , error : true } ) ; } ) ) ; return response$ ; } ; type InterruptActions = actions . InterruptKernel | actions . InterruptKernelFailed | actions . InterruptKernelSuccessful ; export const interruptKernelEpic = ( action$ , state$ ) : Observable < InterruptActions > => action$ . pipe ( ofType ( actions . INTERRUPT_KERNEL ) , concatMap ( ( action ) : Observable < InterruptActions > => { const { contentRef } = action . payload ; let kernel ; if ( contentRef ) { kernel = selectors . kernelByContentRef ( state$ . value , { contentRef } ) ; } else { kernel = selectors . currentKernel ( state$ . value ) ; } if ( ! kernel ) { return of ( actions . interruptKernelFailed ( { error : new Error ( "Can't interrupt a kernel we don't have" ) , kernelRef : action . payload . kernelRef } ) ) ; } if ( kernel . type !== "zeromq" || ! kernel . spawn ) { return of ( actions . interruptKernelFailed ( { error : new Error ( "Invalid kernel type for interrupting" ) , kernelRef : action . payload . kernelRef } ) ) ; } const spawn = kernel . spawn ; spawn . kill ( "SIGINT" ) ; return of ( actions . interruptKernelSuccessful ( { kernelRef : action . payload . kernelRef , contentRef } ) ) ; } ) ) ; function killSpawn ( spawn ) { if ( spawn . stdin && spawn . stdin . destroy ) { spawn . stdin . destroy ( ) ; } spawn . stdout ?. destroy ( ) ; spawn . stderr ?. destroy ( ) ; spawn . kill ( "SIGKILL" ) ; } export const killKernelEpic = ( action$ , state$ ) : Observable < Actions > => action$ . pipe ( ofType ( actions . KILL_KERNEL ) , concatMap ( ( action ) => { const kernelRef = action . payload . kernelRef ; if ( ! kernelRef ) { console . warn ( "tried to kill a kernel without a kernelRef" ) ; return empty ( ) ; } const kernel = selectors . kernel ( state$ . value , { kernelRef } ) ; if ( ! kernel ) { console . warn ( "tried to kill a kernel that doesn't exist" ) ; return empty ( ) ; } if ( kernel . type !== "zeromq" ) { return empty ( ) ; } const request = shutdownRequest ( { restart : false } ) ; const shutDownHandling = kernel . channels . pipe ( childOf ( request ) , ofMessageType ( "shutdown_reply" ) , first ( ) , map ( ( msg ) => actions . shutdownReplySucceeded ( { content : msg . content , kernelRef } ) ) , timeout ( 1000 * 2 ) , catchError ( err => of ( actions . shutdownReplyTimedOut ( { kernelRef } ) ) ) , mergeMap ( resultingAction => { kernel . channels . complete ( ) ; if ( kernel . spawn ) { killSpawn ( kernel . spawn ) ; } return merge ( of ( resultingAction ) , of ( actions . killKernelSuccessful ( { kernelRef } ) ) , of ( actions . setExecutionState ( { kernelStatus : KernelStatus . ShuttingDown , kernelRef } ) ) ) ; } ) , catchError ( err => of ( actions . killKernelFailed ( { error : err , kernelRef } ) ) ) ) ; return new Observable ( observer => { const subscription = shutDownHandling . subscribe ( observer ) ; kernel . channels . next ( request ) ; return subscription ; } ) ; } ) ) as Observable < Actions > ; export function watchSpawn ( action$ ) { return action$ . pipe ( ofType ( actions . LAUNCH_KERNEL_SUCCESSFUL ) , switchMap ( ( action ) : Observable < Actions > => { if ( action . payload . kernel . type !== "zeromq" ) { throw new Error ( "kernel.type is not zeromq." ) ; } if ( ! action . payload . kernel . spawn ) { throw new Error ( "kernel.spawn is not provided." ) ; } const spawn = action . payload . kernel . spawn ; return new Observable ( ( observer ) => { spawn . on ( "error" , error => { observer . next ( actions . setExecutionState ( { kernelStatus : KernelStatus . Error , kernelRef : action . payload . kernelRef } ) ) ; observer . error ( { type : "ERROR" , payload : error , err : true } ) ; observer . complete ( ) ; } ) ; spawn . on ( "exit" , ( ) => { observer . next ( actions . setExecutionState ( { kernelStatus : KernelStatus . NotConnected , kernelRef : action . payload . kernelRef } ) ) ; observer . complete ( ) ; } ) ; spawn . on ( "disconnect" , ( ) => { observer . next ( actions . setExecutionState ( { kernelStatus : KernelStatus . NotConnected , kernelRef : action . payload . kernelRef } ) ) ; observer . complete ( ) ; } ) ; } ) ; } ) ) ; }