src/main/java/net/batchik/crdt/fiber/FiberChannels.java
/**
     * Constructs a stream that writes bytes to the given channel.
     *
     * <p> The stream will not be buffered. 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 to which bytes will be written
     *
     * @return  A new output stream
     *
     * @since 1.7
     */
Constructs a stream that writes bytes to the given channel.
 The stream will not be buffered. 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 to which bytes will be written

@return  A new output stream


Constructs a stream that writes bytes to the given channel . The stream will not be buffered . 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 to which bytes will be written @return A new output stream
public static OutputStream newOutputStream(final AsynchronousByteChannel ch) throws SuspendExecution {
        checkNotNull(ch, "ch");
        return new OutputStream() {
            private ReentrantLock olock = new ReentrantLock();
            private ByteBuffer bb = null;
            private byte[] bs = null;   
            private byte[] b1 = null;
            @Override
            @Suspendable
            public void write(int b) throws IOException {
                try {
                    olock.lock();
                    if (b1 == null)
                        b1 = new byte[1];
                    b1[0] = (byte) b;
                    this.write(b1);
                } finally {
                    olock.unlock();
                }
            }
            @Override
            @Suspendable
            public void write(byte[] bs, int off, int len)
                    throws IOException
            {
                try {
                    olock.lock();
                    if ((off < 0) || (off > bs.length) || (len < 0) ||
                            ((off + len) > bs.length) || ((off + len) < 0)) {
                        throw new IndexOutOfBoundsException();
                    } else if (len == 0) {
                        return;
                    }
                    ByteBuffer bb = ((this.bs == bs)
                            ? this.bb
                            : ByteBuffer.wrap(bs));
                    bb.limit(Math.min(off + len, bb.capacity()));
                    bb.position(off);
                    this.bb = bb;
                    this.bs = bs;
                    boolean interrupted = false;
                    try {
                        while (bb.remaining() > 0) {
                            try {
                                ch.write(bb).get();
                            } catch (ExecutionException ee) {
                                throw new IOException(ee.getCause());
                            } catch (InterruptedException ie) {
                                interrupted = true;
                            }
                        }
                    } finally {
                        if (interrupted)
                            Thread.currentThread().interrupt();
                    }
                } finally {
                    olock.unlock();
                }
            }
            @Override
            @Suspendable
            public void close() throws IOException {
                ch.close();
            }
        };
    }
public static OutputStream newOutputStream ( final AsynchronousByteChannel ch ) throws SuspendExecution { checkNotNull ( ch , "ch" ) ; return new OutputStream ( ) { private ReentrantLock olock = new ReentrantLock ( ) ; private ByteBuffer bb = null ; private byte [ ] bs = null ; private byte [ ] b1 = null ; @ Override @ Suspendable public void write ( int b ) throws IOException { try { olock . lock ( ) ; if ( b1 == null ) b1 = new byte [ 1 ] ; b1 [ 0 ] = ( byte ) b ; this . write ( b1 ) ; } finally { olock . unlock ( ) ; } } @ Override @ Suspendable public void write ( byte [ ] bs , int off , int len ) throws IOException { try { olock . lock ( ) ; if ( ( off < 0 ) || ( off > bs . length ) || ( len < 0 ) || ( ( off + len ) > bs . length ) || ( ( off + len ) < 0 ) ) { throw new IndexOutOfBoundsException ( ) ; } else if ( len == 0 ) { return ; } ByteBuffer bb = ( ( this . bs == bs ) ? this . bb : ByteBuffer . wrap ( bs ) ) ; bb . limit ( Math . min ( off + len , bb . capacity ( ) ) ) ; bb . position ( off ) ; this . bb = bb ; this . bs = bs ; boolean interrupted = false ; try { while ( bb . remaining ( ) > 0 ) { try { ch . write ( bb ) . get ( ) ; } catch ( ExecutionException ee ) { throw new IOException ( ee . getCause ( ) ) ; } catch ( InterruptedException ie ) { interrupted = true ; } } } finally { if ( interrupted ) Thread . currentThread ( ) . interrupt ( ) ; } } finally { olock . unlock ( ) ; } } @ Override @ Suspendable public void close ( ) throws IOException { ch . close ( ) ; } } ; }
Constructs a stream that writes bytes to the given channel.
Constructs a stream that writes bytes to the given channel .
// Invoker's previous array