src/jexer/io/TimeoutInputStream.java

   1 /*
   2  * Jexer - Java Text User Interface
   3  *
   4  * The MIT License (MIT)
   5  *
   6  * Copyright (C) 2019 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                 if (remaining > 0) {
 183                     return (b.length - remaining);
 184                 }
 185
 186                 long now = System.currentTimeMillis();
 187                 synchronized (this) {
 188                     if ((now - checkTime > timeoutMillis) || (cancel == true)) {
 189                         if (cancel == true) {
 190                             cancel = false;
 191                         }
 192                         throw new ReadTimeoutException("Timeout on read(): " +
 193                             (int) (now - checkTime) + " millis and still no " +
 194                             "data");
 195                     }
 196                 }
 197                 try {
 198                     // How long do we sleep for, eh?  For now we will go with
 199                     // 2 millis.
 200                     Thread.sleep(2);
 201                 } catch (InterruptedException e) {
 202                     // SQUASH
 203                 }
 204             }
 205
 206             if (stream.available() > 0) {
 207                 // At least one byte is available now, read it.
 208                 int n = stream.available();
 209                 if (remaining < n) {
 210                     n = remaining;
 211                 }
 212                 int rc = stream.read(b, b.length - remaining, n);
 213                 if (rc == -1) {
 214                     // This shouldn't happen.
 215                     throw new IOException("InputStream claimed bytes were " +
 216                         "available, but read() returned -1.  What is going " +
 217                         "on?");
 218                 }
 219                 remaining -= rc;
 220                 if (remaining == 0) {
 221                     return b.length;
 222                 }
 223             }
 224         }
 225
 226         throw new IOException("InputStream claimed all bytes were available, " +
 227             "but now it is not.  What is going on?");
 228     }
 229
 230     /**
 231      * Reads up to len bytes of data from the input stream into an array of
 232      * bytes.
 233      *
 234      * @param b the buffer into which the data is read.
 235      * @param off the start offset in array b at which the data is written.
 236      * @param len the maximum number of bytes to read.
 237      * @return the total number of bytes read into the buffer, or -1 if there
 238      * is no more data because the end of the stream has been reached.
 239      * @throws IOException if an I/O error occurs
 240      */
 241     @Override
 242     public int read(final byte[] b, final int off,
 243         final int len) throws IOException {
 244
 245         if (timeoutMillis == 0) {
 246             // Block on the read().
 247             return stream.read(b, off, len);
 248         }
 249
 250         int remaining = len;
 251
 252         if (stream.available() >= remaining) {
 253             // Enough bytes are available now, return them.
 254             return stream.read(b, off, remaining);
 255         }
 256
 257         while (remaining > 0) {
 258
 259             // We will wait up to timeoutMillis to see if a byte is
 260             // available.  If not, we throw ReadTimeoutException.
 261             long checkTime = System.currentTimeMillis();
 262             while (stream.available() == 0) {
 263                 if (remaining > 0) {
 264                     return (len - remaining);
 265                 }
 266
 267                 long now = System.currentTimeMillis();
 268                 synchronized (this) {
 269                     if ((now - checkTime > timeoutMillis) || (cancel == true)) {
 270                         if (cancel == true) {
 271                             cancel = false;
 272                         }
 273                         throw new ReadTimeoutException("Timeout on read(): " +
 274                             (int) (now - checkTime) + " millis and still no " +
 275                             "data");
 276                     }
 277                 }
 278                 try {
 279                     // How long do we sleep for, eh?  For now we will go with
 280                     // 2 millis.
 281                     Thread.sleep(2);
 282                 } catch (InterruptedException e) {
 283                     // SQUASH
 284                 }
 285             }
 286
 287             if (stream.available() > 0) {
 288                 // At least one byte is available now, read it.
 289                 int n = stream.available();
 290                 if (remaining < n) {
 291                     n = remaining;
 292                 }
 293                 int rc = stream.read(b, off + len - remaining, n);
 294                 if (rc == -1) {
 295                     // This shouldn't happen.
 296                     throw new IOException("InputStream claimed bytes were " +
 297                         "available, but read() returned -1.  What is going " +
 298                         "on?");
 299                 }
 300                 remaining -= rc;
 301                 if (remaining == 0) {
 302                     return len;
 303                 }
 304             }
 305         }
 306
 307         throw new IOException("InputStream claimed all bytes were available, " +
 308             "but now it is not.  What is going on?");
 309     }
 310
 311     /**
 312      * Returns an estimate of the number of bytes that can be read (or
 313      * skipped over) from this input stream without blocking by the next
 314      * invocation of a method for this input stream.
 315      *
 316      * @return an estimate of the number of bytes that can be read (or
 317      * skipped over) from this input stream without blocking or 0 when it
 318      * reaches the end of the input stream.
 319      * @throws IOException if an I/O error occurs
 320      */
 321     @Override
 322     public int available() throws IOException {
 323         return stream.available();
 324     }
 325
 326     /**
 327      * Closes this input stream and releases any system resources associated
 328      * with the stream.
 329      *
 330      * @throws IOException if an I/O error occurs
 331      */
 332     @Override
 333     public void close() throws IOException {
 334         stream.close();
 335     }
 336
 337     /**
 338      * Marks the current position in this input stream.
 339      *
 340      * @param readLimit the maximum limit of bytes that can be read before
 341      * the mark position becomes invalid
 342      */
 343     @Override
 344     public void mark(final int readLimit) {
 345         stream.mark(readLimit);
 346     }
 347
 348     /**
 349      * Tests if this input stream supports the mark and reset methods.
 350      *
 351      * @return true if this stream instance supports the mark and reset
 352      * methods; false otherwise
 353      */
 354     @Override
 355     public boolean markSupported() {
 356         return stream.markSupported();
 357     }
 358
 359     /**
 360      * Repositions this stream to the position at the time the mark method
 361      * was last called on this input stream.
 362      *
 363      * @throws IOException if an I/O error occurs
 364      */
 365     @Override
 366     public void reset() throws IOException {
 367         stream.reset();
 368     }
 369
 370     /**
 371      * Skips over and discards n bytes of data from this input stream.
 372      *
 373      * @param n the number of bytes to be skipped
 374      * @return the actual number of bytes skipped
 375      * @throws IOException if an I/O error occurs
 376      */
 377     @Override
 378     public long skip(final long n) throws IOException {
 379         return stream.skip(n);
 380     }
 381
 382     // ------------------------------------------------------------------------
 383     // TimeoutInputStream -----------------------------------------------------
 384     // ------------------------------------------------------------------------
 385
 386     /**
 387      * Request that the current read() operation timeout immediately.
 388      */
 389     public synchronized void cancelRead() {
 390         cancel = true;
 391     }
 392
 393 }