github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

CompletableFuture & CompletionStage integration. #500

Browse Source
pull/802/merge
Nikita 9 years ago
parent 45a0989bd1
commit 27209e88f2
5 changed files with 308 additions and 82 deletions
Show Stats Download Patch File Download Diff File

19
redisson/src/main/java/org/redisson/api/RFuture.java
Unescape Escape View File

@ -15,6 +15,7 @@
*/
package org.redisson.api;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import io.netty.util.concurrent.FutureListener;
@ -26,7 +27,7 @@ import io.netty.util.concurrent.FutureListener;
*
* @param <V>
*/
public interface RFuture<V> extends java.util.concurrent.Future<V> {
public interface RFuture<V> extends java.util.concurrent.Future<V>, CompletionStage<V> {
/**
* Returns {@code true} if and only if the I/O operation was completed
@ -52,6 +53,22 @@ public interface RFuture<V> extends java.util.concurrent.Future<V> {
*/
V getNow();
/**
* Returns the result value when complete, or throws an
* (unchecked) exception if completed exceptionally. To better
* conform with the use of common functional forms, if a
* computation involved in the completion of this
* CompletableFuture threw an exception, this method throws an
* (unchecked) {@link CompletionException} with the underlying
* exception as its cause.
*
* @return the result value
* @throws CancellationException if the computation was cancelled
* @throws CompletionException if this future completed
* exceptionally or a completion computation threw an exception
*/
V join();
/**
* Waits for this future to be completed within the
* specified time limit.

218
redisson/src/main/java/org/redisson/misc/PromiseDelegator.java
Unescape Escape View File

@ -15,9 +15,16 @@
*/
package org.redisson.misc;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import io.netty.util.concurrent.FutureListener;
@ -34,120 +41,261 @@ public class PromiseDelegator<T> implements RPromise<T> {
return promise;
}
@Override
public T join() {
return promise.join();
}
public boolean isSuccess() {
return promise.isSuccess();
}
@Override
public boolean trySuccess(T result) {
return promise.trySuccess(result);
}
@Override
public Throwable cause() {
return promise.cause();
}
@Override
public T getNow() {
return promise.getNow();
}
public boolean tryFailure(Throwable cause) {
return promise.tryFailure(cause);
}
@Override
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
return promise.await(timeout, unit);
}
public boolean setUncancellable() {
return promise.setUncancellable();
}
@Override
public boolean await(long timeoutMillis) throws InterruptedException {
return promise.await(timeoutMillis);
}
public RPromise<T> addListener(FutureListener<? super T> listener) {
return promise.addListener(listener);
}
@Override
public RPromise<T> addListeners(FutureListener<? super T>... listeners) {
return promise.addListeners(listeners);
}
@Override
public RPromise<T> removeListener(FutureListener<? super T> listener) {
return promise.removeListener(listener);
}
@Override
public RPromise<T> removeListeners(FutureListener<? super T>... listeners) {
return promise.removeListeners(listeners);
}
@Override
public RPromise<T> await() throws InterruptedException {
return promise.await();
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return promise.cancel(mayInterruptIfRunning);
}
public RPromise<T> awaitUninterruptibly() {
return promise.awaitUninterruptibly();
}
@Override
public RPromise<T> sync() throws InterruptedException {
return promise.sync();
}
@Override
public RPromise<T> syncUninterruptibly() {
return promise.syncUninterruptibly();
}
@Override
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
return promise.await(timeout, unit);
}
@Override
public boolean isCancelled() {
return promise.isCancelled();
}
@Override
public boolean isDone() {
return promise.isDone();
}
@Override
public boolean await(long timeoutMillis) throws InterruptedException {
return promise.await(timeoutMillis);
}
@Override
public T get() throws InterruptedException, ExecutionException {
return promise.get();
}
@Override
public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
return promise.awaitUninterruptibly(timeout, unit);
}
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return promise.get(timeout, unit);
}
@Override
public <U> CompletionStage<U> thenApply(Function<? super T, ? extends U> fn) {
return promise.thenApply(fn);
}
public boolean awaitUninterruptibly(long timeoutMillis) {
return promise.awaitUninterruptibly(timeoutMillis);
}
@Override
public T getNow() {
return promise.getNow();
public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
return promise.thenApplyAsync(fn);
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return promise.cancel(mayInterruptIfRunning);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
return promise.thenApplyAsync(fn, executor);
}
public CompletionStage<Void> thenAccept(Consumer<? super T> action) {
return promise.thenAccept(action);
}
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action) {
return promise.thenAcceptAsync(action);
}
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
return promise.thenAcceptAsync(action, executor);
}
public CompletionStage<Void> thenRun(Runnable action) {
return promise.thenRun(action);
}
public CompletionStage<Void> thenRunAsync(Runnable action) {
return promise.thenRunAsync(action);
}
public CompletionStage<Void> thenRunAsync(Runnable action, Executor executor) {
return promise.thenRunAsync(action, executor);
}
public <U, V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,
BiFunction<? super T, ? super U, ? extends V> fn) {
return promise.thenCombine(other, fn);
}
public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
BiFunction<? super T, ? super U, ? extends V> fn) {
return promise.thenCombineAsync(other, fn);
}
public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
return promise.thenCombineAsync(other, fn, executor);
}
public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) {
return promise.thenAcceptBoth(other, action);
}
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) {
return promise.thenAcceptBothAsync(other, action);
}
public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action, Executor executor) {
return promise.thenAcceptBothAsync(other, action, executor);
}
public CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
return promise.runAfterBoth(other, action);
}
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
return promise.runAfterBothAsync(other, action);
}
public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
return promise.runAfterBothAsync(other, action, executor);
}
public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
return promise.applyToEither(other, fn);
}
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
return promise.applyToEitherAsync(other, fn);
}
public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn,
Executor executor) {
return promise.applyToEitherAsync(other, fn, executor);
}
public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
return promise.acceptEither(other, action);
}
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
return promise.acceptEitherAsync(other, action);
}
public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,
Executor executor) {
return promise.acceptEitherAsync(other, action, executor);
}
public CompletionStage<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
return promise.runAfterEither(other, action);
}
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
return promise.runAfterEitherAsync(other, action);
}
public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
return promise.runAfterEitherAsync(other, action, executor);
}
public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
return promise.thenCompose(fn);
}
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
return promise.thenComposeAsync(fn);
}
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,
Executor executor) {
return promise.thenComposeAsync(fn, executor);
}
public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn) {
return promise.exceptionally(fn);
}
public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
return promise.whenComplete(action);
}
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
return promise.whenCompleteAsync(action);
}
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
return promise.whenCompleteAsync(action, executor);
}
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
return promise.handle(fn);
}
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
return promise.handleAsync(fn);
}
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
return promise.handleAsync(fn, executor);
}
public CompletableFuture<T> toCompletableFuture() {
return promise.toCompletableFuture();
}
}

130
redisson/src/main/java/org/redisson/misc/RedissonPromise.java
Unescape Escape View File

@ -15,15 +15,20 @@
*/
package org.redisson.misc;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.redisson.api.RFuture;
import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
import io.netty.util.concurrent.Promise;
import io.netty.util.internal.PlatformDependent;
/**
*
@ -31,9 +36,16 @@ import io.netty.util.concurrent.Promise;
*
* @param <T>
*/
public class RedissonPromise<T> implements RPromise<T> {
public class RedissonPromise<T> extends CompletableFuture<T> implements RPromise<T> {
private volatile boolean uncancellable;
private final int SUCCESS = 1;
private final int FAILED = 2;
private final int CANCELED = 3;
private final Promise<T> promise = ImmediateEventExecutor.INSTANCE.newPromise();
private final AtomicInteger status = new AtomicInteger();
public RedissonPromise() {
}
@ -50,34 +62,51 @@ public class RedissonPromise<T> implements RPromise<T> {
return future;
}
public Promise<T> getInnerPromise() {
return promise;
}
@Override
public boolean isSuccess() {
return promise.isSuccess();
return isDone() && !isCompletedExceptionally();
}
@Override
public boolean trySuccess(T result) {
return promise.trySuccess(result);
public synchronized boolean trySuccess(T result) {
if (status.compareAndSet(0, SUCCESS)) {
complete(result);
promise.trySuccess(result);
return true;
}
return false;
}
@Override
public Throwable cause() {
return promise.cause();
try {
getNow(null);
} catch (CompletionException e) {
return e.getCause();
}
return null;
}
@Override
public boolean tryFailure(Throwable cause) {
return promise.tryFailure(cause);
public synchronized boolean tryFailure(Throwable cause) {
if (status.compareAndSet(0, FAILED)) {
completeExceptionally(cause);
promise.tryFailure(cause);
return true;
}
return false;
}
@Override
public boolean setUncancellable() {
return promise.setUncancellable();
if (!isDone()) {
uncancellable = true;
}
return uncancellable;
}
@Override
@ -106,76 +135,97 @@ public class RedissonPromise<T> implements RPromise<T> {
@Override
public RPromise<T> await() throws InterruptedException {
promise.await();
try {
get();
} catch (ExecutionException | CancellationException e) {
// skip
}
return this;
}
@Override
public RPromise<T> awaitUninterruptibly() {
promise.awaitUninterruptibly();
try {
return await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return this;
}
@Override
public RPromise<T> sync() throws InterruptedException {
promise.sync();
try {
get();
} catch (ExecutionException e) {
if (e.getCause() instanceof CancellationException) {
throw (CancellationException)e.getCause();
}
PlatformDependent.throwException(e.getCause());
}
return this;
}
@Override
public RPromise<T> syncUninterruptibly() {
promise.syncUninterruptibly();
try {
join();
} catch (CompletionException e) {
PlatformDependent.throwException(e.getCause());
}
return this;
}
@Override
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
return promise.await(timeout, unit);
}
@Override
public boolean isCancelled() {
return promise.isCancelled();
}
@Override
public boolean isDone() {
return promise.isDone();
try {
get(timeout, unit);
} catch (ExecutionException e) {
if (e.getCause() instanceof CancellationException) {
throw (CancellationException)e.getCause();
}
throw new CompletionException(e.getCause());
} catch (TimeoutException e) {
return false;
}
return isDone();
}
@Override
public boolean await(long timeoutMillis) throws InterruptedException {
return promise.await(timeoutMillis);
}
@Override
public T get() throws InterruptedException, ExecutionException {
return promise.get();
return await(timeoutMillis, TimeUnit.MILLISECONDS);
}
@Override
public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
return promise.awaitUninterruptibly(timeout, unit);
}
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return promise.get(timeout, unit);
try {
return await(timeout, unit);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
@Override
public boolean awaitUninterruptibly(long timeoutMillis) {
return promise.awaitUninterruptibly(timeoutMillis);
return awaitUninterruptibly(timeoutMillis, TimeUnit.MILLISECONDS);
}
@Override
public T getNow() {
return promise.getNow();
return getNow(null);
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return promise.cancel(mayInterruptIfRunning);
public synchronized boolean cancel(boolean mayInterruptIfRunning) {
if (uncancellable) {
return false;
}
if (status.compareAndSet(0, CANCELED)) {
promise.cancel(mayInterruptIfRunning);
return super.cancel(mayInterruptIfRunning);
}
return false;
}
}

8
redisson/src/test/java/org/redisson/RedisClientTest.java
Unescape Escape View File

@ -33,7 +33,6 @@ import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
public class RedisClientTest {
@ -69,12 +68,17 @@ public class RedisClientTest {
public void testConnectAsync() throws InterruptedException {
RedisClient c = new RedisClient("localhost", 6379);
RFuture<RedisConnection> f = c.connectAsync();
final CountDownLatch l = new CountDownLatch(1);
final CountDownLatch l = new CountDownLatch(2);
f.addListener((FutureListener<RedisConnection>) future -> {
RedisConnection conn = future.get();
assertThat(conn.sync(RedisCommands.PING)).isEqualTo("PONG");
l.countDown();
});
f.handle((conn, ex) -> {
assertThat(conn.sync(RedisCommands.PING)).isEqualTo("PONG");
l.countDown();
return null;
});
assertThat(l.await(10, TimeUnit.SECONDS)).isTrue();
}

11
redisson/src/test/java/org/redisson/RedissonMapTest.java
Unescape Escape View File

@ -12,6 +12,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import org.junit.Assert;
@ -627,8 +628,14 @@ public class RedissonMapTest extends BaseTest {
map.put(4, 6);
map.put(7, 8);
assertThat(map.fastRemoveAsync(1, 3, 7).get()).isEqualTo(3);
Thread.sleep(1);
CountDownLatch l = new CountDownLatch(1);
RFuture<Long> future = map.fastRemoveAsync(1, 3, 7);
future.handle((r, ex) -> {
assertThat(r).isEqualTo(3);
l.countDown();
return this;
});
assertThat(future.get()).isEqualTo(3);
assertThat(map.size()).isEqualTo(1);
}

Write Preview
Loading…
Cancel
Save