transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java - servicecomb-java-chassis - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.servicecomb.transport.rest.vertx;

 import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Set;

 import javax.ws.rs.core.MediaType;

 import org.apache.servicecomb.common.rest.codec.RestObjectMapperFactory;
 import org.apache.servicecomb.core.Endpoint;
 import org.apache.servicecomb.core.transport.AbstractTransport;
 import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
 import org.apache.servicecomb.foundation.common.utils.ExceptionUtils;
 import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
 import org.apache.servicecomb.foundation.ssl.SSLCustom;
 import org.apache.servicecomb.foundation.ssl.SSLOption;
 import org.apache.servicecomb.foundation.ssl.SSLOptionFactory;
 import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
 import org.apache.servicecomb.foundation.vertx.metrics.DefaultHttpServerMetrics;
 import org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
 import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
 import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
 import org.apache.servicecomb.transport.rest.vertx.accesslog.AccessLogConfiguration;
 import org.apache.servicecomb.transport.rest.vertx.accesslog.impl.AccessLogHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpHeaders;

 import com.netflix.config.DynamicPropertyFactory;

 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Context;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.Http2Settings;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServer;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.HttpServerResponse;
 import io.vertx.core.net.impl.ConnectionBase;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.ext.web.handler.CorsHandler;

 public class RestServerVerticle extends AbstractVerticle {
   private static final Logger LOGGER = LoggerFactory.getLogger(RestServerVerticle.class);

   private static final String SSL_KEY = "rest.provider";

   private Endpoint endpoint;

   private URIEndpointObject endpointObject;

   @Override
   public void init(Vertx vertx, Context context) {
     super.init(vertx, context);
     this.endpoint = (Endpoint) context.config().getValue(AbstractTransport.ENDPOINT_KEY);
     this.endpointObject = (URIEndpointObject) endpoint.getAddress();
   }

   @Override
   public void start(Promise<Void> startPromise) throws Exception {
     try {
       super.start();
       // 如果本地未配置地址，则表示不必监听，只需要作为客户端使用即可
       if (endpointObject == null) {
         LOGGER.warn("rest listen address is not configured, will not start.");
         startPromise.complete();
         return;
       }
       Router mainRouter = Router.router(vertx);
       mountAccessLogHandler(mainRouter);
       mountCorsHandler(mainRouter);
       initDispatcher(mainRouter);
       mountGlobalRestFailureHandler(mainRouter);
       HttpServer httpServer = createHttpServer();
       httpServer.requestHandler(mainRouter);
       httpServer.connectionHandler(connection -> {
         DefaultHttpServerMetrics serverMetrics = (DefaultHttpServerMetrics) ((ConnectionBase) connection).metrics();
         DefaultServerEndpointMetric endpointMetric = serverMetrics.getEndpointMetric();
         long connectedCount = endpointMetric.getCurrentConnectionCount();
         int connectionLimit = DynamicPropertyFactory.getInstance()
             .getIntProperty("servicecomb.rest.server.connection-limit", Integer.MAX_VALUE).get();
         if (connectedCount > connectionLimit) {
           connection.close();
           endpointMetric.onRejectByConnectionLimit();
         }
       });
       httpServer.exceptionHandler(e -> {
         if (e instanceof ClosedChannelException) {
           // This is quite normal in between browser and ege, so do not print out.
           LOGGER.debug("Unexpected error in server.{}", ExceptionUtils.getExceptionMessageWithoutTrace(e));
         } else {
           LOGGER.error("Unexpected error in server.{}", ExceptionUtils.getExceptionMessageWithoutTrace(e));
         }
       });
       startListen(httpServer, startPromise);
     } catch (Throwable e) {
       // vert.x got some states that not print error and execute call back in VertexUtils.blockDeploy, we add a log our self.
       LOGGER.error("", e);
       throw e;
     }
   }

   private void mountGlobalRestFailureHandler(Router mainRouter) {
     GlobalRestFailureHandler globalRestFailureHandler =
         SPIServiceUtils.getPriorityHighestService(GlobalRestFailureHandler.class);
     Handler<RoutingContext> failureHandler = null == globalRestFailureHandler ?
         ctx -> {
           if (ctx.response().closed() || ctx.response().ended()) {
             // response has been sent, do nothing
             LOGGER.error("get a failure with closed response", ctx.failure());
             return;
           }
           HttpServerResponse response = ctx.response();
           if (ctx.failure() instanceof InvocationException) {
             // ServiceComb defined exception
             InvocationException exception = (InvocationException) ctx.failure();
             response.setStatusCode(exception.getStatusCode());
             response.setStatusMessage(exception.getReasonPhrase());
             response.end(exception.getErrorData().toString());
             return;
           }

           LOGGER.error("unexpected failure happened", ctx.failure());
           try {
             // unknown exception
             CommonExceptionData unknownError = new CommonExceptionData("unknown error");
             ctx.response().setStatusCode(500).putHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON)
                 .end(RestObjectMapperFactory.getRestObjectMapper().writeValueAsString(unknownError));
           } catch (Exception e) {
             LOGGER.error("failed to send error response!", e);
           }
         }
         : globalRestFailureHandler;

     mainRouter.route()
         // this handler does nothing, just ensure the failure handler can catch exception
         .handler(RoutingContext::next)
         .failureHandler(failureHandler);
   }

   private void mountAccessLogHandler(Router mainRouter) {
     if (AccessLogConfiguration.INSTANCE.getAccessLogEnabled()) {
       String pattern = AccessLogConfiguration.INSTANCE.getAccesslogPattern();
       LOGGER.info("access log enabled, pattern = {}", pattern);
       mainRouter.route()
           .handler(new AccessLogHandler(
               pattern
           ));
     }
   }

   /**
    * Support CORS
    */
   void mountCorsHandler(Router mainRouter) {
     if (!TransportConfig.isCorsEnabled()) {
       return;
     }

     CorsHandler corsHandler = getCorsHandler(TransportConfig.getCorsAllowedOrigin());
     // Access-Control-Allow-Credentials
     corsHandler.allowCredentials(TransportConfig.isCorsAllowCredentials());
     // Access-Control-Allow-Headers
     corsHandler.allowedHeaders(TransportConfig.getCorsAllowedHeaders());
     // Access-Control-Allow-Methods
     Set<String> allowedMethods = TransportConfig.getCorsAllowedMethods();
     for (String method : allowedMethods) {
       corsHandler.allowedMethod(HttpMethod.valueOf(method));
     }
     // Access-Control-Expose-Headers
     corsHandler.exposedHeaders(TransportConfig.getCorsExposedHeaders());
     // Access-Control-Max-Age
     int maxAge = TransportConfig.getCorsMaxAge();
     if (maxAge >= 0) {
       corsHandler.maxAgeSeconds(maxAge);
     }

     LOGGER.info("mount CorsHandler");
     mainRouter.route().handler(corsHandler);
   }

   private CorsHandler getCorsHandler(String corsAllowedOrigin) {
     return CorsHandler.create(corsAllowedOrigin);
   }

   private void initDispatcher(Router mainRouter) {
     List<VertxHttpDispatcher> dispatchers = SPIServiceUtils.getSortedService(VertxHttpDispatcher.class);
     for (VertxHttpDispatcher dispatcher : dispatchers) {
       if (dispatcher.enabled()) {
         dispatcher.init(mainRouter);
       }
     }
   }

   private void startListen(HttpServer server, Promise<Void> startPromise) {
     server.listen(endpointObject.getPort(), endpointObject.getHostOrIp(), ar -> {
       if (ar.succeeded()) {
         LOGGER.info("rest listen success. address={}:{}",
             endpointObject.getHostOrIp(),
             ar.result().actualPort());
         startPromise.complete();
         return;
       }

       String msg = String.format("rest listen failed, address=%s:%d",
           endpointObject.getHostOrIp(),
           endpointObject.getPort());
       LOGGER.error(msg, ar.cause());
       startPromise.fail(ar.cause());
     });
   }

   private HttpServer createHttpServer() {
     HttpServerOptions serverOptions = createDefaultHttpServerOptions();
     return vertx.createHttpServer(serverOptions);
   }

   private HttpServerOptions createDefaultHttpServerOptions() {
     HttpServerOptions serverOptions = new HttpServerOptions();
     serverOptions.setUsePooledBuffers(true);
     serverOptions.setIdleTimeout(TransportConfig.getConnectionIdleTimeoutInSeconds());
     serverOptions.setCompressionSupported(TransportConfig.getCompressed());
     serverOptions.setMaxHeaderSize(TransportConfig.getMaxHeaderSize());
     serverOptions.setMaxFormAttributeSize(TransportConfig.getMaxFormAttributeSize());
     serverOptions.setCompressionLevel(TransportConfig.getCompressionLevel());
     serverOptions.setMaxChunkSize(TransportConfig.getMaxChunkSize());
     serverOptions.setDecompressionSupported(TransportConfig.getDecompressionSupported());
     serverOptions.setDecoderInitialBufferSize(TransportConfig.getDecoderInitialBufferSize());
     serverOptions.setHttp2ConnectionWindowSize(TransportConfig.getHttp2ConnectionWindowSize());
     serverOptions.setMaxInitialLineLength(TransportConfig.getMaxInitialLineLength());
     if (endpointObject.isHttp2Enabled()) {
       serverOptions.setUseAlpn(TransportConfig.getUseAlpn())
           .setInitialSettings(new Http2Settings().setPushEnabled(TransportConfig.getPushEnabled())
               .setMaxConcurrentStreams(TransportConfig.getMaxConcurrentStreams())
               .setHeaderTableSize(TransportConfig.getHttp2HeaderTableSize())
               .setInitialWindowSize(TransportConfig.getInitialWindowSize())
               .setMaxFrameSize(TransportConfig.getMaxFrameSize())
               .setMaxHeaderListSize(TransportConfig.getMaxHeaderListSize())
           );
     }
     if (endpointObject.isSslEnabled()) {
       SSLOptionFactory factory =
           SSLOptionFactory.createSSLOptionFactory(SSL_KEY, null);
       SSLOption sslOption;
       if (factory == null) {
         sslOption = SSLOption.buildFromYaml(SSL_KEY);
       } else {
         sslOption = factory.createSSLOption();
       }
       SSLCustom sslCustom = SSLCustom.createSSLCustom(sslOption.getSslCustomClass());
       VertxTLSBuilder.buildNetServerOptions(sslOption, sslCustom, serverOptions);
     }

     return serverOptions;
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.servicecomb.transport.rest.vertx;

	import java.nio.channels.ClosedChannelException;
	import java.util.List;
	import java.util.Set;

	import javax.ws.rs.core.MediaType;

	import org.apache.servicecomb.common.rest.codec.RestObjectMapperFactory;
	import org.apache.servicecomb.core.Endpoint;
	import org.apache.servicecomb.core.transport.AbstractTransport;
	import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
	import org.apache.servicecomb.foundation.common.utils.ExceptionUtils;
	import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
	import org.apache.servicecomb.foundation.ssl.SSLCustom;
	import org.apache.servicecomb.foundation.ssl.SSLOption;
	import org.apache.servicecomb.foundation.ssl.SSLOptionFactory;
	import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
	import org.apache.servicecomb.foundation.vertx.metrics.DefaultHttpServerMetrics;
	import org.apache.servicecomb.foundation.vertx.metrics.metric.DefaultServerEndpointMetric;
	import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
	import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
	import org.apache.servicecomb.transport.rest.vertx.accesslog.AccessLogConfiguration;
	import org.apache.servicecomb.transport.rest.vertx.accesslog.impl.AccessLogHandler;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.http.HttpHeaders;

	import com.netflix.config.DynamicPropertyFactory;

	import io.vertx.core.AbstractVerticle;
	import io.vertx.core.Context;
	import io.vertx.core.Handler;
	import io.vertx.core.Promise;
	import io.vertx.core.Vertx;
	import io.vertx.core.http.Http2Settings;
	import io.vertx.core.http.HttpMethod;
	import io.vertx.core.http.HttpServer;
	import io.vertx.core.http.HttpServerOptions;
	import io.vertx.core.http.HttpServerResponse;
	import io.vertx.core.net.impl.ConnectionBase;
	import io.vertx.ext.web.Router;
	import io.vertx.ext.web.RoutingContext;
	import io.vertx.ext.web.handler.CorsHandler;

	public class RestServerVerticle extends AbstractVerticle {
	private static final Logger LOGGER = LoggerFactory.getLogger(RestServerVerticle.class);

	private static final String SSL_KEY = "rest.provider";

	private Endpoint endpoint;

	private URIEndpointObject endpointObject;

	@Override
	public void init(Vertx vertx, Context context) {
	super.init(vertx, context);
	this.endpoint = (Endpoint) context.config().getValue(AbstractTransport.ENDPOINT_KEY);
	this.endpointObject = (URIEndpointObject) endpoint.getAddress();
	}

	@Override
	public void start(Promise<Void> startPromise) throws Exception {
	try {
	super.start();
	// 如果本地未配置地址，则表示不必监听，只需要作为客户端使用即可
	if (endpointObject == null) {
	LOGGER.warn("rest listen address is not configured, will not start.");
	startPromise.complete();
	return;
	}
	Router mainRouter = Router.router(vertx);
	mountAccessLogHandler(mainRouter);
	mountCorsHandler(mainRouter);
	initDispatcher(mainRouter);
	mountGlobalRestFailureHandler(mainRouter);
	HttpServer httpServer = createHttpServer();
	httpServer.requestHandler(mainRouter);
	httpServer.connectionHandler(connection -> {
	DefaultHttpServerMetrics serverMetrics = (DefaultHttpServerMetrics) ((ConnectionBase) connection).metrics();
	DefaultServerEndpointMetric endpointMetric = serverMetrics.getEndpointMetric();
	long connectedCount = endpointMetric.getCurrentConnectionCount();
	int connectionLimit = DynamicPropertyFactory.getInstance()
	.getIntProperty("servicecomb.rest.server.connection-limit", Integer.MAX_VALUE).get();
	if (connectedCount > connectionLimit) {
	connection.close();
	endpointMetric.onRejectByConnectionLimit();
	}
	});
	httpServer.exceptionHandler(e -> {
	if (e instanceof ClosedChannelException) {
	// This is quite normal in between browser and ege, so do not print out.
	LOGGER.debug("Unexpected error in server.{}", ExceptionUtils.getExceptionMessageWithoutTrace(e));
	} else {
	LOGGER.error("Unexpected error in server.{}", ExceptionUtils.getExceptionMessageWithoutTrace(e));
	}
	});
	startListen(httpServer, startPromise);
	} catch (Throwable e) {
	// vert.x got some states that not print error and execute call back in VertexUtils.blockDeploy, we add a log our self.
	LOGGER.error("", e);
	throw e;
	}
	}

	private void mountGlobalRestFailureHandler(Router mainRouter) {
	GlobalRestFailureHandler globalRestFailureHandler =
	SPIServiceUtils.getPriorityHighestService(GlobalRestFailureHandler.class);
	Handler<RoutingContext> failureHandler = null == globalRestFailureHandler ?
	ctx -> {
	if (ctx.response().closed() \|\| ctx.response().ended()) {
	// response has been sent, do nothing
	LOGGER.error("get a failure with closed response", ctx.failure());
	return;
	}
	HttpServerResponse response = ctx.response();
	if (ctx.failure() instanceof InvocationException) {
	// ServiceComb defined exception
	InvocationException exception = (InvocationException) ctx.failure();
	response.setStatusCode(exception.getStatusCode());
	response.setStatusMessage(exception.getReasonPhrase());
	response.end(exception.getErrorData().toString());
	return;
	}

	LOGGER.error("unexpected failure happened", ctx.failure());
	try {
	// unknown exception
	CommonExceptionData unknownError = new CommonExceptionData("unknown error");
	ctx.response().setStatusCode(500).putHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON)
	.end(RestObjectMapperFactory.getRestObjectMapper().writeValueAsString(unknownError));
	} catch (Exception e) {
	LOGGER.error("failed to send error response!", e);
	}
	}
	: globalRestFailureHandler;

	mainRouter.route()
	// this handler does nothing, just ensure the failure handler can catch exception
	.handler(RoutingContext::next)
	.failureHandler(failureHandler);
	}

	private void mountAccessLogHandler(Router mainRouter) {
	if (AccessLogConfiguration.INSTANCE.getAccessLogEnabled()) {
	String pattern = AccessLogConfiguration.INSTANCE.getAccesslogPattern();
	LOGGER.info("access log enabled, pattern = {}", pattern);
	mainRouter.route()
	.handler(new AccessLogHandler(
	pattern
	));
	}
	}

	/**
	* Support CORS
	*/
	void mountCorsHandler(Router mainRouter) {
	if (!TransportConfig.isCorsEnabled()) {
	return;
	}

	CorsHandler corsHandler = getCorsHandler(TransportConfig.getCorsAllowedOrigin());
	// Access-Control-Allow-Credentials
	corsHandler.allowCredentials(TransportConfig.isCorsAllowCredentials());
	// Access-Control-Allow-Headers
	corsHandler.allowedHeaders(TransportConfig.getCorsAllowedHeaders());
	// Access-Control-Allow-Methods
	Set<String> allowedMethods = TransportConfig.getCorsAllowedMethods();
	for (String method : allowedMethods) {
	corsHandler.allowedMethod(HttpMethod.valueOf(method));
	}
	// Access-Control-Expose-Headers
	corsHandler.exposedHeaders(TransportConfig.getCorsExposedHeaders());
	// Access-Control-Max-Age
	int maxAge = TransportConfig.getCorsMaxAge();
	if (maxAge >= 0) {
	corsHandler.maxAgeSeconds(maxAge);
	}

	LOGGER.info("mount CorsHandler");
	mainRouter.route().handler(corsHandler);
	}

	private CorsHandler getCorsHandler(String corsAllowedOrigin) {
	return CorsHandler.create(corsAllowedOrigin);
	}

	private void initDispatcher(Router mainRouter) {
	List<VertxHttpDispatcher> dispatchers = SPIServiceUtils.getSortedService(VertxHttpDispatcher.class);
	for (VertxHttpDispatcher dispatcher : dispatchers) {
	if (dispatcher.enabled()) {
	dispatcher.init(mainRouter);
	}
	}
	}

	private void startListen(HttpServer server, Promise<Void> startPromise) {
	server.listen(endpointObject.getPort(), endpointObject.getHostOrIp(), ar -> {
	if (ar.succeeded()) {
	LOGGER.info("rest listen success. address={}:{}",
	endpointObject.getHostOrIp(),
	ar.result().actualPort());
	startPromise.complete();
	return;
	}

	String msg = String.format("rest listen failed, address=%s:%d",
	endpointObject.getHostOrIp(),
	endpointObject.getPort());
	LOGGER.error(msg, ar.cause());
	startPromise.fail(ar.cause());
	});
	}

	private HttpServer createHttpServer() {
	HttpServerOptions serverOptions = createDefaultHttpServerOptions();
	return vertx.createHttpServer(serverOptions);
	}

	private HttpServerOptions createDefaultHttpServerOptions() {
	HttpServerOptions serverOptions = new HttpServerOptions();
	serverOptions.setUsePooledBuffers(true);
	serverOptions.setIdleTimeout(TransportConfig.getConnectionIdleTimeoutInSeconds());
	serverOptions.setCompressionSupported(TransportConfig.getCompressed());
	serverOptions.setMaxHeaderSize(TransportConfig.getMaxHeaderSize());
	serverOptions.setMaxFormAttributeSize(TransportConfig.getMaxFormAttributeSize());
	serverOptions.setCompressionLevel(TransportConfig.getCompressionLevel());
	serverOptions.setMaxChunkSize(TransportConfig.getMaxChunkSize());
	serverOptions.setDecompressionSupported(TransportConfig.getDecompressionSupported());
	serverOptions.setDecoderInitialBufferSize(TransportConfig.getDecoderInitialBufferSize());
	serverOptions.setHttp2ConnectionWindowSize(TransportConfig.getHttp2ConnectionWindowSize());
	serverOptions.setMaxInitialLineLength(TransportConfig.getMaxInitialLineLength());
	if (endpointObject.isHttp2Enabled()) {
	serverOptions.setUseAlpn(TransportConfig.getUseAlpn())
	.setInitialSettings(new Http2Settings().setPushEnabled(TransportConfig.getPushEnabled())
	.setMaxConcurrentStreams(TransportConfig.getMaxConcurrentStreams())
	.setHeaderTableSize(TransportConfig.getHttp2HeaderTableSize())
	.setInitialWindowSize(TransportConfig.getInitialWindowSize())
	.setMaxFrameSize(TransportConfig.getMaxFrameSize())
	.setMaxHeaderListSize(TransportConfig.getMaxHeaderListSize())
	);
	}
	if (endpointObject.isSslEnabled()) {
	SSLOptionFactory factory =
	SSLOptionFactory.createSSLOptionFactory(SSL_KEY, null);
	SSLOption sslOption;
	if (factory == null) {
	sslOption = SSLOption.buildFromYaml(SSL_KEY);
	} else {
	sslOption = factory.createSSLOption();
	}
	SSLCustom sslCustom = SSLCustom.createSSLCustom(sslOption.getSslCustomClass());
	VertxTLSBuilder.buildNetServerOptions(sslOption, sslCustom, serverOptions);
	}

	return serverOptions;
	}
	}