src/main/java/net/batchik/crdt/fiber/FiberChannels.java
/**
     * Constructs a stream that reads bytes from the given channel.
     *
     * <p> The stream will not be buffered, and it will not support the {@link
     * InputStream#mark mark} or {@link InputStream#reset reset} methods.  The
     * stream will be safe for access by multiple concurrent threads.  Closing
     * the stream will in turn cause the channel to be closed.  </p>
     *
     * @param  ch
     *         The channel from which bytes will be read
     *
     * @return  A new input stream
     *
     * @since 1.7
     */
Constructs a stream that reads bytes from the given channel.
 The stream will not be buffered, and it will not support the {@link
InputStream#mark mark} or  InputStream#reset reset methods.  The
stream will be safe for access by multiple concurrent threads.  Closing
the stream will in turn cause the channel to be closed.

@param  ch
The channel from which bytes will be read

@return  A new input stream


Constructs a stream that reads bytes from the given channel . The stream will not be buffered and it will not support the { @link InputStream#mark mark } or InputStream#reset reset methods . The stream will be safe for access by multiple concurrent threads . Closing the stream will in turn cause the channel to be closed . @param ch The channel from which bytes will be read @return A new input stream
public static InputStream newInputStream(final AsynchronousByteChannel ch) throws SuspendExecution {
        checkNotNull(ch, "ch");
        return new InputStream() {
            private ReentrantLock ilock = new ReentrantLock();
            private ByteBuffer bb = null;
            private byte[] bs = null;           
            private byte[] b1 = null;
            @Override
            @Suspendable
            public int read() throws IOException {
                try {
                    ilock.lock();
                    if (b1 == null)
                        b1 = new byte[1];
                    int n = this.read(b1);
                    if (n == 1)
                        return b1[0] & 0xff;
                    return -1;
                } finally {
                    ilock.unlock();
                }
            }
            @Override
            @Suspendable
            public int read(byte[] bs, int off, int len)
                    throws IOException
            {
                try {
                    ilock.lock();
                    if ((off < 0) || (off > bs.length) || (len < 0) ||
                            ((off + len) > bs.length) || ((off + len) < 0)) {
                        throw new IndexOutOfBoundsException();
                    } else if (len == 0)
                        return 0;
                    ByteBuffer bb = ((this.bs == bs)
                            ? this.bb
                            : ByteBuffer.wrap(bs));
                    bb.position(off);
                    bb.limit(Math.min(off + len, bb.capacity()));
                    this.bb = bb;
                    this.bs = bs;
                    boolean interrupted = false;
                    try {
                        for (; ; ) {
                            try {
                                return ch.read(bb).get();
                            } catch (ExecutionException ee) {
                                throw new IOException(ee.getCause());
                            } catch (InterruptedException ie) {
                                interrupted = true;
                            }
                        }
                    } finally {
                        if (interrupted)
                            Strand.currentStrand().interrupt();
                    }
                } finally {
                    ilock.unlock();
                }
            }
            @Override
            @Suspendable
            public void close() throws IOException {
                ch.close();
            }
        };
    }
public static InputStream newInputStream ( final AsynchronousByteChannel ch ) throws SuspendExecution { checkNotNull ( ch , "ch" ) ; return new InputStream ( ) { private ReentrantLock ilock = new ReentrantLock ( ) ; private ByteBuffer bb = null ; private byte [ ] bs = null ; private byte [ ] b1 = null ; @ Override @ Suspendable public int read ( ) throws IOException { try { ilock . lock ( ) ; if ( b1 == null ) b1 = new byte [ 1 ] ; int n = this . read ( b1 ) ; if ( n == 1 ) return b1 [ 0 ] & 0xff ; return - 1 ; } finally { ilock . unlock ( ) ; } } @ Override @ Suspendable public int read ( byte [ ] bs , int off , int len ) throws IOException { try { ilock . lock ( ) ; if ( ( off < 0 ) || ( off > bs . length ) || ( len < 0 ) || ( ( off + len ) > bs . length ) || ( ( off + len ) < 0 ) ) { throw new IndexOutOfBoundsException ( ) ; } else if ( len == 0 ) return 0 ; ByteBuffer bb = ( ( this . bs == bs ) ? this . bb : ByteBuffer . wrap ( bs ) ) ; bb . position ( off ) ; bb . limit ( Math . min ( off + len , bb . capacity ( ) ) ) ; this . bb = bb ; this . bs = bs ; boolean interrupted = false ; try { for ( ; ; ) { try { return ch . read ( bb ) . get ( ) ; } catch ( ExecutionException ee ) { throw new IOException ( ee . getCause ( ) ) ; } catch ( InterruptedException ie ) { interrupted = true ; } } } finally { if ( interrupted ) Strand . currentStrand ( ) . interrupt ( ) ; } } finally { ilock . unlock ( ) ; } } @ Override @ Suspendable public void close ( ) throws IOException { ch . close ( ) ; } } ; }
Constructs a stream that reads bytes from the given channel.
Constructs a stream that reads bytes from the given channel .
// Invoker's previous array