src/main/java/net/batchik/crdt/fiber/FiberChannels.java
/**
     * Constructs a stream that writes bytes to the given channel.
     *
     * <p> The <tt>write</tt> methods of the resulting stream will throw an
     * {@link IllegalBlockingModeException} if invoked while the underlying
     * channel is in non-blocking mode.  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
     */
Constructs a stream that writes bytes to the given channel.
 The write methods of the resulting stream will throw an
IllegalBlockingModeException if invoked while the underlying
channel is in non-blocking mode.  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 write methods of the resulting stream will throw an IllegalBlockingModeException if invoked while the underlying channel is in non - blocking mode . 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 WritableByteChannel 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;
            @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();
                }
            }
            @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;
                    try {
                        FiberChannels.writeFully(ch, bb);
                    } catch (InterruptedException | ExecutionException | SuspendExecution e) {
                        e.printStackTrace();
                    }
                } finally {
                    olock.unlock();
                }
            }
            @Suspendable
            public void close() throws IOException {
                ch.close();
            }
        };
    }
public static OutputStream newOutputStream ( final WritableByteChannel 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 ; @ 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 ( ) ; } } @ 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 ; try { FiberChannels . writeFully ( ch , bb ) ; } catch ( InterruptedException | ExecutionException | SuspendExecution e ) { e . printStackTrace ( ) ; } } finally { olock . unlock ( ) ; } } @ 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