void pumpEvents ( InputStream eventStream ) { try { Deserializer deserializer = new Deserializer ( eventStream , refLoader ) ; IEvent event = null ; while ( ( event = deserializer . deserialize ( ) ) != null ) { switch ( event . getType ( ) ) { case APPEND_STDERR : case APPEND_STDOUT : // Ignore these two on activity heartbeats. GH-117 break ; default : lastActivity = System . currentTimeMillis ( ) ; break ; } try { switch ( event . getType ( ) ) { case QUIT : eventBus . post ( event ) ; return ; case IDLE : eventBus . post ( new SlaveIdle ( stdinWriter ) ) ; break ; case BOOTSTRAP : clientCharset = Charset . forName ( ( ( BootstrapEvent ) event ) . getDefaultCharsetName ( ) ) ; stdinWriter = new OutputStreamWriter ( stdin , clientCharset ) ; eventBus . post ( event ) ; break ; case APPEND_STDERR : case APPEND_STDOUT : assert streamsBuffer . getFilePointer ( ) == streamsBuffer . length ( ) ; final long bufferStart = streamsBuffer . getFilePointer ( ) ; IStreamEvent streamEvent = ( IStreamEvent ) event ; streamEvent . copyTo ( streamsBufferWrapper ) ; final long bufferEnd = streamsBuffer . getFilePointer ( ) ; event = new OnDiskStreamEvent ( event . getType ( ) , streamsBuffer , bufferStart , bufferEnd ) ; eventBus . post ( event ) ; break ; default : eventBus . post ( event ) ; } } catch ( Throwable t ) { warnStream . println ( "Event bus dispatch error: " + t . toString ( ) ) ; t . printStackTrace ( warnStream ) ; } } lastActivity = null ; } catch ( Throwable e ) { if ( ! stopping ) { warnStream . println ( "Event stream error: " + e . toString ( ) ) ; e . printStackTrace ( warnStream ) ; } } }
Pump events from event stream .