lib/jetty/org/eclipse/jetty/server/HttpInput.java

   1 //
   2 //  ========================================================================
   3 //  Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.
   4 //  ------------------------------------------------------------------------
   5 //  All rights reserved. This program and the accompanying materials
   6 //  are made available under the terms of the Eclipse Public License v1.0
   7 //  and Apache License v2.0 which accompanies this distribution.
   8 //
   9 //      The Eclipse Public License is available at
  10 //      http://www.eclipse.org/legal/epl-v10.html
  11 //
  12 //      The Apache License v2.0 is available at
  13 //      http://www.opensource.org/licenses/apache2.0.php
  14 //
  15 //  You may elect to redistribute this code under either of these licenses.
  16 //  ========================================================================
  17 //
  18
  19 package org.eclipse.jetty.server;
  20
  21 import java.io.IOException;
  22 import java.util.Objects;
  23
  24 import javax.servlet.ReadListener;
  25 import javax.servlet.ServletInputStream;
  26
  27 import org.eclipse.jetty.io.EofException;
  28 import org.eclipse.jetty.io.RuntimeIOException;
  29 import org.eclipse.jetty.util.log.Log;
  30 import org.eclipse.jetty.util.log.Logger;
  31
  32 /**
  33  * {@link HttpInput} provides an implementation of {@link ServletInputStream} for {@link HttpChannel}.
  34  * <p/>
  35  * Content may arrive in patterns such as [content(), content(), messageComplete()] so that this class
  36  * maintains two states: the content state that tells whether there is content to consume and the EOF
  37  * state that tells whether an EOF has arrived.
  38  * Only once the content has been consumed the content state is moved to the EOF state.
  39  */
  40 public abstract class HttpInput<T> extends ServletInputStream implements Runnable
  41 {
  42     private final static Logger LOG = Log.getLogger(HttpInput.class);
  43
  44     private final byte[] _oneByteBuffer = new byte[1];
  45     private final Object _lock;
  46     private HttpChannelState _channelState;
  47     private ReadListener _listener;
  48     private Throwable _onError;
  49     private boolean _notReady;
  50     private State _contentState = STREAM;
  51     private State _eofState;
  52     private long _contentRead;
  53
  54     protected HttpInput()
  55     {
  56         this(null);
  57     }
  58
  59     protected HttpInput(Object lock)
  60     {
  61         _lock = lock == null ? this : lock;
  62     }
  63
  64     public void init(HttpChannelState state)
  65     {
  66         synchronized (lock())
  67         {
  68             _channelState = state;
  69         }
  70     }
  71
  72     public final Object lock()
  73     {
  74         return _lock;
  75     }
  76
  77     public void recycle()
  78     {
  79         synchronized (lock())
  80         {
  81             _listener = null;
  82             _onError = null;
  83             _notReady = false;
  84             _contentState = STREAM;
  85             _eofState = null;
  86             _contentRead = 0;
  87         }
  88     }
  89
  90     @Override
  91     public int available()
  92     {
  93         try
  94         {
  95             synchronized (lock())
  96             {
  97                 T item = getNextContent();
  98                 return item == null ? 0 : remaining(item);
  99             }
 100         }
 101         catch (IOException e)
 102         {
 103             throw new RuntimeIOException(e);
 104         }
 105     }
 106
 107     @Override
 108     public int read() throws IOException
 109     {
 110         int read = read(_oneByteBuffer, 0, 1);
 111         return read < 0 ? -1 : _oneByteBuffer[0] & 0xFF;
 112     }
 113
 114     @Override
 115     public int read(byte[] b, int off, int len) throws IOException
 116     {
 117         synchronized (lock())
 118         {
 119             T item = getNextContent();
 120             if (item == null)
 121             {
 122                 _contentState.waitForContent(this);
 123                 item = getNextContent();
 124                 if (item == null)
 125                     return _contentState.noContent();
 126             }
 127             int l = get(item, b, off, len);
 128             _contentRead += l;
 129             return l;
 130         }
 131     }
 132
 133     /**
 134      * A convenience method to call nextContent and to check the return value, which if null then the
 135      * a check is made for EOF and the state changed accordingly.
 136      *
 137      * @return Content or null if none available.
 138      * @throws IOException
 139      * @see #nextContent()
 140      */
 141     protected T getNextContent() throws IOException
 142     {
 143         T content = nextContent();
 144         if (content == null)
 145         {
 146             synchronized (lock())
 147             {
 148                 if (_eofState != null)
 149                 {
 150                     if (LOG.isDebugEnabled())
 151                         LOG.debug("{} eof {}", this, _eofState);
 152                     _contentState = _eofState;
 153                 }
 154             }
 155         }
 156         return content;
 157     }
 158
 159     /**
 160      * Access the next content to be consumed from.   Returning the next item does not consume it
 161      * and it may be returned multiple times until it is consumed.
 162      * <p/>
 163      * Calls to {@link #get(Object, byte[], int, int)}
 164      * or {@link #consume(Object, int)} are required to consume data from the content.
 165      *
 166      * @return the content or null if none available.
 167      * @throws IOException if retrieving the content fails
 168      */
 169     protected abstract T nextContent() throws IOException;
 170
 171     /**
 172      * @param item the content
 173      * @return how many bytes remain in the given content
 174      */
 175     protected abstract int remaining(T item);
 176
 177     /**
 178      * Copies the given content into the given byte buffer.
 179      *
 180      * @param item   the content to copy from
 181      * @param buffer the buffer to copy into
 182      * @param offset the buffer offset to start copying from
 183      * @param length the space available in the buffer
 184      * @return the number of bytes actually copied
 185      */
 186     protected abstract int get(T item, byte[] buffer, int offset, int length);
 187
 188     /**
 189      * Consumes the given content.
 190      *
 191      * @param item   the content to consume
 192      * @param length the number of bytes to consume
 193      */
 194     protected abstract void consume(T item, int length);
 195
 196     /**
 197      * Blocks until some content or some end-of-file event arrives.
 198      *
 199      * @throws IOException if the wait is interrupted
 200      */
 201     protected abstract void blockForContent() throws IOException;
 202
 203     /**
 204      * Adds some content to this input stream.
 205      *
 206      * @param item the content to add
 207      */
 208     public abstract void content(T item);
 209
 210     protected boolean onAsyncRead()
 211     {
 212         synchronized (lock())
 213         {
 214             if (_listener == null)
 215                 return false;
 216         }
 217         _channelState.onReadPossible();
 218         return true;
 219     }
 220
 221     public long getContentRead()
 222     {
 223         synchronized (lock())
 224         {
 225             return _contentRead;
 226         }
 227     }
 228
 229     /**
 230      * This method should be called to signal that an EOF has been
 231      * detected before all the expected content arrived.
 232      * <p/>
 233      * Typically this will result in an EOFException being thrown
 234      * from a subsequent read rather than a -1 return.
 235      */
 236     public void earlyEOF()
 237     {
 238         synchronized (lock())
 239         {
 240             if (!isEOF())
 241             {
 242                 if (LOG.isDebugEnabled())
 243                     LOG.debug("{} early EOF", this);
 244                 _eofState = EARLY_EOF;
 245                 if (_listener == null)
 246                     return;
 247             }
 248         }
 249         _channelState.onReadPossible();
 250     }
 251
 252
 253     public boolean isEarlyEOF()
 254     {
 255         synchronized (lock())
 256         {
 257             return _contentState==EARLY_EOF;
 258         }
 259     }
 260
 261     /**
 262      * This method should be called to signal that all the expected
 263      * content arrived.
 264      */
 265     public void messageComplete()
 266     {
 267         synchronized (lock())
 268         {
 269             if (!isEOF())
 270             {
 271                 if (LOG.isDebugEnabled())
 272                     LOG.debug("{} EOF", this);
 273                 _eofState = EOF;
 274                 if (_listener == null)
 275                     return;
 276             }
 277         }
 278         _channelState.onReadPossible();
 279     }
 280
 281     public boolean consumeAll()
 282     {
 283         synchronized (lock())
 284         {
 285             // Don't bother reading if we already know there was an error.
 286             if (_onError != null)
 287                 return false;
 288
 289             try
 290             {
 291                 while (!isFinished())
 292                 {
 293                     T item = getNextContent();
 294                     if (item == null)
 295                         _contentState.waitForContent(this);
 296                     else
 297                         consume(item, remaining(item));
 298                 }
 299                 return true;
 300             }
 301             catch (IOException e)
 302             {
 303                 LOG.debug(e);
 304                 return false;
 305             }
 306         }
 307     }
 308
 309     public boolean isAsync()
 310     {
 311         synchronized (lock())
 312         {
 313             return _contentState==ASYNC;
 314         }
 315     }
 316
 317     /**
 318      * @return whether an EOF has been detected, even though there may be content to consume.
 319      */
 320     public boolean isEOF()
 321     {
 322         synchronized (lock())
 323         {
 324             return _eofState != null && _eofState.isEOF();
 325         }
 326     }
 327
 328     @Override
 329     public boolean isFinished()
 330     {
 331         synchronized (lock())
 332         {
 333             return _contentState.isEOF();
 334         }
 335     }
 336
 337
 338     @Override
 339     public boolean isReady()
 340     {
 341         boolean finished;
 342         synchronized (lock())
 343         {
 344             if (_contentState.isEOF())
 345                 return true;
 346             if (_listener == null )
 347                 return true;
 348             if (available() > 0)
 349                 return true;
 350             if (_notReady)
 351                 return false;
 352             _notReady = true;
 353             finished = isFinished();
 354         }
 355         if (finished)
 356             _channelState.onReadPossible();
 357         else
 358             unready();
 359         return false;
 360     }
 361
 362     protected void unready()
 363     {
 364     }
 365
 366     @Override
 367     public void setReadListener(ReadListener readListener)
 368     {
 369         try
 370         {
 371             readListener = Objects.requireNonNull(readListener);
 372             boolean content;
 373             synchronized (lock())
 374             {
 375                 if (_contentState != STREAM)
 376                     throw new IllegalStateException("state=" + _contentState);
 377                 _contentState = ASYNC;
 378                 _listener = readListener;
 379                 _notReady = true;
 380
 381                 content = getNextContent()!=null || isEOF();
 382
 383             }
 384             if (content)
 385                 _channelState.onReadPossible();
 386             else
 387                 unready();
 388         }
 389         catch(IOException e)
 390         {
 391             throw new RuntimeIOException(e);
 392         }
 393     }
 394
 395     public void failed(Throwable x)
 396     {
 397         synchronized (lock())
 398         {
 399             if (_onError != null)
 400                 LOG.warn(x);
 401             else
 402                 _onError = x;
 403         }
 404     }
 405
 406     @Override
 407     public void run()
 408     {
 409         final Throwable error;
 410         final ReadListener listener;
 411         boolean available = false;
 412         final boolean eof;
 413
 414         synchronized (lock())
 415         {
 416             if (!_notReady || _listener == null)
 417                 return;
 418
 419             error = _onError;
 420             listener = _listener;
 421
 422             try
 423             {
 424                 T item = getNextContent();
 425                 available = item != null && remaining(item) > 0;
 426             }
 427             catch (Exception e)
 428             {
 429                 failed(e);
 430             }
 431
 432             eof = !available && isFinished();
 433             _notReady = !available && !eof;
 434         }
 435
 436         try
 437         {
 438             if (error != null)
 439                 listener.onError(error);
 440             else if (available)
 441                 listener.onDataAvailable();
 442             else if (eof)
 443                 listener.onAllDataRead();
 444             else
 445                 unready();
 446         }
 447         catch (Throwable e)
 448         {
 449             LOG.warn(e.toString());
 450             LOG.debug(e);
 451             listener.onError(e);
 452         }
 453     }
 454
 455     @Override
 456     public String toString()
 457     {
 458         return String.format("%s@%x[r=%d,s=%s,e=%s,f=%s]",
 459                 getClass().getSimpleName(),
 460                 hashCode(),
 461                 _contentRead,
 462                 _contentState,
 463                 _eofState,
 464                 _onError);
 465     }
 466
 467     protected static abstract class State
 468     {
 469         public void waitForContent(HttpInput<?> in) throws IOException
 470         {
 471         }
 472
 473         public int noContent() throws IOException
 474         {
 475             return -1;
 476         }
 477
 478         public boolean isEOF()
 479         {
 480             return false;
 481         }
 482     }
 483
 484     protected static final State STREAM = new State()
 485     {
 486         @Override
 487         public void waitForContent(HttpInput<?> input) throws IOException
 488         {
 489             input.blockForContent();
 490         }
 491
 492         @Override
 493         public String toString()
 494         {
 495             return "STREAM";
 496         }
 497     };
 498
 499     protected static final State ASYNC = new State()
 500     {
 501         @Override
 502         public int noContent() throws IOException
 503         {
 504             return 0;
 505         }
 506
 507         @Override
 508         public String toString()
 509         {
 510             return "ASYNC";
 511         }
 512     };
 513
 514     protected static final State EARLY_EOF = new State()
 515     {
 516         @Override
 517         public int noContent() throws IOException
 518         {
 519             throw new EofException("Early EOF");
 520         }
 521
 522         @Override
 523         public boolean isEOF()
 524         {
 525             return true;
 526         }
 527
 528         @Override
 529         public String toString()
 530         {
 531             return "EARLY_EOF";
 532         }
 533     };
 534
 535     protected static final State EOF = new State()
 536     {
 537         @Override
 538         public boolean isEOF()
 539         {
 540             return true;
 541         }
 542
 543         @Override
 544         public String toString()
 545         {
 546             return "EOF";
 547         }
 548     };
 549 }