HttpClientRequestImpl.java

/*
 * Copyright (c) 2011-2017 Contributors to the Eclipse Foundation
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
 * which is available at https://www.apache.org/licenses/LICENSE-2.0.
 *
 * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
 */

package io.vertx.core.http.impl;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.vertx.codegen.annotations.Nullable;
import io.vertx.core.*;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.CaseInsensitiveHeaders;
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpClientResponse;
import io.vertx.core.http.HttpConnection;
import io.vertx.core.http.HttpFrame;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpVersion;
import io.vertx.core.impl.VertxInternal;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.net.NetSocket;

import java.util.List;
import java.util.Objects;

import static io.vertx.core.http.HttpHeaders.*;

/**
 * This class is optimised for performance when used on the same event loop that is was passed to the handler with.
 * However it can be used safely from other threads.
 *
 * The internal state is protected using the synchronized keyword. If always used on the same event loop, then
 * we benefit from biased locking which makes the overhead of synchronized near zero.
 *
 * @author <a href="http://tfox.org">Tim Fox</a>
 */
public class HttpClientRequestImpl extends HttpClientRequestBase implements HttpClientRequest {

  static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);

  private final VertxInternal vertx;
  private Handler<HttpClientResponse> respHandler;
  private Handler<Void> endHandler;
  private boolean chunked;
  private String hostHeader;
  private String rawMethod;
  private Handler<Void> continueHandler;
  private HttpClientStream stream;
  private volatile Object lock;
  private Handler<Void> drainHandler;
  private Handler<HttpClientRequest> pushHandler;
  private Handler<HttpConnection> connectionHandler;
  private Handler<HttpVersion> headersCompletionHandler;
  private boolean completed;
  private Handler<Void> completionHandler;
  private Long reset;
  private ByteBuf pendingChunks;
  private int pendingMaxSize = -1;
  private int followRedirects;
  private boolean connecting;
  private long written;
  private CaseInsensitiveHeaders headers;

  // completed => drainHandler = null

  HttpClientRequestImpl(HttpClientImpl client, boolean ssl, HttpMethod method, String host, int port,
                        String relativeURI, VertxInternal vertx) {
    super(client, ssl, method, host, port, relativeURI);
    this.chunked = false;
    this.vertx = vertx;
  }

  @Override
  public int streamId() {
    synchronized (getLock()) {
      return stream != null ? stream.id() : -1;
    }
  }

  @Override
  public  HttpClientRequest handler(Handler<HttpClientResponse> handler) {
    synchronized (getLock()) {
      if (handler != null) {
        checkComplete();
        respHandler = checkConnect(method, handler);
      } else {
        respHandler = null;
      }
      return this;
    }
  }

  @Override
  public HttpClientRequest pause() {
    return this;
  }

  @Override
  public HttpClientRequest resume() {
    return this;
  }

  @Override
  public HttpClientRequest setFollowRedirects(boolean followRedirects) {
    synchronized (getLock()) {
      checkComplete();
      if (followRedirects) {
        this.followRedirects = client.getOptions().getMaxRedirects() - 1;
      } else {
        this.followRedirects = 0;
      }
      return this;
    }
  }

  @Override
  public HttpClientRequest endHandler(Handler<Void> endHandler) {
    synchronized (getLock()) {
      if (endHandler != null) {
        checkComplete();
      }
      this.endHandler = endHandler;
      return this;
    }
  }

  @Override
  public HttpClientRequestImpl setChunked(boolean chunked) {

    // If connecting -> write buffer in list
    // If connected -> check if the buffer needs to be written

    synchronized (getLock()) {
      checkComplete();
      if (written > 0) {
        throw new IllegalStateException("Cannot set chunked after data has been written on request");
      }
      // HTTP 1.0 does not support chunking so we ignore this if HTTP 1.0
      if (client.getOptions().getProtocolVersion() != io.vertx.core.http.HttpVersion.HTTP_1_0) {
        this.chunked = chunked;
      }
      return this;
    }
  }

  @Override
  public boolean isChunked() {
    synchronized (getLock()) {
      return chunked;
    }
  }

  @Override
  public String getRawMethod() {
    synchronized (getLock()) {
      return rawMethod;
    }
  }

  @Override
  public HttpClientRequest setRawMethod(String method) {
    synchronized (getLock()) {
      this.rawMethod = method;
      return this;
    }
  }

  @Override
  public HttpClientRequest setHost(String host) {
    synchronized (getLock()) {
      this.hostHeader = host;
      return this;
    }
  }

  @Override
  public String getHost() {
    synchronized (getLock()) {
      return hostHeader;
    }
  }

  @Override
  public MultiMap headers() {
    synchronized (getLock()) {
      if (headers == null) {
        headers = new CaseInsensitiveHeaders();
      }
      return headers;
    }
  }

  @Override
  public HttpClientRequest putHeader(String name, String value) {
    synchronized (getLock()) {
      checkComplete();
      headers().set(name, value);
      return this;
    }
  }

  @Override
  public HttpClientRequest putHeader(String name, Iterable<String> values) {
    synchronized (getLock()) {
      checkComplete();
      headers().set(name, values);
      return this;
    }
  }

  @Override
  public HttpClientRequest setWriteQueueMaxSize(int maxSize) {
    synchronized (getLock()) {
      checkComplete();
      if (stream != null) {
        stream.doSetWriteQueueMaxSize(maxSize);
      } else {
        pendingMaxSize = maxSize;
      }
      return this;
    }
  }

  @Override
  public boolean writeQueueFull() {
    synchronized (getLock()) {
      checkComplete();
      return stream != null && stream.isNotWritable();
    }
  }

  @Override
  public HttpClientRequest drainHandler(Handler<Void> handler) {
    synchronized (getLock()) {
      if (handler != null) {
        checkComplete();
        drainHandler = handler;
        if (stream != null) {
          stream.getContext().runOnContext(v -> {
            synchronized (getLock()) {
              if (stream != null) {
                stream.checkDrained();
              }
            }
          });
        }
      } else {
        drainHandler = null;
      }
      return this;
    }
  }

  @Override
  public HttpClientRequest continueHandler(Handler<Void> handler) {
    synchronized (getLock()) {
      if (handler != null) {
        checkComplete();
      }
      this.continueHandler = handler;
      return this;
    }
  }

  @Override
  public HttpClientRequest sendHead() {
    return sendHead(null);
  }

  @Override
  public HttpClientRequest sendHead(Handler<HttpVersion> completionHandler) {
    synchronized (getLock()) {
      checkComplete();
      checkResponseHandler();
      if (stream != null) {
        throw new IllegalStateException("Head already written");
      } else {
        headersCompletionHandler = completionHandler;
        connect();
      }
      return this;
    }
  }

  @Override
  public HttpClientRequest putHeader(CharSequence name, CharSequence value) {
    synchronized (getLock()) {
      checkComplete();
      headers().set(name, value);
      return this;
    }
  }

  @Override
  public HttpClientRequest putHeader(CharSequence name, Iterable<CharSequence> values) {
    synchronized (getLock()) {
      checkComplete();
      headers().set(name, values);
      return this;
    }
  }

  @Override
  public HttpClientRequest pushHandler(Handler<HttpClientRequest> handler) {
    synchronized (getLock()) {
      pushHandler = handler;
    }
    return this;
  }

  @Override
  public boolean reset(long code) {
    synchronized (getLock()) {
      if (reset == null) {
        reset = code;
        if (tryComplete()) {
          if (completionHandler != null) {
            completionHandler.handle(null);
          }
        }
        if (stream != null) {
          stream.reset(code);
        }
        return true;
      }
      return false;
    }
  }

  private boolean tryComplete() {
    if (!completed) {
      completed = true;
      drainHandler = null;
      return true;
    } else {
      return false;
    }
  }

  @Override
  public HttpConnection connection() {
    synchronized (getLock()) {
      return stream != null ? stream.connection() : null;
    }
  }

  @Override
  public HttpClientRequest connectionHandler(@Nullable Handler<HttpConnection> handler) {
    synchronized (getLock()) {
      connectionHandler = handler;
      return this;
    }
  }

  @Override
  public HttpClientRequest writeCustomFrame(int type, int flags, Buffer payload) {
    synchronized (getLock()) {
      if (stream == null) {
        throw new IllegalStateException("Not yet connected");
      }
      stream.writeFrame(type, flags, payload.getByteBuf());
    }
    return this;
  }

  void handleDrained() {
    synchronized (getLock()) {
      if (drainHandler != null) {
        try {
          drainHandler.handle(null);
        } catch (Throwable t) {
          handleException(t);
        }
      }
    }
  }

  private void handleNextRequest(HttpClientRequestImpl next, long timeoutMs) {
    next.handler(respHandler);
    next.exceptionHandler(exceptionHandler());
    exceptionHandler(null);
    next.endHandler(endHandler);
    next.pushHandler = pushHandler;
    next.followRedirects = followRedirects - 1;
    next.written = written;
    if (next.hostHeader == null) {
      next.hostHeader = hostHeader;
    }
    if (headers != null && next.headers == null) {
      next.headers().addAll(headers);
    }
    Future<Void> fut = Future.future();
    fut.setHandler(ar -> {
      if (ar.succeeded()) {
        if (timeoutMs > 0) {
          next.setTimeout(timeoutMs);
        }
        next.end();
      } else {
        next.handleException(ar.cause());
      }
    });
    if (exceptionOccurred != null) {
      fut.fail(exceptionOccurred);
    }
    else if (completed) {
      fut.complete();
    } else {
      exceptionHandler(err -> {
        if (!fut.isComplete()) {
          fut.fail(err);
        }
      });
      completionHandler = v -> {
        if (!fut.isComplete()) {
          fut.complete();
        }
      };
    }
  }

  protected void doHandleResponse(HttpClientResponseImpl resp, long timeoutMs) {
    if (reset == null) {
      int statusCode = resp.statusCode();
      if (followRedirects > 0 && statusCode >= 300 && statusCode < 400) {
        Future<HttpClientRequest> next = client.redirectHandler().apply(resp);
        if (next != null) {
          next.setHandler(ar -> {
            if (ar.succeeded()) {
              handleNextRequest((HttpClientRequestImpl) ar.result(), timeoutMs);
            } else {
              handleException(ar.cause());
            }
          });
          return;
        }
      }
      if (statusCode == 100) {
        if (continueHandler != null) {
          continueHandler.handle(null);
        }
      } else {
        if (respHandler != null) {
          respHandler.handle(resp);
        }
        if (endHandler != null) {
          endHandler.handle(null);
        }
      }
    }
  }

  @Override
  protected String hostHeader() {
    return hostHeader != null ? hostHeader : super.hostHeader();
  }

  // After connecting we should synchronize on the client connection instance to prevent deadlock conditions
  // but there is a catch - the client connection is null before connecting so we synchronized on this before that
  // point
  protected Object getLock() {
    // We do the initial check outside the synchronized block to prevent the hit of synchronized once the conn has
    // been set
    if (lock != null) {
      return lock;
    } else {
      synchronized (this) {
        if (lock != null) {
          return lock;
        } else {
          return this;
        }
      }
    }
  }

  private Handler<HttpClientResponse> checkConnect(io.vertx.core.http.HttpMethod method, Handler<HttpClientResponse> handler) {
    if (method == io.vertx.core.http.HttpMethod.CONNECT) {
      // special handling for CONNECT
      handler = connectHandler(handler);
    }
    return handler;
  }

  private Handler<HttpClientResponse> connectHandler(Handler<HttpClientResponse> responseHandler) {
    Objects.requireNonNull(responseHandler, "no null responseHandler accepted");
    return resp -> {
      HttpClientResponse response;
      if (resp.statusCode() == 200) {
        // connect successful force the modification of the ChannelPipeline
        // beside this also pause the socket for now so the user has a chance to register its dataHandler
        // after received the NetSocket
        NetSocket socket = resp.netSocket();
        socket.pause();

        response = new HttpClientResponse() {
          private boolean resumed;

          @Override
          public HttpClientRequest request() {
            return resp.request();
          }

          @Override
          public int statusCode() {
            return resp.statusCode();
          }

          @Override
          public String statusMessage() {
            return resp.statusMessage();
          }

          @Override
          public MultiMap headers() {
            return resp.headers();
          }

          @Override
          public String getHeader(String headerName) {
            return resp.getHeader(headerName);
          }

          @Override
          public String getHeader(CharSequence headerName) {
            return resp.getHeader(headerName);
          }

          @Override
          public String getTrailer(String trailerName) {
            return resp.getTrailer(trailerName);
          }

          @Override
          public MultiMap trailers() {
            return resp.trailers();
          }

          @Override
          public List<String> cookies() {
            return resp.cookies();
          }

          @Override
          public HttpVersion version() {
            return resp.version();
          }

          @Override
          public HttpClientResponse bodyHandler(Handler<Buffer> bodyHandler) {
            resp.bodyHandler(bodyHandler);
            return this;
          }

          @Override
          public HttpClientResponse customFrameHandler(Handler<HttpFrame> handler) {
            resp.customFrameHandler(handler);
            return this;
          }

          @Override
          public synchronized NetSocket netSocket() {
            if (!resumed) {
              resumed = true;
              vertx.getContext().runOnContext((v) -> socket.resume()); // resume the socket now as the user had the chance to register a dataHandler
            }
            return socket;
          }

          @Override
          public HttpClientResponse endHandler(Handler<Void> endHandler) {
            resp.endHandler(endHandler);
            return this;
          }

          @Override
          public HttpClientResponse handler(Handler<Buffer> handler) {
            resp.handler(handler);
            return this;
          }

          @Override
          public HttpClientResponse pause() {
            resp.pause();
            return this;
          }

          @Override
          public HttpClientResponse resume() {
            resp.resume();
            return this;
          }

          @Override
          public HttpClientResponse exceptionHandler(Handler<Throwable> handler) {
            resp.exceptionHandler(handler);
            return this;
          }
        };
      } else {
        response = resp;
      }
      responseHandler.handle(response);
    };
  }

  private synchronized void connect() {
    if (!connecting) {

      if (method == HttpMethod.OTHER && rawMethod == null) {
        throw new IllegalStateException("You must provide a rawMethod when using an HttpMethod.OTHER method");
      }

      String peerHost;
      if (hostHeader != null) {
        int idx = hostHeader.lastIndexOf(':');
        if (idx != -1) {
          peerHost = hostHeader.substring(0, idx);
        } else {
          peerHost = hostHeader;
        }
      } else {
        peerHost = host;
      }

      // Capture the handler
      Handler<HttpConnection> initializer = connectionHandler;

      // We defer actual connection until the first part of body is written or end is called
      // This gives the user an opportunity to set an exception handler before connecting so
      // they can capture any exceptions on connection
      client.getConnectionForRequest(peerHost, ssl, port, host, (ctx, conn) -> {
        // No need to synchronize as the thread is the same that set exceptionOccurred to true
        // exceptionOccurred=true getting the connection => it's a TimeoutException
        if (exceptionOccurred != null || reset != null) {
          return false;
        }

        //
        if (!conn.checkInitialized() && initializer != null) {
          ctx.executeFromIO(() -> {
            initializer.handle(conn);
          });
        }

        //
        conn.createStream(HttpClientRequestImpl.this, ar -> {
          if (ar.succeeded()) {
            HttpClientStream stream = ar.result();
            ctx.executeFromIO(v -> {
              connected(stream, HttpClientRequestImpl.this.headersCompletionHandler);
            });
          } else {
            throw new RuntimeException(ar.cause());
          }
        });
        return true;
      }, (ctx, failure) -> {
        ctx.executeFromIO(v -> {
          handleException(failure);
        });
      });
      connecting = true;
    }
  }

  synchronized void retry() {
    connecting = false;
    connect();
  }

  private void connected(HttpClientStream stream, Handler<HttpVersion> headersCompletionHandler) {

    HttpClientConnection conn = stream.connection();

    synchronized (this) {
      this.stream = stream;
      stream.beginRequest();

      // If anything was written or the request ended before we got the connection, then
      // we need to write it now

      if (pendingMaxSize != -1) {
        this.stream.doSetWriteQueueMaxSize(pendingMaxSize);
      }

      if (pendingChunks != null) {
        ByteBuf pending = pendingChunks;
        pendingChunks = null;

        if (completed) {
          // we also need to write the head so optimize this and write all out in once
          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, pending, true);
          conn.reportBytesWritten(written);
          if (respHandler != null) {
            this.stream.endRequest();
          }
        } else {
          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, pending, false);
          if (headersCompletionHandler != null) {
            headersCompletionHandler.handle(stream.version());
          }
        }
      } else {
        if (completed) {
          // we also need to write the head so optimize this and write all out in once
          stream.writeHeadWithContent(method, rawMethod, uri, headers, hostHeader(), chunked, null, true);
          conn.reportBytesWritten(written);
          if (respHandler != null) {
            this.stream.endRequest();
          }
        } else {
          stream.writeHead(method, rawMethod, uri, headers, hostHeader(), chunked);
          if (headersCompletionHandler != null) {
            headersCompletionHandler.handle(stream.version());
          }
        }
      }

      // Set the lock at the end of the block so we are sure that another non vertx thread will get access to the connection
      // when this callback runs on the 'this' lock
      this.lock = conn;
    }
  }

  private boolean contentLengthSet() {
    return headers != null && headers().contains(CONTENT_LENGTH);
  }

  @Override
  public void end(String chunk) {
    end(Buffer.buffer(chunk));
  }

  @Override
  public void end(String chunk, String enc) {
    Objects.requireNonNull(enc, "no null encoding accepted");
    end(Buffer.buffer(chunk, enc));
  }

  @Override
  public void end(Buffer chunk) {
    write(chunk.getByteBuf(), true);
  }

  @Override
  public void end() {
    write(null, true);
  }

  @Override
  public HttpClientRequestImpl write(Buffer chunk) {
    ByteBuf buf = chunk.getByteBuf();
    write(buf, false);
    return this;
  }

  @Override
  public HttpClientRequestImpl write(String chunk) {
    return write(Buffer.buffer(chunk));
  }

  @Override
  public HttpClientRequestImpl write(String chunk, String enc) {
    Objects.requireNonNull(enc, "no null encoding accepted");
    return write(Buffer.buffer(chunk, enc));
  }

  private void write(ByteBuf buff, boolean end) {
    synchronized (getLock()) {
      checkComplete();
      checkResponseHandler();
      _write(buff, end);
    }
  }

  private void _write(ByteBuf buff, boolean end) {
    if (buff == null && !end) {
      // nothing to write to the connection just return
      return;
    }

    if (end) {
      if (buff != null && !chunked && !contentLengthSet()) {
        headers().set(CONTENT_LENGTH, String.valueOf(buff.readableBytes()));
      }
    } else {
      if (!chunked && !contentLengthSet()) {
        throw new IllegalStateException("You must set the Content-Length header to be the total size of the message "
            + "body BEFORE sending any data if you are not using HTTP chunked encoding.");
      }
    }

    if (buff != null) {
      written += buff.readableBytes();
    }

    if (stream == null) {
      if (buff != null) {
        if (pendingChunks == null) {
          pendingChunks = buff;
        } else {
          CompositeByteBuf pending;
          if (pendingChunks instanceof CompositeByteBuf) {
            pending = (CompositeByteBuf) pendingChunks;
          } else {
            pending = Unpooled.compositeBuffer();
            pending.addComponent(true, pendingChunks);
            pendingChunks = pending;
          }
          pending.addComponent(true, buff);
        }
      }
      if (end) {
        tryComplete();
        if (completionHandler != null) {
          completionHandler.handle(null);
        }
      }
      connect();
    } else {
      stream.writeBuffer(buff, end);
      if (end) {
        stream.connection().reportBytesWritten(written);
        if (respHandler != null) {
          stream.endRequest();
        }
        tryComplete();
        if (completionHandler != null) {
          completionHandler.handle(null);
        }
      }
    }
  }

  protected void checkComplete() {
    if (completed) {
      throw new IllegalStateException("Request already complete");
    }
  }

  private void checkResponseHandler() {
    if (respHandler == null) {
      throw new IllegalStateException("You must set an handler for the HttpClientResponse before connecting");
    }
  }

  Handler<HttpClientRequest> pushHandler() {
    synchronized (getLock()) {
      return pushHandler;
    }
  }
}