Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch to reactor snapshots and make use of API change

Browse Source
This commit is contained in:
Rossen Stoyanchev
2013-10-22 22:32:49 -04:00
parent 6b0a62569b
commit f3ca3c108c
7 changed files with 78 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java
View File

@@ -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<byte[]> HEARTBEAT_MESSAGE = MessageBuilder.withPayload(new byte[] {'\n'}).build();
private static final Message<byte[]> 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<byte[]> conn = tcpConnection;
if (conn != null) {
conn.send(HEARTBEAT_MESSAGE).addCallback(
new ListenableFutureCallback<Boolean>() {
new ListenableFutureCallback<Void>() {
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<Boolean> forward(final Message<?> message) {
public ListenableFuture<Void> 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<Boolean>(new Callable<Boolean>() {
return new ListenableFutureTask<Void>(new Callable<Void>() {
@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<Boolean> future = this.tcpConnection.send((Message<byte[]>) message);
ListenableFuture<Void> future = this.tcpConnection.send((Message<byte[]>) message);
future.addCallback(new ListenableFutureCallback<Boolean>() {
future.addCallback(new ListenableFutureCallback<Void>() {
@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<Boolean> forward(Message<?> message) {
public ListenableFuture<Void> forward(Message<?> message) {
try {
ListenableFuture<Boolean> future = super.forward(message);
if (!future.get()) {
throw new MessageDeliveryException(message);
}
ListenableFuture<Void> future = super.forward(message);
future.get();
return future;
}
catch (Throwable t) {

10
spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PromiseToListenableFutureAdapter.java → spring-messaging/src/main/java/org/springframework/messaging/support/tcp/AbstractPromiseToListenableFutureAdapter.java
View File

@@ -38,14 +38,14 @@ import reactor.function.Consumer;
* @author Rossen Stoyanchev
* @since 4.0
*/
abstract class PromiseToListenableFutureAdapter<S, T> implements ListenableFuture<T> {
abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements ListenableFuture<T> {
private final Promise<S> promise;
private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<T>();
protected PromiseToListenableFutureAdapter(Promise<S> promise) {
protected AbstractPromiseToListenableFutureAdapter(Promise<S> promise) {
Assert.notNull(promise, "promise is required");
this.promise = promise;
@@ -70,11 +70,11 @@ abstract class PromiseToListenableFutureAdapter<S, T> 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);
}

41
spring-messaging/src/main/java/org/springframework/messaging/support/tcp/PassThroughPromiseToListenableFutureAdapter.java Normal file
View File

@@ -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<T> extends AbstractPromiseToListenableFutureAdapter<T, T> {
public PassThroughPromiseToListenableFutureAdapter(Promise<T> promise) {
super(promise);
}
@Override
protected T adapt(T result) {
return result;
}
}

2
spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorNettyTcpClient.java
View File

@@ -112,7 +112,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
public ListenableFuture<Void> shutdown() {
try {
Promise<Void> promise = this.tcpClient.close();
return new PromiseToListenableFutureAdapter<Void, Void>(promise) {
return new AbstractPromiseToListenableFutureAdapter<Void, Void>(promise) {
@Override
protected Void adapt(Void result) {
return result;

76
spring-messaging/src/main/java/org/springframework/messaging/support/tcp/ReactorTcpConnection.java
View File

@@ -41,10 +41,9 @@ public class ReactorTcpConnection<P> implements TcpConnection<P> {
}
@Override
public ListenableFuture<Boolean> send(Message<P> message) {
ConsumerListenableFuture future = new ConsumerListenableFuture();
this.reactorTcpConnection.send(message, future);
return future;
public ListenableFuture<Void> send(Message<P> message) {
Promise<Void> promise = this.reactorTcpConnection.send(message);
return new PassThroughPromiseToListenableFutureAdapter<Void>(promise);
}
@Override
@@ -62,73 +61,4 @@ public class ReactorTcpConnection<P> implements TcpConnection<P> {
this.reactorTcpConnection.close();
}
// Use this temporarily until reactor provides a send method returning a Promise
private static class ConsumerListenableFuture implements ListenableFuture<Boolean>, Consumer<Boolean> {
final Deferred<Boolean, Promise<Boolean>> deferred = new DeferredPromiseSpec<Boolean>().get();
private final ListenableFutureCallbackRegistry<Boolean> registry =
new ListenableFutureCallbackRegistry<Boolean>();
@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<? super Boolean> callback) {
this.registry.addCallback(callback);
}
}
}

2
spring-messaging/src/main/java/org/springframework/messaging/support/tcp/TcpConnection.java
View File

@@ -34,7 +34,7 @@ public interface TcpConnection<P> {
* @param message the message
* @return whether the send succeeded or not
*/
ListenableFuture<Boolean> send(Message<P> message);
ListenableFuture<Void> send(Message<P> message);
/**
* Register a task to invoke after a period of of read inactivity.