Add simple URL mapping and handling

This commit adds support for simple URL handler mapping (exact path
match) and an adapter for the HttpHandler interface to be used to
handle the request.

The SimpleUrlHandlerMappingIntegrationTests then maps the URLs
"/foo" and "/bar" to two different handlers.

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java

@@ -19,21 +19,19 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.reactivestreams.Publisher;
-import reactor.core.reactivestreams.PublisherFactory;
-import reactor.core.reactivestreams.SubscriberWithContext;
 import reactor.rx.Streams;
 
 import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.context.ApplicationContext;
 import org.springframework.http.HttpStatus;
-import org.springframework.reactive.web.http.ServerHttpHandler;
+import org.springframework.reactive.web.http.HttpHandler;
 import org.springframework.reactive.web.http.ServerHttpRequest;
 import org.springframework.reactive.web.http.ServerHttpResponse;
 
 /**
  * @author Rossen Stoyanchev
  */
-public class DispatcherHandler implements ServerHttpHandler {
+public class DispatcherHandler implements HttpHandler {
 
 	private List<HandlerMapping> handlerMappings;
 
@@ -62,7 +60,7 @@ public class DispatcherHandler implements ServerHttpHandler {
 		if (handler == null) {
 			// No exception handling mechanism yet
 			response.setStatusCode(HttpStatus.NOT_FOUND);
-			return PublisherFactory.forEach(SubscriberWithContext::onComplete);
+			return Streams.empty();
 		}
 
 		HandlerAdapter handlerAdapter = getHandlerAdapter(handler);

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/handler/HttpHandlerAdapter.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * 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.dispatch.handler;
+
+import org.reactivestreams.Publisher;
+import reactor.rx.Streams;
+
+import org.springframework.reactive.web.dispatch.HandlerAdapter;
+import org.springframework.reactive.web.dispatch.HandlerResult;
+import org.springframework.reactive.web.http.HttpHandler;
+import org.springframework.reactive.web.http.ServerHttpRequest;
+import org.springframework.reactive.web.http.ServerHttpResponse;
+
+
+/**
+ * Support use of {@link HttpHandler} with
+ * {@link org.springframework.reactive.web.dispatch.DispatcherHandler
+ * DispatcherHandler} (which implements the same contract).
+ * The use of {@code DispatcherHandler} this way enables routing requests to
+ * one of many {@code HttpHandler} instances depending on the configured
+ * handler mappings.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class HttpHandlerAdapter implements HandlerAdapter {
+
+
+	@Override
+	public boolean supports(Object handler) {
+		return HttpHandler.class.isAssignableFrom(handler.getClass());
+	}
+
+	@Override
+	public Publisher<HandlerResult> handle(ServerHttpRequest request, ServerHttpResponse response, Object handler) {
+		HttpHandler httpHandler = (HttpHandler) handler;
+		Publisher<Void> publisher = httpHandler.handle(request, response);
+		return Streams.wrap(publisher).map(aVoid -> null);
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/handler/SimpleUrlHandlerMapping.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * 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.dispatch.handler;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.reactive.web.dispatch.HandlerMapping;
+import org.springframework.reactive.web.http.ServerHttpRequest;
+
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public class SimpleUrlHandlerMapping implements HandlerMapping {
+
+	private final Map<String, Object> handlerMap = new HashMap<>();
+
+
+	public void setHandlers(Map<String, Object> handlers) {
+		this.handlerMap.clear();
+		if (handlers != null) {
+			this.handlerMap.putAll(handlers);
+		}
+	}
+
+
+	@Override
+	public Object getHandler(ServerHttpRequest request) {
+		return this.handlerMap.get(request.getURI().getPath());
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/HttpHandler.java

@@ -18,14 +18,12 @@ package org.springframework.reactive.web.http;
 
 import org.reactivestreams.Publisher;
 
-import org.springframework.reactive.web.http.ServerHttpRequest;
-import org.springframework.reactive.web.http.ServerHttpResponse;
 
 /**
  * @author Arjen Poutsma
  * @author Rossen Stoyanchev
  */
-public interface ServerHttpHandler {
+public interface HttpHandler {
 
 	Publisher<Void> handle(ServerHttpRequest request, ServerHttpResponse response);
 

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/HttpMessage.java

@@ -15,10 +15,7 @@
  */
 package org.springframework.reactive.web.http;
 
-import java.net.URI;
-
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
 
 /**
  * @author Rossen Stoyanchev

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/rxnetty/RequestHandlerAdapter.java

@@ -23,7 +23,7 @@ import org.reactivestreams.Publisher;
 import rx.Observable;
 import rx.RxReactiveStreams;
 
-import org.springframework.reactive.web.http.ServerHttpHandler;
+import org.springframework.reactive.web.http.HttpHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -31,10 +31,10 @@ import org.springframework.util.Assert;
  */
 public class RequestHandlerAdapter implements RequestHandler<ByteBuf, ByteBuf> {
 
-	private final ServerHttpHandler httpHandler;
+	private final HttpHandler httpHandler;
 
 
-	public RequestHandlerAdapter(ServerHttpHandler httpHandler) {
+	public RequestHandlerAdapter(HttpHandler httpHandler) {
 		Assert.notNull(httpHandler, "'httpHandler' is required.");
 		this.httpHandler = httpHandler;
 	}

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/servlet/HttpHandlerServlet.java

@@ -29,7 +29,7 @@ import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-import org.springframework.reactive.web.http.ServerHttpHandler;
+import org.springframework.reactive.web.http.HttpHandler;
 
 /**
  * @author Arjen Poutsma
@@ -43,10 +43,10 @@ public class HttpHandlerServlet extends HttpServlet {
 	private static Log logger = LogFactory.getLog(HttpHandlerServlet.class);
 
 
-	private ServerHttpHandler handler;
+	private HttpHandler handler;
 
 
-	public void setHandler(ServerHttpHandler handler) {
+	public void setHandler(HttpHandler handler) {
 		this.handler = handler;
 	}
 

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/servlet/RequestBodyPublisher.java

@@ -55,7 +55,6 @@ public class RequestBodyPublisher implements ReadListener, Publisher<byte[]> {
 	@Override
 	public void subscribe(Subscriber<? super byte[]> s) {
 		this.subscriber = s;
-
 		this.subscriber.onSubscribe(new RequestBodySubscription());
 	}
 
@@ -99,13 +98,17 @@ public class RequestBodyPublisher implements ReadListener, Publisher<byte[]> {
 	public void onAllDataRead() throws IOException {
 		logger.debug("All data read");
 		this.synchronizer.readComplete();
-		this.subscriber.onComplete();
+		if (this.subscriber != null) {
+			this.subscriber.onComplete();
+		}
 	}
 
 	@Override
 	public void onError(Throwable t) {
 		logger.error("RequestBodyPublisher Error", t);
-		this.subscriber.onError(t);
+		if (this.subscriber != null) {
+			this.subscriber.onError(t);
+		}
 	}
 
 	private class RequestBodySubscription implements Subscription {

spring-web-reactive/src/main/java/org/springframework/reactive/web/http/servlet/ServletServerHttpRequest.java

@@ -20,7 +20,6 @@ import java.net.URISyntaxException;
 import java.nio.charset.Charset;
 import java.util.Enumeration;
 import java.util.Map;
-
 import javax.servlet.http.HttpServletRequest;
 
 import org.reactivestreams.Publisher;

spring-web-reactive/src/test/java/org/springframework/reactive/web/dispatch/SimpleUrlHandlerMappingIntegrationTests.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * 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.dispatch;
+
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.rx.Streams;
+
+import org.springframework.http.RequestEntity;
+import org.springframework.http.ResponseEntity;
+import org.springframework.reactive.web.dispatch.handler.HttpHandlerAdapter;
+import org.springframework.reactive.web.dispatch.handler.SimpleUrlHandlerMapping;
+import org.springframework.reactive.web.http.AbstractHttpHandlerIntegrationTests;
+import org.springframework.reactive.web.http.HttpHandler;
+import org.springframework.reactive.web.http.ServerHttpRequest;
+import org.springframework.reactive.web.http.ServerHttpResponse;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.context.support.StaticWebApplicationContext;
+
+import static org.junit.Assert.assertArrayEquals;
+
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public class SimpleUrlHandlerMappingIntegrationTests extends AbstractHttpHandlerIntegrationTests {
+
+	private static final Charset CHARSET = Charset.forName("UTF-8");
+
+
+	@Override
+	protected HttpHandler createHttpHandler() {
+
+		StaticWebApplicationContext wac = new StaticWebApplicationContext();
+		wac.registerSingleton("hm", TestHandlerMapping.class);
+		wac.registerSingleton("ha", HttpHandlerAdapter.class);
+		wac.refresh();
+
+		DispatcherHandler dispatcherHandler = new DispatcherHandler();
+		dispatcherHandler.initStrategies(wac);
+		return dispatcherHandler;
+	}
+
+	@Test
+	public void testFoo() throws Exception {
+
+		RestTemplate restTemplate = new RestTemplate();
+
+		URI url = new URI("http://localhost:" + port + "/foo");
+		RequestEntity<Void> request = RequestEntity.get(url).build();
+		ResponseEntity<byte[]> response = restTemplate.exchange(request, byte[].class);
+
+		assertArrayEquals("foo".getBytes(CHARSET), response.getBody());
+	}
+
+	@Test
+	public void testBar() throws Exception {
+
+		RestTemplate restTemplate = new RestTemplate();
+
+		URI url = new URI("http://localhost:" + port + "/bar");
+		RequestEntity<Void> request = RequestEntity.get(url).build();
+		ResponseEntity<byte[]> response = restTemplate.exchange(request, byte[].class);
+
+		assertArrayEquals("bar".getBytes(CHARSET), response.getBody());
+	}
+
+
+	private static class TestHandlerMapping extends SimpleUrlHandlerMapping {
+
+		public TestHandlerMapping() {
+			Map<String, Object> map = new HashMap<>();
+			map.put("/foo", new FooHandler());
+			map.put("/bar", new BarHandler());
+			setHandlers(map);
+		}
+	}
+
+	private static class FooHandler implements HttpHandler {
+
+		@Override
+		public Publisher<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
+			return response.writeWith(Streams.just("foo".getBytes(CHARSET)));
+		}
+	}
+
+	private static class BarHandler implements HttpHandler {
+
+		@Override
+		public Publisher<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
+			return response.writeWith(Streams.just("bar".getBytes(CHARSET)));
+		}
+	}
+
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2002-2015 the original author or authors.
+ *
+ * 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.junit.After;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.springframework.util.SocketUtils;
+
+
+@RunWith(Parameterized.class)
+public abstract class AbstractHttpHandlerIntegrationTests {
+
+	protected static int port = SocketUtils.findAvailableTcpPort();
+
+	@Parameterized.Parameter(0)
+	public HttpServer server;
+
+
+	@Parameterized.Parameters(name = "server [{0}]")
+	public static Object[][] arguments() {
+		return new Object[][] {
+				{new JettyHttpServer()},
+				{new TomcatHttpServer()},
+				{new RxNettyHttpServer()}
+		};
+	}
+
+
+	@Before
+	public void setup() throws Exception {
+		this.server.setPort(port);
+		this.server.setHandler(createHttpHandler());
+		this.server.afterPropertiesSet();
+		this.server.start();
+	}
+
+	protected abstract HttpHandler createHttpHandler();
+
+	@After
+	public void tearDown() throws Exception {
+		this.server.stop();
+	}
+
+}

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

@@ -18,14 +18,10 @@ package org.springframework.reactive.web.http;
 
 import org.reactivestreams.Publisher;
 
-import org.springframework.reactive.web.http.ServerHttpHandler;
-import org.springframework.reactive.web.http.ServerHttpRequest;
-import org.springframework.reactive.web.http.ServerHttpResponse;
-
 /**
  * @author Arjen Poutsma
  */
-public class EchoHandler implements ServerHttpHandler {
+public class EchoHandler implements HttpHandler {
 
 	@Override
 	public Publisher<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {

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

@@ -19,56 +19,26 @@ package org.springframework.reactive.web.http;
 import java.net.URI;
 import java.util.Random;
 
-import org.junit.After;
-import org.junit.Before;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
 
 import org.springframework.http.RequestEntity;
 import org.springframework.http.ResponseEntity;
-import org.springframework.util.SocketUtils;
 import org.springframework.web.client.RestTemplate;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 
 
-@RunWith(Parameterized.class)
-public class EchoHandlerIntegrationTests {
+public class EchoHandlerIntegrationTests extends AbstractHttpHandlerIntegrationTests {
 
 	private static final int REQUEST_SIZE = 4096 * 3;
 
-	private static int port = SocketUtils.findAvailableTcpPort();
-
-
-	@Parameterized.Parameter(0)
-	public HttpServer server;
-
 	private Random rnd = new Random();
 
 
-	@Parameterized.Parameters(name = "server [{0}]")
-	public static Object[][] arguments() {
-		return new Object[][] {
-				{new JettyHttpServer()},
-				{new TomcatHttpServer()},
-				{new RxNettyHttpServer()}
-		};
-	}
-
-
-	@Before
-	public void setup() throws Exception {
-		this.server.setPort(port);
-		this.server.setHandler(new EchoHandler());
-		this.server.afterPropertiesSet();
-		this.server.start();
-	}
-
-	@After
-	public void tearDown() throws Exception {
-		this.server.stop();
+	@Override
+	protected EchoHandler createHttpHandler() {
+		return new EchoHandler();
 	}
 
 

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

@@ -26,6 +26,6 @@ public interface HttpServer extends InitializingBean, Lifecycle {
 
 	void setPort(int port);
 
-	void setHandler(ServerHttpHandler handler);
+	void setHandler(HttpHandler handler);
 
 }

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

@@ -23,7 +23,7 @@ public class HttpServerSupport {
 
 	private int port = -1;
 
-	private ServerHttpHandler httpHandler;
+	private HttpHandler httpHandler;
 
 
 	public void setPort(int port) {
@@ -34,11 +34,11 @@ public class HttpServerSupport {
 		return this.port;
 	}
 
-	public void setHandler(ServerHttpHandler handler) {
+	public void setHandler(HttpHandler handler) {
 		this.httpHandler = handler;
 	}
 
-	public ServerHttpHandler getHttpHandler() {
+	public HttpHandler getHttpHandler() {
 		return this.httpHandler;
 	}
 

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

@@ -22,6 +22,7 @@ import org.eclipse.jetty.servlet.ServletHolder;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.reactive.web.http.servlet.HttpHandlerServlet;
+import org.springframework.util.Assert;
 import org.springframework.util.SocketUtils;
 
 /**
@@ -48,8 +49,9 @@ public class JettyHttpServer extends HttpServerSupport implements InitializingBe
 
 		this.jettyServer = new Server();
 
+		Assert.notNull(getHttpHandler());
 		HttpHandlerServlet servlet = new HttpHandlerServlet();
-		servlet.setHandler(new EchoHandler());
+		servlet.setHandler(getHttpHandler());
 		ServletHolder servletHolder = new ServletHolder(servlet);
 
 		ServletContextHandler contextHandler = new ServletContextHandler(this.jettyServer, "", false, false);