Merge pull request #20 from spring-projects/add-reactor-net

Add reactor net

spring-web-reactive/build.gradle

@@ -20,20 +20,24 @@ group = 'org.springframework.reactive'
 
 repositories {
 	mavenCentral()
+	mavenLocal()
 	maven { url 'https://oss.jfrog.org/libs-snapshot' }  // RxNetty 0.5.x snapshots
+	maven { url 'http://repo.spring.io/snapshot' }  // Reactor snapshot
 }
 
 dependencies {
 	compile "org.springframework:spring-core:4.2.0.RELEASE"
 	compile "org.springframework:spring-web:4.2.0.RELEASE"
 	compile "org.reactivestreams:reactive-streams:1.0.0"
-	compile "io.projectreactor:reactor-stream:2.0.5.RELEASE"
+	compile "io.projectreactor:reactor-stream:2.0.6.BUILD-SNAPSHOT"
 	compile "commons-logging:commons-logging:1.2"
 	compile "com.fasterxml.jackson.core:jackson-databind:2.6.1"
 
 	optional "io.reactivex:rxnetty:0.5.0-SNAPSHOT"
 	optional "io.reactivex:rxjava-reactive-streams:1.0.1"
 
+	optional "io.projectreactor:reactor-net:2.0.6.BUILD-SNAPSHOT"
+
 	provided "javax.servlet:javax.servlet-api:3.1.0"
 
 	testCompile "junit:junit:4.12"

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<Buffer, ?> channel;
+
+	private HttpHeaders headers;
+
+
+	public ReactorServerHttpRequest(HttpChannel<Buffer, ?> 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<ByteBuffer> getBody() {
+		return this.channel.map(Buffer::byteBuffer);
+	}
+
+}

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<?, Buffer> channel;
+
+	private final HttpHeaders headers;
+
+	private boolean headersWritten = false;
+
+
+	public ReactorServerHttpResponse(HttpChannel<?, Buffer> 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<Void> writeWith(Publisher<ByteBuffer> 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);
+				}
+			}
+		}
+	}
+}

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<Buffer, Buffer, HttpChannel<Buffer, Buffer>> {
+
+	private final HttpHandler httpHandler;
+
+
+	public RequestHandlerAdapter(HttpHandler httpHandler) {
+		Assert.notNull(httpHandler, "'httpHandler' is required.");
+		this.httpHandler = httpHandler;
+	}
+
+	@Override
+	public Publisher<Void> apply(HttpChannel<Buffer, Buffer> channel) {
+		ReactorServerHttpRequest adaptedRequest = new ReactorServerHttpRequest(channel);
+		ReactorServerHttpResponse adaptedResponse = new ReactorServerHttpResponse(channel);
+		return this.httpHandler.handle(adaptedRequest, adaptedResponse);
+	}
+}

spring-web-reactive/src/test/java/org/springframework/reactive/web/http/AbstractHttpHandlerIntegrationTests.java

@@ -38,7 +38,8 @@ public abstract class AbstractHttpHandlerIntegrationTests {
 		return new Object[][] {
 				{new JettyHttpServer()},
 				{new TomcatHttpServer()},
-				{new RxNettyHttpServer()}
+				{new RxNettyHttpServer()},
+		        {new ReactorHttpServer()}
 		};
 	}
 

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<Buffer, Buffer> 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();
+		}
+	}
+
+}