--- 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.
+ *
+ * <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
+ */
+ 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.
+ *
+ * <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
+ */
+ 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);
}
-
}