code: public final Stream<T> setTimeout(@Nullable final Long timestamp) {
          Preconditions.checkArgument(timestamp == null || timestamp > System.currentTimeMillis());
          synchronized (this.state) {
              if (this.state.closed) {
                  return this; // NOP, already closed
              }
              if (this.state.timeoutFuture != null) {
                  if (!this.state.timeoutFuture.cancel(false)) {
                      return this; // NOP, timeout already occurred
                  }
              }
              if (timestamp != null) {
                  this.state.timeoutFuture = Data.getExecutor().schedule(new Runnable() {
  
                      @Override
                      public void run() {
                          close();
                      }
  
                  }, Math.max(0, timestamp - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
              }
              return this;
          }
      }
docs: Sets a timeout by supplying the absolute milliseconds timestamp when the {@code Stream}
  will be forcedly closed. If the supplied value is null, any previously set timeout is
  removed. In case the {@code Stream has already been closed}, or has just timed out due to a
  previously set timeout, calling this method has no effect.
  
  @param timestamp
  the milliseconds timestamp when the {@code Stream} will be closed
  @return this {@code Stream}, for call chaining