import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Override
public Mono<? extends NettyContext> newHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler) {
	final BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>>
			targetHandler =
			null == handler ? ChannelOperations.noopHandler() : handler;

	return Mono.create(sink -> {
		Bootstrap b = options.get();
		SocketAddress adr = options.getAddress();
		if(adr == null){
			sink.error(new NullPointerException("Provided UdpServerOptions do not " +
					"define any address to bind to "));
			return;
		}
		b.localAddress(adr);
		ContextHandler<DatagramChannel> c = doHandler(targetHandler, sink, adr);
		b.handler(c);
		c.setFuture(b.bind());
	});
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Override
public Mono<? extends NettyContext> newHandler(BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>> handler) {
	final BiFunction<? super UdpInbound, ? super UdpOutbound, ? extends Publisher<Void>>
			targetHandler =
			null == handler ? ChannelOperations.noopHandler() : handler;

	return Mono.create(sink -> {
		Bootstrap b = options.get();
		SocketAddress adr = options.getAddress();
		if(adr == null){
			sink.error(new NullPointerException("Provided UdpClientOptions do not " +
					"define any address to bind to "));
			return;
		}
		b.remoteAddress(adr);
		ContextHandler<DatagramChannel> c = doHandler(targetHandler, sink, adr);
		b.handler(c);
		c.setFuture(b.connect());
	});
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
/**
 * Create a new client context
 *
 * @param sink
 * @param options
 * @param loggingHandler
 * @param secure
 * @param channelOpFactory
 * @param <CHANNEL>
 *
 * @return a new {@link ContextHandler} for clients
 */
public static <CHANNEL extends Channel> ContextHandler<CHANNEL> newClientContext(
		MonoSink<NettyContext> sink,
		ClientOptions options,
		LoggingHandler loggingHandler,
		boolean secure,
		SocketAddress providedAddress,
		ChannelOperations.OnNew<CHANNEL> channelOpFactory) {
	return newClientContext(sink,
			options,
			loggingHandler,
			secure,
			providedAddress,
			null,
			channelOpFactory);
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
/**
 * @param channelOpFactory
 * @param options
 * @param sink
 * @param loggingHandler
 * @param providedAddress the {@link InetSocketAddress} targeted by the operation
 * associated with that handler (useable eg. for SNI), or null if unavailable.
 */
@SuppressWarnings("unchecked")
protected ContextHandler(ChannelOperations.OnNew<CHANNEL> channelOpFactory,
		NettyOptions<?, ?> options,
		MonoSink<NettyContext> sink,
		LoggingHandler loggingHandler,
		SocketAddress providedAddress) {
	this.channelOpFactory =
			Objects.requireNonNull(channelOpFactory, "channelOpFactory");
	this.options = options;
	this.sink = sink;
	this.loggingHandler = loggingHandler;
	this.autoCreateOperations = true;
	this.providedAddress = providedAddress;

}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
final void release(CHANNEL c) {
	if (log.isDebugEnabled()) {
		log.debug("Releasing channel: {}", c.toString());
	}

	if (!NettyContext.isPersistent(c) && c.isActive()) {
		c.close();
	}

	pool.release(c)
	    .addListener(f -> {
		    if (f.isSuccess()) {
			    onReleaseEmitter.onComplete();
		    }
		    else {
			    onReleaseEmitter.onError(f.cause());
		    }
	    });

}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p>
 * This adds a ChunkedWriter to the pipeline to extract chunks from the
 * {@link io.netty.handler.stream.ChunkedInput} that the strategy produces. This step
 * is skipped if the handler is already present, and the placement of the handler
 * depends on the presence of the ReactiveBridge handler (see {@link NettyPipeline}).
 *
 * @param context the context from which to obtain the channel and pipeline
 */
@Override
public final void preparePipeline(NettyContext context) {
	this.addHandler = context.channel()
	                         .pipeline()
	                         .get(NettyPipeline.ChunkedWriter) == null;
	if (addHandler) {
		boolean hasReactiveBridge = context.channel()
		                                   .pipeline()
		                                   .get(NettyPipeline.ReactiveBridge) != null;

		if (hasReactiveBridge) {
			context.channel()
			       .pipeline()
			       .addBefore(NettyPipeline.ReactiveBridge,
					       NettyPipeline.ChunkedWriter,
					       new ChunkedWriteHandler());
		}
		else {
			context.channel()
			       .pipeline()
			       .addLast(NettyPipeline.ChunkedWriter, new ChunkedWriteHandler());
		}
	}
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void testIssue186() {
	NettyContext server =
			HttpServer.create(0)
			          .newHandler((req, res) -> res.status(200).send())
			          .block(Duration.ofSeconds(300));

	HttpClient client =
			HttpClient.create(ops -> ops.connectAddress(() -> server.address())
					                    .poolResources(PoolResources.fixed("test", 1)));

	try {
		doTestIssue186(client);
		doTestIssue186(client);
	}
	finally {
		server.dispose();
	}

}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
/**
 * Create a {@link ContextHandler} for {@link Bootstrap#handler()}
 *
 * @param handler user provided in/out handler
 * @param sink user provided bind handler
 * @param secure if operation should be secured
 * @param pool if channel pool
 * @param onSetup if operation has local setup callback
 *
 * @return a new {@link ContextHandler}
 */
protected ContextHandler<SocketChannel> doHandler(BiFunction<? super NettyInbound, ? super NettyOutbound, ? extends Publisher<Void>> handler,
		MonoSink<NettyContext> sink,
		boolean secure,
		SocketAddress providedAddress,
		ChannelPool pool,
		Consumer<? super Channel> onSetup) {
	return ContextHandler.newClientContext(sink,
			options,
			loggingHandler,
			secure,
			providedAddress,
			pool,
			handler == null ? EMPTY :
					(ch, c, msg) -> ChannelOperations.bind(ch, handler, c));
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Override
protected ContextHandler<SocketChannel> doHandler(BiFunction<? super NettyInbound, ? super NettyOutbound, ? extends Publisher<Void>> handler,
		MonoSink<NettyContext> sink,
		boolean secure,
		SocketAddress providedAddress,
		ChannelPool pool,
		Consumer<? super Channel> onSetup) {
	return ContextHandler.<SocketChannel>newClientContext(sink,
			options,
			loggingHandler,
			secure,
			providedAddress,
			pool,
			handler != null ? (ch, c, msg) -> {
				if(onSetup != null){
					onSetup.accept(ch);
				}
				return HttpClientOperations.bindHttp(ch, handler, c);
			} : EMPTY).onPipeline(this);
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Override
public WebsocketOutbound sendWebsocket(String subprotocols) {
	Mono<Void> m = withWebsocketSupport(websocketUri(), subprotocols, noopHandler());

	return new WebsocketOutbound() {

		@Override
		public String selectedSubprotocol() {
			return null;
		}

		@Override
		public NettyContext context() {
			return HttpClientOperations.this;
		}

		@Override
		public Mono<Void> then() {
			return m;
		}
	};
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void nonContentStatusCodes() {
	NettyContext server =
			HttpServer.create(ops -> ops.host("localhost"))
			          .newRouter(r -> r.get("/204-1", (req, res) -> res.status(HttpResponseStatus.NO_CONTENT)
			                                                           .sendHeaders())
			                           .get("/204-2", (req, res) -> res.status(HttpResponseStatus.NO_CONTENT))
			                           .get("/205-1", (req, res) -> res.status(HttpResponseStatus.RESET_CONTENT)
			                                                           .sendHeaders())
			                           .get("/205-2", (req, res) -> res.status(HttpResponseStatus.RESET_CONTENT))
			                           .get("/304-1", (req, res) -> res.status(HttpResponseStatus.NOT_MODIFIED)
			                                                           .sendHeaders())
			                           .get("/304-2", (req, res) -> res.status(HttpResponseStatus.NOT_MODIFIED)))
			          .block(Duration.ofSeconds(30));

	checkResponse("/204-1", server.address());
	checkResponse("/204-2", server.address());
	checkResponse("/205-1", server.address());
	checkResponse("/205-2", server.address());
	checkResponse("/304-1", server.address());
	checkResponse("/304-2", server.address());

	server.dispose();
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
static void autoAddHttpExtractor(NettyContext c, String name, ChannelHandler
		handler){

	if (handler instanceof ByteToMessageDecoder
			|| handler instanceof ByteToMessageCodec
			|| handler instanceof CombinedChannelDuplexHandler) {
		String extractorName = name+"$extractor";

		if(c.channel().pipeline().context(extractorName) != null){
			return;
		}

		c.channel().pipeline().addBefore(name, extractorName, HTTP_EXTRACTOR);

		if(NettyContext.isPersistent(c.channel())){
			c.onClose(() -> c.removeHandler(extractorName));
		}

	}
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void testTcpClient() throws InterruptedException {
	final CountDownLatch latch = new CountDownLatch(1);

	NettyContext client = TcpClient.create("localhost", echoServerPort)
	                               .newHandler((in, out) -> {
		                               in.receive()
		                                 .log("conn")
		                                 .subscribe(s -> latch.countDown());

		                               return out.sendString(Flux.just("Hello World!"))
		                                  .neverComplete();
	                               })
	                               .block(Duration.ofSeconds(30));

	latch.await(30, TimeUnit.SECONDS);

	client.dispose();

	assertThat("latch was counted down", latch.getCount(), is(0L));
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void testTcpClientWithInetSocketAddress() throws InterruptedException {
	final CountDownLatch latch = new CountDownLatch(1);

	TcpClient client =
			TcpClient.create(echoServerPort);

	NettyContext s = client.newHandler((in, out) -> {
		in.receive()
		  .subscribe(d -> latch.countDown());

		return out.sendString(Flux.just("Hello"))
		   .neverComplete();
	})
	                       .block(Duration.ofSeconds(5));

	latch.await(5, TimeUnit.SECONDS);

	s.dispose();

	assertThat("latch was counted down", latch.getCount(), is(0L));
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void readIdleDoesNotFireWhileDataIsBeingRead()
		throws InterruptedException, IOException {
	final CountDownLatch latch = new CountDownLatch(1);
	long start = System.currentTimeMillis();

	TcpClient client = TcpClient.create("localhost", heartbeatServerPort);

	NettyContext s = client.newHandler((in, out) -> {
		in.onReadIdle(500, latch::countDown);
		return Flux.never();
	})
	                       .block(Duration.ofSeconds(30));

	assertTrue(latch.await(15, TimeUnit.SECONDS));
	heartbeatServer.close();

	long duration = System.currentTimeMillis() - start;

	assertThat(duration, is(greaterThanOrEqualTo(500L)));
	s.dispose();
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void flushOnComplete() {

	Flux<String> test = Flux.range(0, 100)
	                        .map(n -> String.format("%010d", n));

	NettyContext c = HttpServer.create(0)
	                           .newHandler((req, resp) -> resp.sendString(test.map(s -> s + "\n")))
	                           .block(Duration.ofSeconds(30));

	Flux<String> client = HttpClient.create(c.address()
	                                         .getPort())
	                                .get("/")
	                                .block(Duration.ofSeconds(30))
	                                .addHandler(new LineBasedFrameDecoder(10))
	                                .receive()
	                                .asString();

	StepVerifier.create(client)
	            .expectNextSequence(test.toIterable())
	            .expectComplete()
	            .verify(Duration.ofSeconds(30));

	c.dispose();
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
	public void prematureCancel() throws Exception {
		DirectProcessor<Void> signal = DirectProcessor.create();
		NettyContext x = TcpServer.create("localhost", 0)
		                          .newHandler((in, out) -> {
										signal.onComplete();
										return out.context(c -> c.addHandlerFirst(
												new HttpResponseEncoder()))
										          .sendObject(Mono.delay(Duration
												          .ofSeconds(2))
												          .map(t ->
												          new DefaultFullHttpResponse(
														          HttpVersion.HTTP_1_1,
														          HttpResponseStatus
																          .PROCESSING)))
												.neverComplete();
		                          })
		                          .block(Duration.ofSeconds(30));

		StepVerifier.create(createHttpClientForContext(x)
		                              .get("/")
		                              .timeout(signal)
		)
		            .verifyError(TimeoutException.class);
//		Thread.sleep(1000000);
	}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
private void doTestGzip(boolean gzipEnabled) {
	String expectedResponse = gzipEnabled ? "gzip" : "no gzip";
	NettyContext server = HttpServer.create(0)
	        .newHandler((req,res) -> res.sendString(
	                Mono.just(req.requestHeaders().get(HttpHeaderNames.ACCEPT_ENCODING, "no gzip"))))
	        .block(Duration.ofSeconds(30));
	StepVerifier.create(
	        HttpClient.create(ops -> ops.port(server.address().getPort()).compression(gzipEnabled))
	                  .get("/")
	                  .flatMap(r -> r.receive()
	                                 .asString()
	                                 .elementAt(0)
	                                 .zipWith(Mono.just(r)))
	        )
	            .expectNextMatches(tuple -> {
	                tuple.getT2().dispose();
	                return expectedResponse.equals(tuple.getT1());
	            })
	            .expectComplete()
	            .verify(Duration.ofSeconds(30));

	server.dispose();
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
@Test
public void sshExchangeAbsoluteGet() throws CertificateException, SSLException {
	SelfSignedCertificate ssc = new SelfSignedCertificate();
	SslContext sslServer = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
	SslContext sslClient = SslContextBuilder.forClient()
	                                        .trustManager(ssc.cert()).build();

	NettyContext context =
			HttpServer.create(opt -> opt.sslContext(sslServer))
			          .newHandler((req, resp) -> resp.sendString(Flux.just("hello ", req.uri())))
			          .block();

	HttpClientResponse response = HttpClient.create(
			opt -> applyHostAndPortFromContext(opt, context)
					.sslContext(sslClient))
			.get("/foo").block();
	context.dispose();
	context.onClose().block();

	String responseString = response.receive().aggregate().asString(CharsetUtil.UTF_8).block();
	assertThat(responseString).isEqualTo("hello /foo");
}
 

import reactor.ipc.netty.NettyContext; //导入依赖的package包/类
private void assertSendFile(Function<HttpServerResponse, NettyOutbound> fn) {
	NettyContext context =
			HttpServer.create(opt -> opt.host("localhost"))
			          .newHandler((req, resp) -> fn.apply(resp))
			          .block();


	HttpClientResponse response =
			HttpClient.create(opt -> opt.connectAddress(() -> context.address()))
			          .get("/foo")
			          .block(Duration.ofSeconds(120));

	context.dispose();
	context.onClose().block();

	String body = response.receive().aggregate().asString(StandardCharsets.UTF_8).block();

	assertThat(body)
			.startsWith("This is an UTF-8 file that is larger than 1024 bytes. " + "It contains accents like é.")
			.contains("1024 mark here -><- 1024 mark here")
			.endsWith("End of File");
}