src/jexer/io/TimeoutInputStream.java

   1 /*
   2  * Jexer - Java Text User Interface
   3  *
   4  * The MIT License (MIT)
   5  *
   6  * Copyright (C) 2017 Kevin Lamonte
   7  *
   8  * Permission is hereby granted, free of charge, to any person obtaining a
   9  * copy of this software and associated documentation files (the "Software"),
  10  * to deal in the Software without restriction, including without limitation
  11  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
  12  * and/or sell copies of the Software, and to permit persons to whom the
  13  * Software is furnished to do so, subject to the following conditions:
  14  *
  15  * The above copyright notice and this permission notice shall be included in
  16  * all copies or substantial portions of the Software.
  17  *
  18  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  19  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  20  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  21  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  22  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  23  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  24  * DEALINGS IN THE SOFTWARE.
  25  *
  26  * @author Kevin Lamonte [kevin.lamonte@gmail.com]
  27  * @version 1
  28  */
  29 package jexer.io;
  30
  31 import java.io.IOException;
  32 import java.io.InputStream;
  33
  34 /**
  35  * This class provides an optional millisecond timeout on its read()
  36  * operations.  This permits callers to bail out rather than block.
  37  */
  38 public class TimeoutInputStream extends InputStream {
  39
  40     // ------------------------------------------------------------------------
  41     // Variables --------------------------------------------------------------
  42     // ------------------------------------------------------------------------
  43
  44     /**
  45      * The wrapped stream.
  46      */
  47     private InputStream stream;
  48
  49     /**
  50      * The timeout value in millis.  If it takes longer than this for bytes
  51      * to be available for read then a ReadTimeoutException is thrown.  A
  52      * value of 0 means to block as a normal InputStream would.
  53      */
  54     private int timeoutMillis;
  55
  56     /**
  57      * If true, the current read() will timeout soon.
  58      */
  59     private volatile boolean cancel = false;
  60
  61     // ------------------------------------------------------------------------
  62     // Constructors -----------------------------------------------------------
  63     // ------------------------------------------------------------------------
  64
  65     /**
  66      * Public constructor, at the default timeout of 10000 millis (10
  67      * seconds).
  68      *
  69      * @param stream the wrapped InputStream
  70      */
  71     public TimeoutInputStream(final InputStream stream) {
  72         this.stream             = stream;
  73         this.timeoutMillis      = 10000;
  74     }
  75
  76     /**
  77      * Public constructor.
  78      *
  79      * @param stream the wrapped InputStream
  80      * @param timeoutMillis the timeout value in millis.  If it takes longer
  81      * than this for bytes to be available for read then a
  82      * ReadTimeoutException is thrown.  A value of 0 means to block as a
  83      * normal InputStream would.
  84      */
  85     public TimeoutInputStream(final InputStream stream,
  86         final int timeoutMillis) {
  87
  88         if (timeoutMillis < 0) {
  89             throw new IllegalArgumentException("Invalid timeoutMillis value, " +
  90                 "must be >= 0");
  91         }
  92
  93         this.stream             = stream;
  94         this.timeoutMillis      = timeoutMillis;
  95     }
  96
  97     // ------------------------------------------------------------------------
  98     // InputStream ------------------------------------------------------------
  99     // ------------------------------------------------------------------------
 100
 101     /**
 102      * Reads the next byte of data from the input stream.
 103      *
 104      * @return the next byte of data, or -1 if there is no more data because
 105      * the end of the stream has been reached.
 106      * @throws IOException if an I/O error occurs
 107      */
 108     @Override
 109     public int read() throws IOException {
 110
 111         if (timeoutMillis == 0) {
 112             // Block on the read().
 113             return stream.read();
 114         }
 115
 116         if (stream.available() > 0) {
 117             // A byte is available now, return it.
 118             return stream.read();
 119         }
 120
 121         // We will wait up to timeoutMillis to see if a byte is available.
 122         // If not, we throw ReadTimeoutException.
 123         long checkTime = System.currentTimeMillis();
 124         while (stream.available() == 0) {
 125             long now = System.currentTimeMillis();
 126             synchronized (this) {
 127                 if ((now - checkTime > timeoutMillis) || (cancel == true)) {
 128                     if (cancel == true) {
 129                         cancel = false;
 130                     }
 131                     throw new ReadTimeoutException("Timeout on read(): " +
 132                         (int) (now - checkTime) + " millis and still no data");
 133                 }
 134             }
 135             try {
 136                 // How long do we sleep for, eh?  For now we will go with 2
 137                 // millis.
 138                 Thread.sleep(2);
 139             } catch (InterruptedException e) {
 140                 // SQUASH
 141             }
 142         }
 143
 144         if (stream.available() > 0) {
 145             // A byte is available now, return it.
 146             return stream.read();
 147         }
 148
 149         throw new IOException("InputStream claimed a byte was available, but " +
 150             "now it is not.  What is going on?");
 151     }
 152
 153     /**
 154      * Reads some number of bytes from the input stream and stores them into
 155      * the buffer array b.
 156      *
 157      * @param b the buffer into which the data is read.
 158      * @return the total number of bytes read into the buffer, or -1 if there
 159      * is no more data because the end of the stream has been reached.
 160      * @throws IOException if an I/O error occurs
 161      */
 162     @Override
 163     public int read(final byte[] b) throws IOException {
 164         if (timeoutMillis == 0) {
 165             // Block on the read().
 166             return stream.read(b);
 167         }
 168
 169         int remaining = b.length;
 170
 171         if (stream.available() >= remaining) {
 172             // Enough bytes are available now, return them.
 173             return stream.read(b);
 174         }
 175
 176         while (remaining > 0) {
 177
 178             // We will wait up to timeoutMillis to see if a byte is
 179             // available.  If not, we throw ReadTimeoutException.
 180             long checkTime = System.currentTimeMillis();
 181             while (stream.available() == 0) {
 182                 long now = System.currentTimeMillis();
 183
 184                 synchronized (this) {
 185                     if ((now - checkTime > timeoutMillis) || (cancel == true)) {
 186                         if (cancel == true) {
 187                             cancel = false;
 188                         }
 189                         throw new ReadTimeoutException("Timeout on read(): " +
 190                             (int) (now - checkTime) + " millis and still no " +
 191                             "data");
 192                     }
 193                 }
 194                 try {
 195                     // How long do we sleep for, eh?  For now we will go with
 196                     // 2 millis.
 197                     Thread.sleep(2);
 198                 } catch (InterruptedException e) {
 199                     // SQUASH
 200                 }
 201             }
 202
 203             if (stream.available() > 0) {
 204                 // At least one byte is available now, read it.
 205                 int n = stream.available();
 206                 if (remaining < n) {
 207                     n = remaining;
 208                 }
 209                 int rc = stream.read(b, b.length - remaining, n);
 210                 if (rc == -1) {
 211                     // This shouldn't happen.
 212                     throw new IOException("InputStream claimed bytes were " +
 213                         "available, but read() returned -1.  What is going " +
 214                         "on?");
 215                 }
 216                 remaining -= rc;
 217                 if (remaining == 0) {
 218                     return b.length;
 219                 }
 220             }
 221         }
 222
 223         throw new IOException("InputStream claimed all bytes were available, " +
 224             "but now it is not.  What is going on?");
 225     }
 226
 227     /**
 228      * Reads up to len bytes of data from the input stream into an array of
 229      * bytes.
 230      *
 231      * @param b the buffer into which the data is read.
 232      * @param off the start offset in array b at which the data is written.
 233      * @param len the maximum number of bytes to read.
 234      * @return the total number of bytes read into the buffer, or -1 if there
 235      * is no more data because the end of the stream has been reached.
 236      * @throws IOException if an I/O error occurs
 237      */
 238     @Override
 239     public int read(final byte[] b, final int off,
 240         final int len) throws IOException {
 241
 242         if (timeoutMillis == 0) {
 243             // Block on the read().
 244             return stream.read(b);
 245         }
 246
 247         int remaining = len;
 248
 249         if (stream.available() >= remaining) {
 250             // Enough bytes are available now, return them.
 251             return stream.read(b, off, remaining);
 252         }
 253
 254         while (remaining > 0) {
 255
 256             // We will wait up to timeoutMillis to see if a byte is
 257             // available.  If not, we throw ReadTimeoutException.
 258             long checkTime = System.currentTimeMillis();
 259             while (stream.available() == 0) {
 260                 long now = System.currentTimeMillis();
 261                 synchronized (this) {
 262                     if ((now - checkTime > timeoutMillis) || (cancel == true)) {
 263                         if (cancel == true) {
 264                             cancel = false;
 265                         }
 266                         throw new ReadTimeoutException("Timeout on read(): " +
 267                             (int) (now - checkTime) + " millis and still no " +
 268                             "data");
 269                     }
 270                 }
 271                 try {
 272                     // How long do we sleep for, eh?  For now we will go with
 273                     // 2 millis.
 274                     Thread.sleep(2);
 275                 } catch (InterruptedException e) {
 276                     // SQUASH
 277                 }
 278             }
 279
 280             if (stream.available() > 0) {
 281                 // At least one byte is available now, read it.
 282                 int n = stream.available();
 283                 if (remaining < n) {
 284                     n = remaining;
 285                 }
 286                 int rc = stream.read(b, off + len - remaining, n);
 287                 if (rc == -1) {
 288                     // This shouldn't happen.
 289                     throw new IOException("InputStream claimed bytes were " +
 290                         "available, but read() returned -1.  What is going " +
 291                         "on?");
 292                 }
 293                 remaining -= rc;
 294                 if (remaining == 0) {
 295                     return len;
 296                 }
 297             }
 298         }
 299
 300         throw new IOException("InputStream claimed all bytes were available, " +
 301             "but now it is not.  What is going on?");
 302     }
 303
 304     /**
 305      * Returns an estimate of the number of bytes that can be read (or
 306      * skipped over) from this input stream without blocking by the next
 307      * invocation of a method for this input stream.
 308      *
 309      * @return an estimate of the number of bytes that can be read (or
 310      * skipped over) from this input stream without blocking or 0 when it
 311      * reaches the end of the input stream.
 312      * @throws IOException if an I/O error occurs
 313      */
 314     @Override
 315     public int available() throws IOException {
 316         return stream.available();
 317     }
 318
 319     /**
 320      * Closes this input stream and releases any system resources associated
 321      * with the stream.
 322      *
 323      * @throws IOException if an I/O error occurs
 324      */
 325     @Override
 326     public void close() throws IOException {
 327         stream.close();
 328     }
 329
 330     /**
 331      * Marks the current position in this input stream.
 332      *
 333      * @param readLimit the maximum limit of bytes that can be read before
 334      * the mark position becomes invalid
 335      */
 336     @Override
 337     public void mark(final int readLimit) {
 338         stream.mark(readLimit);
 339     }
 340
 341     /**
 342      * Tests if this input stream supports the mark and reset methods.
 343      *
 344      * @return true if this stream instance supports the mark and reset
 345      * methods; false otherwise
 346      */
 347     @Override
 348     public boolean markSupported() {
 349         return stream.markSupported();
 350     }
 351
 352     /**
 353      * Repositions this stream to the position at the time the mark method
 354      * was last called on this input stream.
 355      *
 356      * @throws IOException if an I/O error occurs
 357      */
 358     @Override
 359     public void reset() throws IOException {
 360         stream.reset();
 361     }
 362
 363     /**
 364      * Skips over and discards n bytes of data from this input stream.
 365      *
 366      * @param n the number of bytes to be skipped
 367      * @return the actual number of bytes skipped
 368      * @throws IOException if an I/O error occurs
 369      */
 370     @Override
 371     public long skip(final long n) throws IOException {
 372         return stream.skip(n);
 373     }
 374
 375     // ------------------------------------------------------------------------
 376     // TimeoutInputStream -----------------------------------------------------
 377     // ------------------------------------------------------------------------
 378
 379     /**
 380      * Request that the current read() operation timeout immediately.
 381      */
 382     public synchronized void cancelRead() {
 383         cancel = true;
 384     }
 385
 386 }