flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
/**
	 * Starts this REST server endpoint.
	 *
	 * @throws Exception if we cannot start the RestServerEndpoint
	 */
Starts this REST server endpoint.
@throws Exception if we cannot start the RestServerEndpoint
Starts this REST server endpoint . @throws Exception if we cannot start the RestServerEndpoint
public void start() throws Exception {
		synchronized (lock) {
			if (started) {
				return;
			}
			log.info("Starting rest endpoint.");
			final Router router = new Router();
			final CompletableFuture<String> restAddressFuture = new CompletableFuture<>();
			initializeHandlers(restAddressFuture).forEach(handler -> registerHandler(router, handler));
			ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
				@Override
				protected void initChannel(SocketChannel ch) {
					Handler handler = new RouterHandler(router);
					if (sslEngine != null) {
						ch.pipeline().addLast("ssl", new SslHandler(sslEngine));
					}
					ch.pipeline()
						.addLast(new HttpServerCodec())
						.addLast(new HttpObjectAggregator(MAX_REQUEST_SIZE_BYTES))
						.addLast(handler.name(), handler)
						.addLast(new PipelineErrorHandler(log));
				}
			};
			NioEventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("flink-rest-server-netty-boss"));
			NioEventLoopGroup workerGroup = new NioEventLoopGroup(0, new DefaultThreadFactory("flink-rest-server-netty-worker"));
			bootstrap = new ServerBootstrap();
			bootstrap
				.group(bossGroup, workerGroup)
				.channel(NioServerSocketChannel.class)
				.childHandler(initializer);
			final ChannelFuture channel;
			if (configuredAddress == null) {
				channel = bootstrap.bind(configuredPort);
			} else {
				channel = bootstrap.bind(configuredAddress, configuredPort);
			}
			serverChannel = channel.syncUninterruptibly().channel();
			InetSocketAddress bindAddress = (InetSocketAddress) serverChannel.localAddress();
			String address = bindAddress.getAddress().getHostAddress();
			int port = bindAddress.getPort();
			log.info("Rest endpoint listening at {}" + ':' + "{}", address, port);
			final String protocol;
			if (sslEngine != null) {
				protocol = "https://";
			} else {
				protocol = "http://";
			}
			restAddress = protocol + address + ':' + port;
			restAddressFuture.complete(restAddress);
			started = true;
		}
	}
public void start ( ) throws Exception { synchronized ( lock ) { if ( started ) { return ; } log . info ( "Starting rest endpoint." ) ; final Router router = new Router ( ) ; final CompletableFuture < String > restAddressFuture = new CompletableFuture < > ( ) ; initializeHandlers ( restAddressFuture ) . forEach ( handler -> registerHandler ( router , handler ) ) ; ChannelInitializer < SocketChannel > initializer = new ChannelInitializer < SocketChannel > ( ) { @ Override protected void initChannel ( SocketChannel ch ) { Handler handler = new RouterHandler ( router ) ; if ( sslEngine != null ) { ch . pipeline ( ) . addLast ( "ssl" , new SslHandler ( sslEngine ) ) ; } ch . pipeline ( ) . addLast ( new HttpServerCodec ( ) ) . addLast ( new HttpObjectAggregator ( MAX_REQUEST_SIZE_BYTES ) ) . addLast ( handler . name ( ) , handler ) . addLast ( new PipelineErrorHandler ( log ) ) ; } } ; NioEventLoopGroup bossGroup = new NioEventLoopGroup ( 1 , new DefaultThreadFactory ( "flink-rest-server-netty-boss" ) ) ; NioEventLoopGroup workerGroup = new NioEventLoopGroup ( 0 , new DefaultThreadFactory ( "flink-rest-server-netty-worker" ) ) ; bootstrap = new ServerBootstrap ( ) ; bootstrap . group ( bossGroup , workerGroup ) . channel ( NioServerSocketChannel . class ) . childHandler ( initializer ) ; final ChannelFuture channel ; if ( configuredAddress == null ) { channel = bootstrap . bind ( configuredPort ) ; } else { channel = bootstrap . bind ( configuredAddress , configuredPort ) ; } serverChannel = channel . syncUninterruptibly ( ) . channel ( ) ; InetSocketAddress bindAddress = ( InetSocketAddress ) serverChannel . localAddress ( ) ; String address = bindAddress . getAddress ( ) . getHostAddress ( ) ; int port = bindAddress . getPort ( ) ; log . info ( "Rest endpoint listening at {}" + ':' + "{}" , address , port ) ; final String protocol ; if ( sslEngine != null ) { protocol = "https://" ; } else { protocol = "http://" ; } restAddress = protocol + address + ':' + port ; restAddressFuture . complete ( restAddress ) ; started = true ; } }
Starts this REST server endpoint.
Starts this REST server endpoint .
// RestServerEndpoint already started // SSL should be the first handler in the pipeline