From 20c401109473a22e6d1bfe638374e885a372d0f8 Mon Sep 17 00:00:00 2001 From: Stephane Maldini Date: Tue, 29 Sep 2015 14:07:36 +0100 Subject: [PATCH] initialize reactor support, not passing requestmapping tests yet --- spring-web-reactive/build.gradle | 2 + .../reactor/ReactorServerHttpRequest.java | 79 +++++++++++++++++++ .../reactor/ReactorServerHttpResponse.java | 74 +++++++++++++++++ .../http/reactor/RequestHandlerAdapter.java | 44 +++++++++++ .../AbstractHttpHandlerIntegrationTests.java | 1 + .../reactive/web/http/ReactorHttpServer.java | 75 ++++++++++++++++++ 6 files changed, 275 insertions(+) create mode 100644 spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpRequest.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java create mode 100644 spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/RequestHandlerAdapter.java create mode 100644 spring-web-reactive/src/test/java/org/springframework/reactive/web/http/ReactorHttpServer.java diff --git a/spring-web-reactive/build.gradle b/spring-web-reactive/build.gradle index 5619827d0b..5ab432e69e 100644 --- a/spring-web-reactive/build.gradle +++ b/spring-web-reactive/build.gradle @@ -34,6 +34,8 @@ dependencies { optional "io.reactivex:rxnetty:0.5.0-SNAPSHOT" optional "io.reactivex:rxjava-reactive-streams:1.0.1" + optional "io.projectreactor:reactor-net:2.0.5.RELEASE" + provided "javax.servlet:javax.servlet-api:3.1.0" testCompile "junit:junit:4.12" diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpRequest.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpRequest.java new file mode 100644 index 0000000000..6a9d0ed44d --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpRequest.java @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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.springframework.reactive.web.http.reactor; + +import org.reactivestreams.Publisher; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.reactive.web.http.ServerHttpRequest; +import org.springframework.util.Assert; +import reactor.io.buffer.Buffer; +import reactor.io.net.http.HttpChannel; + +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.ByteBuffer; + +/** + * @author Stephane Maldini + */ +public class ReactorServerHttpRequest implements ServerHttpRequest { + + private final HttpChannel channel; + + private HttpHeaders headers; + + + public ReactorServerHttpRequest(HttpChannel request) { + Assert.notNull("'request', request must not be null."); + this.channel = request; + } + + + @Override + public HttpHeaders getHeaders() { + if (this.headers == null) { + this.headers = new HttpHeaders(); + for (String name : this.channel.headers().names()) { + for (String value : this.channel.headers().getAll(name)) { + this.headers.add(name, value); + } + } + } + return this.headers; + } + + @Override + public HttpMethod getMethod() { + return HttpMethod.valueOf(this.channel.method().getName()); + } + + @Override + public URI getURI() { + try { + return new URI(this.channel.uri()); + } catch (URISyntaxException ex) { + throw new IllegalStateException("Could not get URI: " + ex.getMessage(), ex); + } + + } + + @Override + public Publisher getBody() { + return this.channel.map(Buffer::byteBuffer); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java new file mode 100644 index 0000000000..647994c6bb --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java @@ -0,0 +1,74 @@ +/* + * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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.springframework.reactive.web.http.reactor; + +import org.reactivestreams.Publisher; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.reactive.web.http.ServerHttpResponse; +import org.springframework.util.Assert; +import reactor.io.buffer.Buffer; +import reactor.io.net.http.HttpChannel; +import reactor.io.net.http.model.Status; +import reactor.rx.Streams; + +import java.nio.ByteBuffer; + +/** + * @author Stephane Maldini + */ +public class ReactorServerHttpResponse implements ServerHttpResponse { + + private final HttpChannel channel; + + private final HttpHeaders headers; + + private boolean headersWritten = false; + + + public ReactorServerHttpResponse(HttpChannel response) { + Assert.notNull("'response', response must not be null."); + this.channel = response; + this.headers = new HttpHeaders(); + } + + + @Override + public void setStatusCode(HttpStatus status) { + this.channel.responseStatus(Status.valueOf(status.value())); + } + + @Override + public HttpHeaders getHeaders() { + return (this.headersWritten ? HttpHeaders.readOnlyHttpHeaders(this.headers) : this.headers); + } + + @Override + public Publisher writeWith(Publisher contentPublisher) { + writeHeaders(); + return this.channel.writeWith(Streams.wrap(contentPublisher).map(Buffer::new)); + } + + private void writeHeaders() { + if (!this.headersWritten) { + for (String name : this.headers.keySet()) { + for (String value : this.headers.get(name)) { + this.channel.responseHeaders().add(name, value); + } + } + } + } +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/RequestHandlerAdapter.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/RequestHandlerAdapter.java new file mode 100644 index 0000000000..c8173299c6 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/RequestHandlerAdapter.java @@ -0,0 +1,44 @@ +/* + * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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.springframework.reactive.web.http.reactor; + +import org.reactivestreams.Publisher; +import org.springframework.reactive.web.http.HttpHandler; +import org.springframework.util.Assert; +import reactor.io.buffer.Buffer; +import reactor.io.net.ReactorChannelHandler; +import reactor.io.net.http.HttpChannel; + +/** + * @author Stephane Maldini + */ +public class RequestHandlerAdapter implements ReactorChannelHandler> { + + private final HttpHandler httpHandler; + + + public RequestHandlerAdapter(HttpHandler httpHandler) { + Assert.notNull(httpHandler, "'httpHandler' is required."); + this.httpHandler = httpHandler; + } + + @Override + public Publisher apply(HttpChannel channel) { + ReactorServerHttpRequest adaptedRequest = new ReactorServerHttpRequest(channel); + ReactorServerHttpResponse adaptedResponse = new ReactorServerHttpResponse(channel); + return this.httpHandler.handle(adaptedRequest, adaptedResponse); + } +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/AbstractHttpHandlerIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/AbstractHttpHandlerIntegrationTests.java index 240c0ff82b..9f0111a6ad 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/AbstractHttpHandlerIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/AbstractHttpHandlerIntegrationTests.java @@ -38,6 +38,7 @@ public abstract class AbstractHttpHandlerIntegrationTests { return new Object[][] { {new JettyHttpServer()}, {new TomcatHttpServer()}, + {new ReactorHttpServer()}, {new RxNettyHttpServer()} }; } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/ReactorHttpServer.java b/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/ReactorHttpServer.java new file mode 100644 index 0000000000..5ed3e75216 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/web/http/ReactorHttpServer.java @@ -0,0 +1,75 @@ +/* + * Copyright (c) 2011-2015 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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.springframework.reactive.web.http; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.reactive.web.http.reactor.RequestHandlerAdapter; +import org.springframework.util.Assert; +import reactor.Environment; +import reactor.bus.selector.Selectors; +import reactor.io.buffer.Buffer; +import reactor.io.net.NetStreams; + + +/** + * @author Stephane Maldini + */ +public class ReactorHttpServer extends HttpServerSupport implements InitializingBean, HttpServer { + + private RequestHandlerAdapter reactorHandler; + + private reactor.io.net.http.HttpServer reactorServer; + + private boolean running; + + + @Override + public boolean isRunning() { + return this.running; + } + + + @Override + public void afterPropertiesSet() throws Exception { + + Assert.notNull(getHttpHandler()); + this.reactorHandler = new RequestHandlerAdapter(getHttpHandler()); + + Environment.initializeIfEmpty(); + + this.reactorServer = (getPort() != -1 ? + NetStreams.httpServer(getPort()) : + NetStreams.httpServer()); + } + + + @Override + public void start() { + if (!this.running) { + this.running = true; + this.reactorServer.route(Selectors.matchAll(), this.reactorHandler).start(); + } + } + + @Override + public void stop() { + if (this.running) { + this.running = false; + this.reactorServer.shutdown(); + } + } + +}