public AsyncWork < OutputResultType , OutputErrorType > write ( InputType data ) throws IOException { do { SynchronizationPoint < NoException > lk ; synchronized ( waiting ) { // if cancelled or errored, return immediately if ( error != null ) return new AsyncWork <> ( null , error ) ; if ( cancelled != null ) return new AsyncWork <> ( null , null , cancelled ) ; AsyncWork < OutputResultType , OutputErrorType > op ; // if ready, write immediately if ( isReady ) { isReady = false ; op = lastWrite = executor . execute ( data ) ; lastWrite . listenInline ( new WriteListener ( data , op , null ) ) ; return op ; } // not ready if ( ! waiting . isFull ( ) ) { op = new AsyncWork <> ( ) ; waiting . addLast ( new Pair <> ( data , op ) ) ; lastWrite = op ; return op ; } // full if ( lock != null ) throw new IOException ( "Concurrent write" ) ; lock = new SynchronizationPoint <> ( ) ; lk = lock ; } lk . block ( 0 ) ; } while ( true ) ; }
Queue the data to write . If there is no pending write the write operation is started . If too many write operations are pending the method is blocking .