|
|
|
|
@ -15,27 +15,15 @@
|
|
|
|
|
*/
|
|
|
|
|
package org.redisson.reactive;
|
|
|
|
|
|
|
|
|
|
import static org.redisson.client.protocol.RedisCommands.LINDEX;
|
|
|
|
|
import static org.redisson.client.protocol.RedisCommands.LREM_SINGLE;
|
|
|
|
|
import static org.redisson.client.protocol.RedisCommands.RPUSH;
|
|
|
|
|
|
|
|
|
|
import java.util.ArrayList;
|
|
|
|
|
import java.util.Collection;
|
|
|
|
|
import java.util.Collections;
|
|
|
|
|
import java.util.List;
|
|
|
|
|
|
|
|
|
|
import org.reactivestreams.Publisher;
|
|
|
|
|
import org.reactivestreams.Subscriber;
|
|
|
|
|
import org.reactivestreams.Subscription;
|
|
|
|
|
import org.redisson.RedissonList;
|
|
|
|
|
import org.redisson.RedissonObject;
|
|
|
|
|
import org.redisson.api.RFuture;
|
|
|
|
|
import org.redisson.client.codec.Codec;
|
|
|
|
|
import org.redisson.client.protocol.RedisCommands;
|
|
|
|
|
import org.redisson.client.protocol.convertor.LongReplayConvertor;
|
|
|
|
|
import org.redisson.command.CommandReactiveExecutor;
|
|
|
|
|
|
|
|
|
|
import reactor.fn.Supplier;
|
|
|
|
|
import io.netty.util.concurrent.Future;
|
|
|
|
|
import io.netty.util.concurrent.FutureListener;
|
|
|
|
|
import reactor.rx.Stream;
|
|
|
|
|
import reactor.rx.subscription.ReactiveSubscription;
|
|
|
|
|
|
|
|
|
|
@ -49,15 +37,12 @@ import reactor.rx.subscription.ReactiveSubscription;
|
|
|
|
|
public class RedissonListReactive<V> {
|
|
|
|
|
|
|
|
|
|
private final RedissonList<V> instance;
|
|
|
|
|
private final CommandReactiveExecutor commandExecutor;
|
|
|
|
|
|
|
|
|
|
public RedissonListReactive(CommandReactiveExecutor commandExecutor, String name) {
|
|
|
|
|
this.commandExecutor = commandExecutor;
|
|
|
|
|
this.instance = new RedissonList<V>(commandExecutor, name, null);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public RedissonListReactive(Codec codec, CommandReactiveExecutor commandExecutor, String name) {
|
|
|
|
|
this.commandExecutor = commandExecutor;
|
|
|
|
|
this.instance = new RedissonList<V>(codec, commandExecutor, name, null);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -89,33 +74,25 @@ public class RedissonListReactive<V> {
|
|
|
|
|
@Override
|
|
|
|
|
protected void onRequest(final long n) {
|
|
|
|
|
final ReactiveSubscription<V> m = this;
|
|
|
|
|
get(currentIndex).subscribe(new Subscriber<V>() {
|
|
|
|
|
V currValue;
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public void onSubscribe(Subscription s) {
|
|
|
|
|
s.request(Long.MAX_VALUE);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
instance.getAsync(currentIndex).addListener(new FutureListener<V>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void onNext(V value) {
|
|
|
|
|
currValue = value;
|
|
|
|
|
m.onNext(value);
|
|
|
|
|
if (forward) {
|
|
|
|
|
currentIndex++;
|
|
|
|
|
} else {
|
|
|
|
|
currentIndex--;
|
|
|
|
|
public void operationComplete(Future<V> future) throws Exception {
|
|
|
|
|
if (!future.isSuccess()) {
|
|
|
|
|
m.onError(future.cause());
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public void onError(Throwable error) {
|
|
|
|
|
m.onError(error);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public void onComplete() {
|
|
|
|
|
if (currValue == null) {
|
|
|
|
|
|
|
|
|
|
V value = future.getNow();
|
|
|
|
|
if (value != null) {
|
|
|
|
|
m.onNext(value);
|
|
|
|
|
if (forward) {
|
|
|
|
|
currentIndex++;
|
|
|
|
|
} else {
|
|
|
|
|
currentIndex--;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (value == null) {
|
|
|
|
|
m.onComplete();
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
@ -132,120 +109,15 @@ public class RedissonListReactive<V> {
|
|
|
|
|
};
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Integer> add(V e) {
|
|
|
|
|
return commandExecutor.writeReactive(instance.getName(), instance.getCodec(), RPUSH, instance.getName(), ((RedissonObject)instance).encode(e));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
protected Publisher<Boolean> remove(Object o, int count) {
|
|
|
|
|
return commandExecutor.writeReactive(instance.getName(), instance.getCodec(), LREM_SINGLE, instance.getName(), count, ((RedissonObject)instance).encode(o));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Integer> addAll(Publisher<? extends V> c) {
|
|
|
|
|
public Publisher<Boolean> addAll(Publisher<? extends V> c) {
|
|
|
|
|
return new PublisherAdder<V>() {
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public Integer sum(Integer first, Integer second) {
|
|
|
|
|
return second;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
public Publisher<Integer> add(Object o) {
|
|
|
|
|
return RedissonListReactive.this.add((V)o);
|
|
|
|
|
public RFuture<Boolean> add(Object o) {
|
|
|
|
|
return instance.addAsync((V)o);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}.addAll(c);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Integer> addAll(Collection<? extends V> c) {
|
|
|
|
|
if (c.isEmpty()) {
|
|
|
|
|
return commandExecutor.reactive(new Supplier<RFuture<Integer>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public RFuture<Integer> get() {
|
|
|
|
|
return instance.sizeAsync();
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
List<Object> args = new ArrayList<Object>(c.size() + 1);
|
|
|
|
|
args.add(instance.getName());
|
|
|
|
|
((RedissonObject)instance).encode(args, c);
|
|
|
|
|
return commandExecutor.writeReactive(instance.getName(), instance.getCodec(), RPUSH, args.toArray());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Integer> addAll(long index, Collection<? extends V> coll) {
|
|
|
|
|
if (index < 0) {
|
|
|
|
|
throw new IndexOutOfBoundsException("index: " + index);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (coll.isEmpty()) {
|
|
|
|
|
return commandExecutor.reactive(new Supplier<RFuture<Integer>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public RFuture<Integer> get() {
|
|
|
|
|
return instance.sizeAsync();
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (index == 0) { // prepend elements to list
|
|
|
|
|
List<Object> elements = new ArrayList<Object>();
|
|
|
|
|
((RedissonObject)instance).encode(elements, coll);
|
|
|
|
|
Collections.reverse(elements);
|
|
|
|
|
elements.add(0, instance.getName());
|
|
|
|
|
|
|
|
|
|
return commandExecutor.writeReactive(instance.getName(), instance.getCodec(), RedisCommands.LPUSH, elements.toArray());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
List<Object> args = new ArrayList<Object>(coll.size() + 1);
|
|
|
|
|
args.add(index);
|
|
|
|
|
((RedissonObject)instance).encode(args, coll);
|
|
|
|
|
return commandExecutor.evalWriteReactive(instance.getName(), instance.getCodec(), RedisCommands.EVAL_INTEGER,
|
|
|
|
|
"local ind = table.remove(ARGV, 1); " + // index is the first parameter
|
|
|
|
|
"local size = redis.call('llen', KEYS[1]); " +
|
|
|
|
|
"assert(tonumber(ind) <= size, 'index: ' .. ind .. ' but current size: ' .. size); " +
|
|
|
|
|
"local tail = redis.call('lrange', KEYS[1], ind, -1); " +
|
|
|
|
|
"redis.call('ltrim', KEYS[1], 0, ind - 1); " +
|
|
|
|
|
"for i, v in ipairs(ARGV) do redis.call('rpush', KEYS[1], v) end;" +
|
|
|
|
|
"for i, v in ipairs(tail) do redis.call('rpush', KEYS[1], v) end;" +
|
|
|
|
|
"return redis.call('llen', KEYS[1]);",
|
|
|
|
|
Collections.<Object>singletonList(instance.getName()), args.toArray());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<V> get(long index) {
|
|
|
|
|
return commandExecutor.readReactive(instance.getName(), instance.getCodec(), LINDEX, instance.getName(), index);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<V> set(long index, V element) {
|
|
|
|
|
return commandExecutor.evalWriteReactive(instance.getName(), instance.getCodec(), RedisCommands.EVAL_OBJECT,
|
|
|
|
|
"local v = redis.call('lindex', KEYS[1], ARGV[1]); " +
|
|
|
|
|
"redis.call('lset', KEYS[1], ARGV[1], ARGV[2]); " +
|
|
|
|
|
"return v",
|
|
|
|
|
Collections.<Object>singletonList(instance.getName()), index, ((RedissonObject)instance).encode(element));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Void> fastSet(long index, V element) {
|
|
|
|
|
return commandExecutor.writeReactive(instance.getName(), instance.getCodec(), RedisCommands.LSET, instance.getName(), index, ((RedissonObject)instance).encode(element));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Integer> add(long index, V element) {
|
|
|
|
|
return addAll(index, Collections.singleton(element));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Long> indexOf(final Object o) {
|
|
|
|
|
return commandExecutor.reactive(new Supplier<RFuture<Long>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public RFuture<Long> get() {
|
|
|
|
|
return ((RedissonList)instance).indexOfAsync(o, new LongReplayConvertor());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Publisher<Long> lastIndexOf(final Object o) {
|
|
|
|
|
return commandExecutor.reactive(new Supplier<RFuture<Long>>() {
|
|
|
|
|
@Override
|
|
|
|
|
public RFuture<Long> get() {
|
|
|
|
|
return ((RedissonList)instance).lastIndexOfAsync(o, new LongReplayConvertor());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
Nikita
7 years ago