diff --git a/build.gradle b/build.gradle index 000e28dafe..75b4f031ac 100644 --- a/build.gradle +++ b/build.gradle @@ -384,8 +384,8 @@ project("spring-messaging") { optional(project(":spring-websocket")) optional(project(":spring-webmvc")) optional("com.fasterxml.jackson.core:jackson-databind:2.2.0") - optional("org.projectreactor:reactor-core:1.0.0.RC1") - optional("org.projectreactor:reactor-tcp:1.0.0.RC1") + optional("org.projectreactor:reactor-core:1.0.0.BUILD-SNAPSHOT") + optional("org.projectreactor:reactor-tcp:1.0.0.BUILD-SNAPSHOT") optional("org.eclipse.jetty.websocket:websocket-server:9.0.5.v20130815") optional("org.eclipse.jetty.websocket:websocket-client:9.0.5.v20130815") testCompile(project(":spring-test")) @@ -406,6 +406,11 @@ project("spring-messaging") { testCompile("org.apache.tomcat.embed:tomcat-embed-logging-juli:8.0.0-RC3") testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}") } + + repositories { + maven { url "http://repo.spring.io/snapshot" } // for reactor (until RC2) + } + } project("spring-tx") { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java index cc0f1be16d..3b7bdb5f6e 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java @@ -25,6 +25,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.simp.SimpMessageType; import org.springframework.messaging.simp.handler.AbstractBrokerMessageHandler; import org.springframework.messaging.support.MessageBuilder; @@ -67,10 +68,15 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler private static final byte[] EMPTY_PAYLOAD = new byte[0]; - private static final Message HEARTBEAT_MESSAGE = MessageBuilder.withPayload(new byte[] {'\n'}).build(); + private static final Message HEARTBEAT_MESSAGE; private static final long HEARTBEAT_MULTIPLIER = 3; + static { + SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(SimpMessageType.HEARTBEAT); + HEARTBEAT_MESSAGE = MessageBuilder.withPayload(new byte[] {'\n'}).setHeaders(headers).build(); + } + private final MessageChannel messageChannel; @@ -464,11 +470,11 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler TcpConnection conn = tcpConnection; if (conn != null) { conn.send(HEARTBEAT_MESSAGE).addCallback( - new ListenableFutureCallback() { + new ListenableFutureCallback() { public void onFailure(Throwable t) { - handleTcpConnectionFailure("Failed to send heartbeat", null); + handleTcpConnectionFailure("Failed to send heartbeat", t); } - public void onSuccess(Boolean result) {} + public void onSuccess(Void result) {} }); } } @@ -492,16 +498,16 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler sendStompErrorToClient("Connection to broker closed"); } - public ListenableFuture forward(final Message message) { + public ListenableFuture forward(final Message message) { if (!this.isStompConnected) { if (logger.isWarnEnabled()) { logger.warn("Connection to broker inactive or not ready, ignoring message=" + message); } - return new ListenableFutureTask(new Callable() { + return new ListenableFutureTask(new Callable() { @Override - public Boolean call() throws Exception { - return Boolean.FALSE; + public Void call() throws Exception { + return null; } }); } @@ -511,11 +517,11 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler } @SuppressWarnings("unchecked") - ListenableFuture future = this.tcpConnection.send((Message) message); + ListenableFuture future = this.tcpConnection.send((Message) message); - future.addCallback(new ListenableFutureCallback() { + future.addCallback(new ListenableFutureCallback() { @Override - public void onSuccess(Boolean result) { + public void onSuccess(Void result) { StompCommand command = StompHeaderAccessor.wrap(message).getCommand(); if (command == StompCommand.DISCONNECT) { resetTcpConnection(); @@ -574,12 +580,10 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler } @Override - public ListenableFuture forward(Message message) { + public ListenableFuture forward(Message message) { try { - ListenableFuture future = super.forward(message); - if (!future.get()) { - throw new MessageDeliveryException(message); - } + ListenableFuture future = super.forward(message); + future.get(); return future; } catch (Throwable t) { diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PromiseToListenableFutureAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/AbstractPromiseToListenableFutureAdapter.java similarity index 90% rename from spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PromiseToListenableFutureAdapter.java rename to spring-messaging/src/main/java/org/springframework/messaging/support/tcp/AbstractPromiseToListenableFutureAdapter.java index ac0f5951e0..63e54893c1 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PromiseToListenableFutureAdapter.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/AbstractPromiseToListenableFutureAdapter.java @@ -38,14 +38,14 @@ import reactor.function.Consumer; * @author Rossen Stoyanchev * @since 4.0 */ -abstract class PromiseToListenableFutureAdapter implements ListenableFuture { +abstract class AbstractPromiseToListenableFutureAdapter implements ListenableFuture { private final Promise promise; private final ListenableFutureCallbackRegistry registry = new ListenableFutureCallbackRegistry(); - protected PromiseToListenableFutureAdapter(Promise promise) { + protected AbstractPromiseToListenableFutureAdapter(Promise promise) { Assert.notNull(promise, "promise is required"); this.promise = promise; @@ -70,11 +70,11 @@ abstract class PromiseToListenableFutureAdapter implements ListenableFutur }); } - protected abstract T adapt(S adapteeResult); + protected abstract T adapt(S result); @Override - public T get() { - S result = this.promise.get(); + public T get() throws InterruptedException { + S result = this.promise.await(); return adapt(result); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PassThroughPromiseToListenableFutureAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PassThroughPromiseToListenableFutureAdapter.java new file mode 100644 index 0000000000..03d6f528fd --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PassThroughPromiseToListenableFutureAdapter.java @@ -0,0 +1,41 @@ +/* + * Copyright 2002-2013 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.messaging.support.tcp; + +import reactor.core.composable.Promise; + + +/** + * A Promise-to-ListenableFutureAdapter where the source and the target from the Promise and + * the ListenableFuture respectively are of the same type. + * + * @author Rossen Stoyanchev + * @since 4.0 + */ +class PassThroughPromiseToListenableFutureAdapter extends AbstractPromiseToListenableFutureAdapter { + + + public PassThroughPromiseToListenableFutureAdapter(Promise promise) { + super(promise); + } + + @Override + protected T adapt(T result) { + return result; + } + +} \ No newline at end of file diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorNettyTcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorNettyTcpClient.java index 459258ece2..6e53fa95a5 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorNettyTcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorNettyTcpClient.java @@ -112,7 +112,7 @@ public class ReactorNettyTcpClient

implements TcpOperations

{ public ListenableFuture shutdown() { try { Promise promise = this.tcpClient.close(); - return new PromiseToListenableFutureAdapter(promise) { + return new AbstractPromiseToListenableFutureAdapter(promise) { @Override protected Void adapt(Void result) { return result; diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorTcpConnection.java b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorTcpConnection.java index 81e99db187..85904617e1 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorTcpConnection.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorTcpConnection.java @@ -41,10 +41,9 @@ public class ReactorTcpConnection

implements TcpConnection

{ } @Override - public ListenableFuture send(Message

message) { - ConsumerListenableFuture future = new ConsumerListenableFuture(); - this.reactorTcpConnection.send(message, future); - return future; + public ListenableFuture send(Message

message) { + Promise promise = this.reactorTcpConnection.send(message); + return new PassThroughPromiseToListenableFutureAdapter(promise); } @Override @@ -62,73 +61,4 @@ public class ReactorTcpConnection

implements TcpConnection

{ this.reactorTcpConnection.close(); } - - // Use this temporarily until reactor provides a send method returning a Promise - - - private static class ConsumerListenableFuture implements ListenableFuture, Consumer { - - final Deferred> deferred = new DeferredPromiseSpec().get(); - - private final ListenableFutureCallbackRegistry registry = - new ListenableFutureCallbackRegistry(); - - @Override - public void accept(Boolean result) { - - this.deferred.accept(result); - - if (result == null) { - this.registry.failure(new TimeoutException()); - } - else if (result) { - this.registry.success(result); - } - else { - this.registry.failure(new Exception("Failed send message")); - } - } - - @Override - public Boolean get() { - try { - return this.deferred.compose().await(); - } - catch (InterruptedException e) { - return Boolean.FALSE; - } - } - - @Override - public Boolean get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - - Boolean result = this.deferred.compose().await(timeout, unit); - if (result == null) { - throw new TimeoutException(); - } - return result; - } - - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return this.deferred.compose().isComplete(); - } - - @Override - public void addCallback(ListenableFutureCallback callback) { - this.registry.addCallback(callback); - } - } - } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/TcpConnection.java b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/TcpConnection.java index 57f89036ab..8e69a64962 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/TcpConnection.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/support/tcp/TcpConnection.java @@ -34,7 +34,7 @@ public interface TcpConnection

{ * @param message the message * @return whether the send succeeded or not */ - ListenableFuture send(Message

message); + ListenableFuture send(Message

message); /** * Register a task to invoke after a period of of read inactivity.