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

refactoring

Browse Source
pull/4087/head
Nikita Koksharov 3 years ago
parent 07d606de7c
commit 36bafe4935
2 changed files with 120 additions and 173 deletions
Show Stats Download Patch File Download Diff File

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

@ -32,11 +32,12 @@ import org.redisson.iterator.RedissonBaseIterator;
import org.redisson.iterator.RedissonListIterator; import org.redisson.iterator.RedissonListIterator;
import org.redisson.mapreduce.RedissonCollectionMapReduce; import org.redisson.mapreduce.RedissonCollectionMapReduce;
import org.redisson.misc.CompletableFutureWrapper; import org.redisson.misc.CompletableFutureWrapper;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise; import org.redisson.misc.RedissonPromise;
import java.util.*; import java.util.*;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.function.Predicate; import java.util.function.Predicate;
import static org.redisson.client.protocol.RedisCommands.*; import static org.redisson.client.protocol.RedisCommands.*;
@ -391,25 +392,21 @@ public class RedissonList<V> extends RedissonExpirable implements RList<V> {
@Override @Override
public RFuture<V> setAsync(int index, V element) { public RFuture<V> setAsync(int index, V element) {
RPromise<V> result = new RedissonPromise<V>();
RFuture<V> future = commandExecutor.evalWriteAsync(getRawName(), codec, RedisCommands.EVAL_OBJECT, RFuture<V> future = commandExecutor.evalWriteAsync(getRawName(), codec, RedisCommands.EVAL_OBJECT,
"local v = redis.call('lindex', KEYS[1], ARGV[1]); " + "local v = redis.call('lindex', KEYS[1], ARGV[1]); " +
"redis.call('lset', KEYS[1], ARGV[1], ARGV[2]); " + "redis.call('lset', KEYS[1], ARGV[1], ARGV[2]); " +
"return v", "return v",
Collections.<Object>singletonList(getRawName()), index, encode(element)); Collections.singletonList(getRawName()), index, encode(element));
future.onComplete((res, e) -> { CompletionStage<V> f = future.handle((res, e) -> {
if (e != null) { if (e != null) {
if (e.getMessage().contains("ERR index out of range")) { if (e.getMessage().contains("ERR index out of range")) {
result.tryFailure(new IndexOutOfBoundsException("index out of range")); throw new CompletionException(new IndexOutOfBoundsException("index out of range"));
return;
} }
result.tryFailure(e); throw new CompletionException(e);
return;
} }
return res;
result.trySuccess(res);
}); });
return result; return new CompletableFutureWrapper<>(f);
} }
@Override @Override

274
redisson/src/main/java/org/redisson/RedissonMap.java
Unescape Escape View File

@ -32,7 +32,6 @@ import org.redisson.connection.decoder.MapGetAllDecoder;
import org.redisson.iterator.RedissonMapIterator; import org.redisson.iterator.RedissonMapIterator;
import org.redisson.mapreduce.RedissonMapReduce; import org.redisson.mapreduce.RedissonMapReduce;
import org.redisson.misc.CompletableFutureWrapper; import org.redisson.misc.CompletableFutureWrapper;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise; import org.redisson.misc.RedissonPromise;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -192,7 +191,6 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
newValuePromise.complete(value); newValuePromise.complete(value);
} }
return newValuePromise return newValuePromise
.exceptionally(e -> null)
.thenCompose(newValue -> { .thenCompose(newValue -> {
RFuture<?> future; RFuture<?> future;
if (newValue != null) { if (newValue != null) {
@ -200,17 +198,11 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
} else { } else {
future = fastRemoveAsync(key); future = fastRemoveAsync(key);
} }
return future.thenApply(res -> { return future.thenApply(res -> newValue);
lock.unlockAsync(threadId);
return newValue;
});
}); });
}); }).whenComplete((c, e) -> {
});
f.whenComplete((r, e) -> {
if (e != null) {
lock.unlockAsync(threadId); lock.unlockAsync(threadId);
} });
}); });
return new CompletableFutureWrapper<>(f); return new CompletableFutureWrapper<>(f);
} }
@ -221,64 +213,53 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
Objects.requireNonNull(remappingFunction); Objects.requireNonNull(remappingFunction);
RLock lock = getLock(key); RLock lock = getLock(key);
RPromise<V> result = new RedissonPromise<>();
long threadId = Thread.currentThread().getId(); long threadId = Thread.currentThread().getId();
lock.lockAsync(threadId).onComplete((r, e) -> { CompletionStage<V> f = lock.lockAsync(threadId)
if (e != null) { .thenCompose(r -> {
result.tryFailure(e); RFuture<V> oldValueFuture = getAsync(key);
return; return oldValueFuture.thenCompose(oldValue -> {
} CompletableFuture<V> result = new CompletableFuture<>();
commandExecutor.getConnectionManager().getExecutor().execute(() -> {
RFuture<V> oldValueFuture = getAsync(key); V newValue;
oldValueFuture.onComplete((oldValue, ex) -> { try {
if (ex != null) { newValue = remappingFunction.apply(key, oldValue);
lock.unlockAsync(threadId); } catch (Exception exception) {
result.tryFailure(ex); result.completeExceptionally(exception);
return; return;
} }
commandExecutor.getConnectionManager().getExecutor().execute(() -> { if (newValue == null) {
V newValue; if (oldValue != null) {
try { fastRemoveAsync(key).whenComplete((res, exc) -> {
newValue = remappingFunction.apply(key, oldValue); if (exc != null) {
} catch (Exception exception) { result.completeExceptionally(exc);
lock.unlockAsync(threadId); return;
result.tryFailure(exception); }
return;
}
if (newValue == null) { result.complete(newValue);
if (oldValue != null) { });
fastRemoveAsync(key).onComplete((res, exc) -> { return;
lock.unlockAsync(threadId); }
} else {
fastPutAsync(key, newValue).whenComplete((res, exc) -> {
if (exc != null) { if (exc != null) {
result.tryFailure(exc); result.completeExceptionally(exc);
return; return;
} }
result.trySuccess(newValue); result.complete(newValue);
}); });
return; return;
} }
} else {
fastPutAsync(key, newValue).onComplete((res, exc) -> {
lock.unlockAsync(threadId);
if (exc != null) {
result.tryFailure(exc);
return;
}
result.trySuccess(newValue);
});
return;
}
result.complete(newValue);
});
return result;
}).whenComplete((c, e) -> {
lock.unlockAsync(threadId); lock.unlockAsync(threadId);
result.trySuccess(newValue);
}); });
});
}); });
return result; return new CompletableFutureWrapper<>(f);
} }
@ -312,56 +293,40 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
Objects.requireNonNull(mappingFunction); Objects.requireNonNull(mappingFunction);
RLock lock = getLock(key); RLock lock = getLock(key);
RPromise<V> result = new RedissonPromise<>();
long threadId = Thread.currentThread().getId(); long threadId = Thread.currentThread().getId();
lock.lockAsync(threadId).onComplete((r, e) -> { CompletionStage<V> f = lock.lockAsync(threadId)
if (e != null) { .thenCompose(r -> {
result.tryFailure(e); RFuture<V> oldValueFuture = getAsync(key);
return; return oldValueFuture.thenCompose(oldValue -> {
} if (oldValue != null) {
return CompletableFuture.completedFuture(oldValue);
RFuture<V> oldValueFuture = getAsync(key); }
oldValueFuture.onComplete((oldValue, ex) -> {
if (ex != null) {
lock.unlockAsync(threadId);
result.tryFailure(ex);
return;
}
if (oldValue != null) {
lock.unlockAsync(threadId);
result.trySuccess(oldValue);
return;
}
commandExecutor.getConnectionManager().getExecutor().execute(() -> { CompletableFuture<V> result = new CompletableFuture<>();
V newValue; commandExecutor.getConnectionManager().getExecutor().execute(() -> {
try { V newValue;
newValue = mappingFunction.apply(key); try {
} catch (Exception exception) { newValue = mappingFunction.apply(key);
lock.unlockAsync(threadId); } catch (Exception exception) {
result.tryFailure(exception); result.completeExceptionally(exception);
return; return;
} }
if (newValue != null) { if (newValue != null) {
fastPutAsync(key, newValue).onComplete((res, exc) -> { fastPutAsync(key, newValue).thenAccept(res -> {
lock.unlockAsync(threadId); result.complete(newValue);
if (exc != null) { });
result.tryFailure(exc);
return; return;
} }
result.trySuccess(newValue); result.complete(null);
}); });
return; return result;
} }).whenComplete((c, e) -> {
lock.unlockAsync(threadId);
lock.unlockAsync(threadId); });
result.trySuccess(null);
}); });
});
}); return new CompletableFutureWrapper<>(f);
return result;
} }
@Override @Override
@ -393,63 +358,53 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
Objects.requireNonNull(remappingFunction); Objects.requireNonNull(remappingFunction);
RLock lock = getLock(key); RLock lock = getLock(key);
RPromise<V> result = new RedissonPromise<>();
long threadId = Thread.currentThread().getId(); long threadId = Thread.currentThread().getId();
lock.lockAsync(threadId).onComplete((r, e) -> { CompletionStage<V> f = lock.lockAsync(threadId)
if (e != null) { .thenCompose(r -> {
result.tryFailure(e); RFuture<V> oldValueFuture = getAsync(key);
return; return oldValueFuture.thenCompose(oldValue -> {
} if (oldValue == null) {
RFuture<V> oldValueFuture = getAsync(key); return CompletableFuture.completedFuture(null);
oldValueFuture.onComplete((oldValue, ex) -> { }
if (ex != null) {
lock.unlockAsync(threadId);
result.tryFailure(e);
return;
}
if (oldValue == null) {
lock.unlockAsync(threadId);
result.trySuccess(null);
return;
}
commandExecutor.getConnectionManager().getExecutor().execute(() -> { CompletableFuture<V> result = new CompletableFuture<>();
V newValue; commandExecutor.getConnectionManager().getExecutor().execute(() -> {
try { V newValue;
newValue = remappingFunction.apply(key, oldValue); try {
} catch (Exception exception) { newValue = remappingFunction.apply(key, oldValue);
lock.unlockAsync(threadId); } catch (Exception exception) {
result.tryFailure(exception); result.completeExceptionally(exception);
return;
}
if (newValue != null) {
RFuture<Boolean> fastPutFuture = fastPutAsync(key, newValue);
fastPutFuture.onComplete((re, ex1) -> {
lock.unlockAsync(threadId);
if (ex1 != null) {
result.tryFailure(ex1);
return; return;
} }
if (newValue != null) {
RFuture<Boolean> fastPutFuture = fastPutAsync(key, newValue);
fastPutFuture.whenComplete((re, ex1) -> {
if (ex1 != null) {
result.completeExceptionally(ex1);
return;
}
result.trySuccess(newValue); result.complete(newValue);
}); });
} else { } else {
RFuture<Long> removeFuture = fastRemoveAsync(key); RFuture<Long> removeFuture = fastRemoveAsync(key);
removeFuture.onComplete((re, ex1) -> { removeFuture.whenComplete((re, ex1) -> {
lock.unlockAsync(threadId); if (ex1 != null) {
if (ex1 != null) { result.completeExceptionally(ex1);
result.tryFailure(ex1); return;
return; }
}
result.trySuccess(null); result.complete(null);
});
}
}); });
} return result;
}).whenComplete((c, e) -> {
lock.unlockAsync(threadId);
});
}); });
});
}); return new CompletableFutureWrapper<>(f);
return result;
} }
@Override @Override
@ -702,22 +657,17 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
protected final <M> RFuture<M> mapWriterFuture(RFuture<M> future, MapWriterTask task, Function<M, Boolean> condition) { protected final <M> RFuture<M> mapWriterFuture(RFuture<M> future, MapWriterTask task, Function<M, Boolean> condition) {
if (options != null && options.getWriteMode() == WriteMode.WRITE_BEHIND) { if (options != null && options.getWriteMode() == WriteMode.WRITE_BEHIND) {
future.onComplete((res, e) -> { CompletionStage<M> f = future.whenComplete((res, e) -> {
if (e == null && condition.apply(res)) { if (e == null && condition.apply(res)) {
writeBehindTask.addTask(task); writeBehindTask.addTask(task);
} }
}); });
return future; return new CompletableFutureWrapper<>(f);
} }
final RPromise<M> promise = new RedissonPromise<>();
future.onComplete((res, e) -> {
if (e != null) {
promise.tryFailure(e);
return;
}
CompletionStage<M> f = future.thenCompose(res -> {
if (condition.apply(res)) { if (condition.apply(res)) {
CompletableFuture<M> promise = new CompletableFuture<>();
commandExecutor.getConnectionManager().getExecutor().execute(() -> { commandExecutor.getConnectionManager().getExecutor().execute(() -> {
try { try {
if (task instanceof MapWriterTask.Add) { if (task instanceof MapWriterTask.Add) {
@ -726,17 +676,17 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
options.getWriter().delete(task.getKeys()); options.getWriter().delete(task.getKeys());
} }
} catch (Exception ex) { } catch (Exception ex) {
promise.tryFailure(ex); promise.completeExceptionally(ex);
return; return;
} }
promise.trySuccess(res); promise.complete(res);
}); });
} else { return promise;
promise.trySuccess(res);
} }
return CompletableFuture.completedFuture(res);
}); });
return promise; return new CompletableFutureWrapper<>(f);
} }
protected RFuture<Void> putAllOperationAsync(Map<? extends K, ? extends V> map) { protected RFuture<Void> putAllOperationAsync(Map<? extends K, ? extends V> map) {

Write Preview
Loading…
Cancel
Save