public static ExecutionControl remoteInputOutput ( InputStream input , OutputStream output , Map < String , OutputStream > outputStreamMap , Map < String , InputStream > inputStreamMap , BiFunction < ObjectInput , ObjectOutput , ExecutionControl > factory ) throws IOException { ExecutionControl [ ] result = new ExecutionControl [ 1 ] ; Map < String , OutputStream > augmentedStreamMap = new HashMap <> ( outputStreamMap ) ; ObjectOutput commandOut = new ObjectOutputStream ( Util . multiplexingOutputStream ( "$command" , output ) ) ; for ( Entry < String , InputStream > e : inputStreamMap . entrySet ( ) ) { InputStream in = e . getValue ( ) ; OutputStream inTarget = Util . multiplexingOutputStream ( e . getKey ( ) , output ) ; augmentedStreamMap . put ( "$" + e . getKey ( ) + "-input-requested" , new OutputStream ( ) { @ Override public void write ( int b ) throws IOException { //value ignored, just a trigger to read from the input try { int r = in . read ( ) ; if ( r == ( - 1 ) ) { inTarget . write ( TAG_CLOSED ) ; } else { inTarget . write ( new byte [ ] { TAG_DATA , ( byte ) r } ) ; } } catch ( InterruptedIOException exc ) { try { result [ 0 ] . stop ( ) ; } catch ( ExecutionControlException ex ) { debug ( ex , "$" + e . getKey ( ) + "-input-requested.write" ) ; } } catch ( IOException exc ) { byte [ ] message = exc . getMessage ( ) . getBytes ( "UTF-8" ) ; inTarget . write ( TAG_EXCEPTION ) ; inTarget . write ( ( message . length >> 0 ) & 0xFF ) ; inTarget . write ( ( message . length >> 8 ) & 0xFF ) ; inTarget . write ( ( message . length >> 16 ) & 0xFF ) ; inTarget . write ( ( message . length >> 24 ) & 0xFF ) ; inTarget . write ( message ) ; } } } ) ; } PipeInputStream commandIn = new PipeInputStream ( ) ; OutputStream commandInTarget = commandIn . createOutput ( ) ; augmentedStreamMap . put ( "$command" , commandInTarget ) ; new DemultiplexInput ( input , augmentedStreamMap , Arrays . asList ( commandInTarget ) ) . start ( ) ; return result [ 0 ] = factory . apply ( new ObjectInputStream ( commandIn ) , commandOut ) ; }
Creates an ExecutionControl for given packetized input and output . The given InputStream is de - packetized and content forwarded to ObjectInput and given OutputStreams . The ObjectOutput and values read from the given InputStream are packetized and sent to the given OutputStream .