@ -28,12 +28,7 @@ define([
this . id = null ;
this . name = name ;
this . channels = {
'shell' : null ,
'iopub' : null ,
'stdin' : null
} ;
this . ws = null ;
this . kernel _service _url = kernel _service _url ;
this . kernel _url = null ;
@ -429,7 +424,7 @@ define([
Kernel . prototype . start _channels = function ( ) {
/ * *
* Start the ` shell ` and ` iopub ` channels .
* Start the websocket channels .
* Will stop and restart them if they already exist .
*
* @ function start _channels
@ -440,16 +435,12 @@ define([
console . log ( "Starting WebSockets:" , ws _host _url ) ;
var channel _url = function ( channel ) {
return [
this . ws = new this . WebSocket ( [
that . ws _url ,
utils . url _join _encode ( that . kernel _url , channel ) ,
utils . url _join _encode ( that . kernel _url , 'channels' ) ,
"?session_id=" + that . session _id
] . join ( '' ) ;
} ;
this . channels . shell = new this . WebSocket ( channel _url ( "shell" ) ) ;
this . channels . stdin = new this . WebSocket ( channel _url ( "stdin" ) ) ;
this . channels . iopub = new this . WebSocket ( channel _url ( "iopub" ) ) ;
] . join ( '' )
) ;
var already _called _onclose = false ; // only alert once
var ws _closed _early = function ( evt ) {
@ -489,28 +480,22 @@ define([
that . _ws _closed ( ws _host _url , true ) ;
} ;
for ( var c in this . channels ) {
this . channels [ c ] . onopen = $ . proxy ( this . _ws _opened , this ) ;
this . channels [ c ] . onclose = ws _closed _early ;
this . channels [ c ] . onerror = ws _error ;
}
this . ws . onopen = $ . proxy ( this . _ws _opened , this ) ;
this . ws . onclose = ws _closed _early ;
this . ws . onerror = ws _error ;
// switch from early-close to late-close message after 1s
setTimeout ( function ( ) {
for ( var c in that . channels ) {
if ( that . channels [ c ] !== null ) {
that . channels [ c ] . onclose = ws _closed _late ;
}
if ( that . ws !== null ) {
that . ws . onclose = ws _closed _late ;
}
} , 1000 ) ;
this . channels . shell . onmessage = $ . proxy ( this . _handle _shell _reply , this ) ;
this . channels . iopub . onmessage = $ . proxy ( this . _handle _iopub _message , this ) ;
this . channels . stdin . onmessage = $ . proxy ( this . _handle _input _request , this ) ;
this . ws . onmessage = $ . proxy ( this . _handle _ws _message , this ) ;
} ;
Kernel . prototype . _ws _opened = function ( evt ) {
/ * *
* Handle a websocket entering the open state ,
* signaling that the kernel is connected when all channels are open .
* signaling that the kernel is connected when websocket is open .
*
* @ function _ws _opened
* /
@ -522,8 +507,7 @@ define([
Kernel . prototype . _ws _closed = function ( ws _url , error ) {
/ * *
* Handle a websocket entering the closed state . This closes the
* other communication channels if they are open . If the websocket
* Handle a websocket entering the closed state . If the websocket
* was not closed due to an error , try to reconnect to the kernel .
*
* @ function _ws _closed
@ -560,27 +544,23 @@ define([
Kernel . prototype . stop _channels = function ( ) {
/ * *
* Close the websocket channels . After successful close , the value
* in ` this. channels[channel_name] ` will be null .
* Close the websocket . After successful close , the value
* in ` this. ws ` will be null .
*
* @ function stop _channels
* /
var that = this ;
var close = function ( c ) {
return function ( ) {
if ( that . channels [ c ] && that . channels [ c ] . readyState === WebSocket . CLOSED ) {
that . channels [ c ] = null ;
}
} ;
var close = function ( ) {
if ( that . ws && that . ws . readyState === WebSocket . CLOSED ) {
that . ws = null ;
}
} ;
for ( var c in this . channels ) {
if ( this . channels [ c ] !== null ) {
if ( this . channels [ c ] . readyState === WebSocket . OPEN ) {
this . channels [ c ] . onclose = close ( c ) ;
this . channels [ c ] . close ( ) ;
} else {
close ( c ) ( ) ;
}
if ( this . ws !== null ) {
if ( this . ws . readyState === WebSocket . OPEN ) {
this . ws . onclose = close ;
this . ws . close ( ) ;
} else {
close ( ) ;
}
}
} ;
@ -588,20 +568,18 @@ define([
Kernel . prototype . is _connected = function ( ) {
/ * *
* Check whether there is a connection to the kernel . This
* function only returns true if all channel objects have been
* created and ha ve a state of WebSocket . OPEN .
* function only returns true if websocket has been
* created and ha s a state of WebSocket . OPEN .
*
* @ function is _connected
* @ returns { bool } - whether there is a connection
* /
for ( var c in this . channels ) {
// if any channel is not ready, then we're not connected
if ( this . channels [ c ] === null ) {
return false ;
}
if ( this . channels [ c ] . readyState !== WebSocket . OPEN ) {
return false ;
}
// if any channel is not ready, then we're not connected
if ( this . ws === null ) {
return false ;
}
if ( this . ws . readyState !== WebSocket . OPEN ) {
return false ;
}
return true ;
} ;
@ -615,12 +593,7 @@ define([
* @ function is _fully _disconnected
* @ returns { bool } - whether the kernel is fully disconnected
* /
for ( var c in this . channels ) {
if ( this . channels [ c ] === null ) {
return true ;
}
}
return false ;
return ( this . ws === null ) ;
} ;
Kernel . prototype . send _shell _message = function ( msg _type , content , callbacks , metadata , buffers ) {
@ -633,7 +606,8 @@ define([
throw new Error ( "kernel is not connected" ) ;
}
var msg = this . _get _msg ( msg _type , content , metadata , buffers ) ;
this . channels . shell . send ( serialize . serialize ( msg ) ) ;
msg . channel = 'shell' ;
this . ws . send ( serialize . serialize ( msg ) ) ;
this . set _callbacks _for _msg ( msg . header . msg _id , callbacks ) ;
return msg . header . msg _id ;
} ;
@ -784,7 +758,8 @@ define([
} ;
this . events . trigger ( 'input_reply.Kernel' , { kernel : this , content : content } ) ;
var msg = this . _get _msg ( "input_reply" , content ) ;
this . channels . stdin . send ( serialize . serialize ( msg ) ) ;
msg . channel = 'stdin' ;
this . ws . send ( serialize . serialize ( msg ) ) ;
return msg . header . msg _id ;
} ;
@ -877,15 +852,28 @@ define([
this . last _msg _callbacks = { } ;
}
} ;
/ * *
* @ function _handle _shell _reply
* /
Kernel . prototype . _handle _shell _reply = function ( e ) {
serialize . deserialize ( e . data , $ . proxy ( this . _finish _shell _reply , this ) ) ;
Kernel . prototype . _handle _ws _message = function ( e ) {
serialize . deserialize ( e . data , $ . proxy ( this . _finish _ws _message , this ) ) ;
} ;
Kernel . prototype . _finish _shell _reply = function ( reply ) {
Kernel . prototype . _finish _ws _message = function ( msg ) {
switch ( msg . channel ) {
case 'shell' :
this . _handle _shell _reply ( msg ) ;
break ;
case 'iopub' :
this . _handle _iopub _message ( msg ) ;
break ;
case 'stdin' :
this . _handle _input _request ( msg ) ;
break ;
default :
console . error ( "unrecognized message channel" , msg . channel , msg ) ;
}
} ;
Kernel . prototype . _handle _shell _reply = function ( reply ) {
this . events . trigger ( 'shell_reply.Kernel' , { kernel : this , reply : reply } ) ;
var content = reply . content ;
var metadata = reply . metadata ;
@ -1030,12 +1018,7 @@ define([
*
* @ function _handle _iopub _message
* /
Kernel . prototype . _handle _iopub _message = function ( e ) {
serialize . deserialize ( e . data , $ . proxy ( this . _finish _iopub _message , this ) ) ;
} ;
Kernel . prototype . _finish _iopub _message = function ( msg ) {
Kernel . prototype . _handle _iopub _message = function ( msg ) {
var handler = this . get _iopub _handler ( msg . header . msg _type ) ;
if ( handler !== undefined ) {
handler ( msg ) ;
@ -1045,12 +1028,7 @@ define([
/ * *
* @ function _handle _input _request
* /
Kernel . prototype . _handle _input _request = function ( e ) {
serialize . deserialize ( e . data , $ . proxy ( this . _finish _input _request , this ) ) ;
} ;
Kernel . prototype . _finish _input _request = function ( request ) {
Kernel . prototype . _handle _input _request = function ( request ) {
var header = request . header ;
var content = request . content ;
var metadata = request . metadata ;
Min RK
11 years ago