diff -r 115e09b7a004 -r 3acf8e5e2ca0 jdk/src/share/classes/java/nio/channels/Channels.java --- a/jdk/src/share/classes/java/nio/channels/Channels.java Wed Feb 11 13:16:53 2009 +0000 +++ b/jdk/src/share/classes/java/nio/channels/Channels.java Sun Feb 15 12:25:54 2009 +0000 @@ -1,5 +1,5 @@ /* - * Copyright 2000-2008 Sun Microsystems, Inc. All Rights Reserved. + * Copyright 2000-2009 Sun Microsystems, Inc. All Rights Reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it @@ -33,15 +33,12 @@ import java.io.Writer; import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.CharBuffer; -import java.nio.BufferOverflowException; -import java.nio.BufferUnderflowException; import java.nio.charset.Charset; import java.nio.charset.CharsetDecoder; import java.nio.charset.CharsetEncoder; -import java.nio.charset.CoderResult; import java.nio.charset.UnsupportedCharsetException; import java.nio.channels.spi.AbstractInterruptibleChannel; +import java.util.concurrent.ExecutionException; import sun.nio.ch.ChannelInputStream; import sun.nio.cs.StreamDecoder; import sun.nio.cs.StreamEncoder; @@ -184,6 +181,155 @@ }; } + /** + * {@note new} + * 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 {@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.

+ * + * @param ch + * The channel from which bytes will be read + * + * @return A new input stream + * + * @since 1.7 + */ + public static InputStream newInputStream(final AsynchronousByteChannel ch) { + checkNotNull(ch, "ch"); + return new InputStream() { + + private ByteBuffer bb = null; + private byte[] bs = null; // Invoker's previous array + private byte[] b1 = null; + + @Override + public synchronized int read() throws IOException { + if (b1 == null) + b1 = new byte[1]; + int n = this.read(b1); + if (n == 1) + return b1[0] & 0xff; + return -1; + } + + @Override + public synchronized int read(byte[] bs, int off, int len) + throws IOException + { + 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) + Thread.currentThread().interrupt(); + } + } + + @Override + public void close() throws IOException { + ch.close(); + } + }; + } + + /** + * {@note new} + * 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 + * + * @since 1.7 + */ + public static OutputStream newOutputStream(final AsynchronousByteChannel ch) { + checkNotNull(ch, "ch"); + return new OutputStream() { + + private ByteBuffer bb = null; + private byte[] bs = null; // Invoker's previous array + private byte[] b1 = null; + + @Override + public synchronized void write(int b) throws IOException { + if (b1 == null) + b1 = new byte[1]; + b1[0] = (byte)b; + this.write(b1); + } + + @Override + public synchronized void write(byte[] bs, int off, int len) + throws IOException + { + 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(); + } + } + + @Override + public void close() throws IOException { + ch.close(); + } + }; + } + // -- Channels from streams -- @@ -468,5 +614,4 @@ checkNotNull(csName, "csName"); return newWriter(ch, Charset.forName(csName).newEncoder(), -1); } - }