1 /*
   2  * Copyright (c) 2000, 2009, Oracle and/or its affiliates. All rights reserved.
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * This code is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License version 2 only, as
   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General Public License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 
  26 package java.nio.channels;
  27 
  28 import java.io.FileInputStream;
  29 import java.io.FileOutputStream;
  30 import java.io.InputStream;
  31 import java.io.OutputStream;
  32 import java.io.Reader;
  33 import java.io.Writer;
  34 import java.io.IOException;
  35 import java.nio.ByteBuffer;
  36 import java.nio.charset.Charset;
  37 import java.nio.charset.CharsetDecoder;
  38 import java.nio.charset.CharsetEncoder;
  39 import java.nio.charset.UnsupportedCharsetException;
  40 import java.nio.channels.spi.AbstractInterruptibleChannel;
  41 import java.util.concurrent.ExecutionException;
  42 import sun.nio.ch.ChannelInputStream;
  43 import sun.nio.cs.StreamDecoder;
  44 import sun.nio.cs.StreamEncoder;
  45 
  46 
  47 /**
  48  * Utility methods for channels and streams.
  49  *
  50  * <p> This class defines static methods that support the interoperation of the
  51  * stream classes of the <tt>{@link java.io}</tt> package with the channel
  52  * classes of this package.  </p>
  53  *
  54  *
  55  * @author Mark Reinhold
  56  * @author Mike McCloskey
  57  * @author JSR-51 Expert Group
  58  * @since 1.4
  59  */
  60 
  61 public final class Channels {
  62 
  63     private Channels() { }              // No instantiation
  64 
  65     private static void checkNotNull(Object o, String name) {
  66         if (o == null)
  67             throw new NullPointerException("\"" + name + "\" is null!");
  68     }
  69 
  70     /**
  71      * Write all remaining bytes in buffer to the given channel.
  72      * If the channel is selectable then it must be configured blocking.
  73      */
  74     private static void writeFullyImpl(WritableByteChannel ch, ByteBuffer bb)
  75         throws IOException
  76     {
  77         while (bb.remaining() > 0) {
  78             int n = ch.write(bb);
  79             if (n <= 0)
  80                 throw new RuntimeException("no bytes written");
  81         }
  82     }
  83 
  84     /**
  85      * Write all remaining bytes in buffer to the given channel.
  86      *
  87      * @throws  IllegalBlockingModeException
  88      *          If the channel is selectable and configured non-blocking.
  89      */
  90     private static void writeFully(WritableByteChannel ch, ByteBuffer bb)
  91         throws IOException
  92     {
  93         if (ch instanceof SelectableChannel) {
  94             SelectableChannel sc = (SelectableChannel)ch;
  95             synchronized (sc.blockingLock()) {
  96                 if (!sc.isBlocking())
  97                     throw new IllegalBlockingModeException();
  98                 writeFullyImpl(ch, bb);
  99             }
 100         } else {
 101             writeFullyImpl(ch, bb);
 102         }
 103     }
 104 
 105     // -- Byte streams from channels --
 106 
 107     /**
 108      * Constructs a stream that reads bytes from the given channel.
 109      *
 110      * <p> The <tt>read</tt> methods of the resulting stream will throw an
 111      * {@link IllegalBlockingModeException} if invoked while the underlying
 112      * channel is in non-blocking mode.  The stream will not be buffered, and
 113      * it will not support the {@link InputStream#mark mark} or {@link
 114      * InputStream#reset reset} methods.  The stream will be safe for access by
 115      * multiple concurrent threads.  Closing the stream will in turn cause the
 116      * channel to be closed.  </p>
 117      *
 118      * @param  ch
 119      *         The channel from which bytes will be read
 120      *
 121      * @return  A new input stream
 122      */
 123     public static InputStream newInputStream(ReadableByteChannel ch) {
 124         checkNotNull(ch, "ch");
 125         return new sun.nio.ch.ChannelInputStream(ch);
 126     }
 127 
 128     /**
 129      * Constructs a stream that writes bytes to the given channel.
 130      *
 131      * <p> The <tt>write</tt> methods of the resulting stream will throw an
 132      * {@link IllegalBlockingModeException} if invoked while the underlying
 133      * channel is in non-blocking mode.  The stream will not be buffered.  The
 134      * stream will be safe for access by multiple concurrent threads.  Closing
 135      * the stream will in turn cause the channel to be closed.  </p>
 136      *
 137      * @param  ch
 138      *         The channel to which bytes will be written
 139      *
 140      * @return  A new output stream
 141      */
 142     public static OutputStream newOutputStream(final WritableByteChannel ch) {
 143         checkNotNull(ch, "ch");
 144 
 145         return new OutputStream() {
 146 
 147                 private ByteBuffer bb = null;
 148                 private byte[] bs = null;       // Invoker's previous array
 149                 private byte[] b1 = null;
 150 
 151                 public synchronized void write(int b) throws IOException {
 152                    if (b1 == null)
 153                         b1 = new byte[1];
 154                     b1[0] = (byte)b;
 155                     this.write(b1);
 156                 }
 157 
 158                 public synchronized void write(byte[] bs, int off, int len)
 159                     throws IOException
 160                 {
 161                     if ((off < 0) || (off > bs.length) || (len < 0) ||
 162                         ((off + len) > bs.length) || ((off + len) < 0)) {
 163                         throw new IndexOutOfBoundsException();
 164                     } else if (len == 0) {
 165                         return;
 166                     }
 167                     ByteBuffer bb = ((this.bs == bs)
 168                                      ? this.bb
 169                                      : ByteBuffer.wrap(bs));
 170                     bb.limit(Math.min(off + len, bb.capacity()));
 171                     bb.position(off);
 172                     this.bb = bb;
 173                     this.bs = bs;
 174                     Channels.writeFully(ch, bb);
 175                 }
 176 
 177                 public void close() throws IOException {
 178                     ch.close();
 179                 }
 180 
 181             };
 182     }
 183 
 184     /**
 185      * Constructs a stream that reads bytes from the given channel.
 186      *
 187      * <p> The stream will not be buffered, and it will not support the {@link
 188      * InputStream#mark mark} or {@link InputStream#reset reset} methods.  The
 189      * stream will be safe for access by multiple concurrent threads.  Closing
 190      * the stream will in turn cause the channel to be closed.  </p>
 191      *
 192      * @param  ch
 193      *         The channel from which bytes will be read
 194      *
 195      * @return  A new input stream
 196      *
 197      * @since 1.7
 198      */
 199     public static InputStream newInputStream(final AsynchronousByteChannel ch) {
 200         checkNotNull(ch, "ch");
 201         return new InputStream() {
 202 
 203             private ByteBuffer bb = null;
 204             private byte[] bs = null;           // Invoker's previous array
 205             private byte[] b1 = null;
 206 
 207             @Override
 208             public synchronized int read() throws IOException {
 209                 if (b1 == null)
 210                     b1 = new byte[1];
 211                 int n = this.read(b1);
 212                 if (n == 1)
 213                     return b1[0] & 0xff;
 214                 return -1;
 215             }
 216 
 217             @Override
 218             public synchronized int read(byte[] bs, int off, int len)
 219                 throws IOException
 220             {
 221                 if ((off < 0) || (off > bs.length) || (len < 0) ||
 222                     ((off + len) > bs.length) || ((off + len) < 0)) {
 223                     throw new IndexOutOfBoundsException();
 224                 } else if (len == 0)
 225                     return 0;
 226 
 227                 ByteBuffer bb = ((this.bs == bs)
 228                                  ? this.bb
 229                                  : ByteBuffer.wrap(bs));
 230                 bb.position(off);
 231                 bb.limit(Math.min(off + len, bb.capacity()));
 232                 this.bb = bb;
 233                 this.bs = bs;
 234 
 235                 boolean interrupted = false;
 236                 try {
 237                     for (;;) {
 238                         try {
 239                             return ch.read(bb).get();
 240                         } catch (ExecutionException ee) {
 241                             throw new IOException(ee.getCause());
 242                         } catch (InterruptedException ie) {
 243                             interrupted = true;
 244                         }
 245                     }
 246                 } finally {
 247                     if (interrupted)
 248                         Thread.currentThread().interrupt();
 249                 }
 250             }
 251 
 252             @Override
 253             public void close() throws IOException {
 254                 ch.close();
 255             }
 256         };
 257     }
 258 
 259     /**
 260      * Constructs a stream that writes bytes to the given channel.
 261      *
 262      * <p> The stream will not be buffered. The stream will be safe for access
 263      * by multiple concurrent threads.  Closing the stream will in turn cause
 264      * the channel to be closed.  </p>
 265      *
 266      * @param  ch
 267      *         The channel to which bytes will be written
 268      *
 269      * @return  A new output stream
 270      *
 271      * @since 1.7
 272      */
 273     public static OutputStream newOutputStream(final AsynchronousByteChannel ch) {
 274         checkNotNull(ch, "ch");
 275         return new OutputStream() {
 276 
 277             private ByteBuffer bb = null;
 278             private byte[] bs = null;   // Invoker's previous array
 279             private byte[] b1 = null;
 280 
 281             @Override
 282             public synchronized void write(int b) throws IOException {
 283                if (b1 == null)
 284                     b1 = new byte[1];
 285                 b1[0] = (byte)b;
 286                 this.write(b1);
 287             }
 288 
 289             @Override
 290             public synchronized void write(byte[] bs, int off, int len)
 291                 throws IOException
 292             {
 293                 if ((off < 0) || (off > bs.length) || (len < 0) ||
 294                     ((off + len) > bs.length) || ((off + len) < 0)) {
 295                     throw new IndexOutOfBoundsException();
 296                 } else if (len == 0) {
 297                     return;
 298                 }
 299                 ByteBuffer bb = ((this.bs == bs)
 300                                  ? this.bb
 301                                  : ByteBuffer.wrap(bs));
 302                 bb.limit(Math.min(off + len, bb.capacity()));
 303                 bb.position(off);
 304                 this.bb = bb;
 305                 this.bs = bs;
 306 
 307                 boolean interrupted = false;
 308                 try {
 309                     while (bb.remaining() > 0) {
 310                         try {
 311                             ch.write(bb).get();
 312                         } catch (ExecutionException ee) {
 313                             throw new IOException(ee.getCause());
 314                         } catch (InterruptedException ie) {
 315                             interrupted = true;
 316                         }
 317                     }
 318                 } finally {
 319                     if (interrupted)
 320                         Thread.currentThread().interrupt();
 321                 }
 322             }
 323 
 324             @Override
 325             public void close() throws IOException {
 326                 ch.close();
 327             }
 328         };
 329     }
 330 
 331 
 332     // -- Channels from streams --
 333 
 334     /**
 335      * Constructs a channel that reads bytes from the given stream.
 336      *
 337      * <p> The resulting channel will not be buffered; it will simply redirect
 338      * its I/O operations to the given stream.  Closing the channel will in
 339      * turn cause the stream to be closed.  </p>
 340      *
 341      * @param  in
 342      *         The stream from which bytes are to be read
 343      *
 344      * @return  A new readable byte channel
 345      */
 346     public static ReadableByteChannel newChannel(final InputStream in) {
 347         checkNotNull(in, "in");
 348 
 349         if (in instanceof FileInputStream &&
 350             FileInputStream.class.equals(in.getClass())) {
 351             return ((FileInputStream)in).getChannel();
 352         }
 353 
 354         return new ReadableByteChannelImpl(in);
 355     }
 356 
 357     private static class ReadableByteChannelImpl
 358         extends AbstractInterruptibleChannel    // Not really interruptible
 359         implements ReadableByteChannel
 360     {
 361         InputStream in;
 362         private static final int TRANSFER_SIZE = 8192;
 363         private byte buf[] = new byte[0];
 364         private boolean open = true;
 365         private Object readLock = new Object();
 366 
 367         ReadableByteChannelImpl(InputStream in) {
 368             this.in = in;
 369         }
 370 
 371         public int read(ByteBuffer dst) throws IOException {
 372             int len = dst.remaining();
 373             int totalRead = 0;
 374             int bytesRead = 0;
 375             synchronized (readLock) {
 376                 while (totalRead < len) {
 377                     int bytesToRead = Math.min((len - totalRead),
 378                                                TRANSFER_SIZE);
 379                     if (buf.length < bytesToRead)
 380                         buf = new byte[bytesToRead];
 381                     if ((totalRead > 0) && !(in.available() > 0))
 382                         break; // block at most once
 383                     try {
 384                         begin();
 385                         bytesRead = in.read(buf, 0, bytesToRead);
 386                     } finally {
 387                         end(bytesRead > 0);
 388                     }
 389                     if (bytesRead < 0)
 390                         break;
 391                     else
 392                         totalRead += bytesRead;
 393                     dst.put(buf, 0, bytesRead);
 394                 }
 395                 if ((bytesRead < 0) && (totalRead == 0))
 396                     return -1;
 397 
 398                 return totalRead;
 399             }
 400         }
 401 
 402         protected void implCloseChannel() throws IOException {
 403             in.close();
 404             open = false;
 405         }
 406     }
 407 
 408 
 409     /**
 410      * Constructs a channel that writes bytes to the given stream.
 411      *
 412      * <p> The resulting channel will not be buffered; it will simply redirect
 413      * its I/O operations to the given stream.  Closing the channel will in
 414      * turn cause the stream to be closed.  </p>
 415      *
 416      * @param  out
 417      *         The stream to which bytes are to be written
 418      *
 419      * @return  A new writable byte channel
 420      */
 421     public static WritableByteChannel newChannel(final OutputStream out) {
 422         checkNotNull(out, "out");
 423 
 424         if (out instanceof FileOutputStream &&
 425             FileOutputStream.class.equals(out.getClass())) {
 426                 return ((FileOutputStream)out).getChannel();
 427         }
 428 
 429         return new WritableByteChannelImpl(out);
 430     }
 431 
 432     private static class WritableByteChannelImpl
 433         extends AbstractInterruptibleChannel    // Not really interruptible
 434         implements WritableByteChannel
 435     {
 436         OutputStream out;
 437         private static final int TRANSFER_SIZE = 8192;
 438         private byte buf[] = new byte[0];
 439         private boolean open = true;
 440         private Object writeLock = new Object();
 441 
 442         WritableByteChannelImpl(OutputStream out) {
 443             this.out = out;
 444         }
 445 
 446         public int write(ByteBuffer src) throws IOException {
 447             int len = src.remaining();
 448             int totalWritten = 0;
 449             synchronized (writeLock) {
 450                 while (totalWritten < len) {
 451                     int bytesToWrite = Math.min((len - totalWritten),
 452                                                 TRANSFER_SIZE);
 453                     if (buf.length < bytesToWrite)
 454                         buf = new byte[bytesToWrite];
 455                     src.get(buf, 0, bytesToWrite);
 456                     try {
 457                         begin();
 458                         out.write(buf, 0, bytesToWrite);
 459                     } finally {
 460                         end(bytesToWrite > 0);
 461                     }
 462                     totalWritten += bytesToWrite;
 463                 }
 464                 return totalWritten;
 465             }
 466         }
 467 
 468         protected void implCloseChannel() throws IOException {
 469             out.close();
 470             open = false;
 471         }
 472     }
 473 
 474 
 475     // -- Character streams from channels --
 476 
 477     /**
 478      * Constructs a reader that decodes bytes from the given channel using the
 479      * given decoder.
 480      *
 481      * <p> The resulting stream will contain an internal input buffer of at
 482      * least <tt>minBufferCap</tt> bytes.  The stream's <tt>read</tt> methods
 483      * will, as needed, fill the buffer by reading bytes from the underlying
 484      * channel; if the channel is in non-blocking mode when bytes are to be
 485      * read then an {@link IllegalBlockingModeException} will be thrown.  The
 486      * resulting stream will not otherwise be buffered, and it will not support
 487      * the {@link Reader#mark mark} or {@link Reader#reset reset} methods.
 488      * Closing the stream will in turn cause the channel to be closed.  </p>
 489      *
 490      * @param  ch
 491      *         The channel from which bytes will be read
 492      *
 493      * @param  dec
 494      *         The charset decoder to be used
 495      *
 496      * @param  minBufferCap
 497      *         The minimum capacity of the internal byte buffer,
 498      *         or <tt>-1</tt> if an implementation-dependent
 499      *         default capacity is to be used
 500      *
 501      * @return  A new reader
 502      */
 503     public static Reader newReader(ReadableByteChannel ch,
 504                                    CharsetDecoder dec,
 505                                    int minBufferCap)
 506     {
 507         checkNotNull(ch, "ch");
 508         return StreamDecoder.forDecoder(ch, dec.reset(), minBufferCap);
 509     }
 510 
 511     /**
 512      * Constructs a reader that decodes bytes from the given channel according
 513      * to the named charset.
 514      *
 515      * <p> An invocation of this method of the form
 516      *
 517      * <blockquote><pre>
 518      * Channels.newReader(ch, csname)</pre></blockquote>
 519      *
 520      * behaves in exactly the same way as the expression
 521      *
 522      * <blockquote><pre>
 523      * Channels.newReader(ch,
 524      *                    Charset.forName(csName)
 525      *                        .newDecoder(),
 526      *                    -1);</pre></blockquote>
 527      *
 528      * @param  ch
 529      *         The channel from which bytes will be read
 530      *
 531      * @param  csName
 532      *         The name of the charset to be used
 533      *
 534      * @return  A new reader
 535      *
 536      * @throws  UnsupportedCharsetException
 537      *          If no support for the named charset is available
 538      *          in this instance of the Java virtual machine
 539      */
 540     public static Reader newReader(ReadableByteChannel ch,
 541                                    String csName)
 542     {
 543         checkNotNull(csName, "csName");
 544         return newReader(ch, Charset.forName(csName).newDecoder(), -1);
 545     }
 546 
 547     /**
 548      * Constructs a writer that encodes characters using the given encoder and
 549      * writes the resulting bytes to the given channel.
 550      *
 551      * <p> The resulting stream will contain an internal output buffer of at
 552      * least <tt>minBufferCap</tt> bytes.  The stream's <tt>write</tt> methods
 553      * will, as needed, flush the buffer by writing bytes to the underlying
 554      * channel; if the channel is in non-blocking mode when bytes are to be
 555      * written then an {@link IllegalBlockingModeException} will be thrown.
 556      * The resulting stream will not otherwise be buffered.  Closing the stream
 557      * will in turn cause the channel to be closed.  </p>
 558      *
 559      * @param  ch
 560      *         The channel to which bytes will be written
 561      *
 562      * @param  enc
 563      *         The charset encoder to be used
 564      *
 565      * @param  minBufferCap
 566      *         The minimum capacity of the internal byte buffer,
 567      *         or <tt>-1</tt> if an implementation-dependent
 568      *         default capacity is to be used
 569      *
 570      * @return  A new writer
 571      */
 572     public static Writer newWriter(final WritableByteChannel ch,
 573                                    final CharsetEncoder enc,
 574                                    final int minBufferCap)
 575     {
 576         checkNotNull(ch, "ch");
 577         return StreamEncoder.forEncoder(ch, enc.reset(), minBufferCap);
 578     }
 579 
 580     /**
 581      * Constructs a writer that encodes characters according to the named
 582      * charset and writes the resulting bytes to the given channel.
 583      *
 584      * <p> An invocation of this method of the form
 585      *
 586      * <blockquote><pre>
 587      * Channels.newWriter(ch, csname)</pre></blockquote>
 588      *
 589      * behaves in exactly the same way as the expression
 590      *
 591      * <blockquote><pre>
 592      * Channels.newWriter(ch,
 593      *                    Charset.forName(csName)
 594      *                        .newEncoder(),
 595      *                    -1);</pre></blockquote>
 596      *
 597      * @param  ch
 598      *         The channel to which bytes will be written
 599      *
 600      * @param  csName
 601      *         The name of the charset to be used
 602      *
 603      * @return  A new writer
 604      *
 605      * @throws  UnsupportedCharsetException
 606      *          If no support for the named charset is available
 607      *          in this instance of the Java virtual machine
 608      */
 609     public static Writer newWriter(WritableByteChannel ch,
 610                                    String csName)
 611     {
 612         checkNotNull(csName, "csName");
 613         return newWriter(ch, Charset.forName(csName).newEncoder(), -1);
 614     }
 615 }