Merge branch 'master' into 3.0.0
# Conflicts: # redisson/src/main/java/org/redisson/command/CommandReactiveService.java # redisson/src/main/java/org/redisson/reactive/NettyFuturePublisher.java # redisson/src/main/java/org/redisson/reactive/RedissonAtomicDoubleReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonAtomicLongReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonBitSetReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonBucketReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonHyperLogLogReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonKeysReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonLockReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonPermitExpirableSemaphoreReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonRateLimiterReactive.java # redisson/src/main/java/org/redisson/reactive/RedissonSemaphoreReactive.javapull/1821/head
Nikita
6 years ago
commit
bf7b04ab87
@ -1,147 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonAtomicDouble;
|
||||
import org.redisson.api.RAtomicDoubleAsync;
|
||||
import org.redisson.api.RAtomicDoubleReactive;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Distributed alternative to the {@link java.util.concurrent.atomic.AtomicLong}
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonAtomicDoubleReactive extends RedissonExpirableReactive implements RAtomicDoubleReactive {
|
||||
|
||||
private final RAtomicDoubleAsync instance;
|
||||
|
||||
public RedissonAtomicDoubleReactive(CommandReactiveExecutor commandExecutor, String name) {
|
||||
this(commandExecutor, name, new RedissonAtomicDouble(commandExecutor, name));
|
||||
}
|
||||
|
||||
public RedissonAtomicDoubleReactive(CommandReactiveExecutor commandExecutor, String name, RAtomicDoubleAsync instance) {
|
||||
super(commandExecutor, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Publisher<Double> addAndGet(final double delta) {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.addAndGetAsync(delta);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> compareAndSet(final double expect, final double update) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.compareAndSetAsync(expect, update);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> decrementAndGet() {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.decrementAndGetAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> get() {
|
||||
return addAndGet(0);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> getAndAdd(final double delta) {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.getAndAddAsync(delta);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Publisher<Double> getAndSet(final double newValue) {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.getAndSetAsync(newValue);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> incrementAndGet() {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.incrementAndGetAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> getAndIncrement() {
|
||||
return getAndAdd(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> getAndDecrement() {
|
||||
return getAndAdd(-1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final double newValue) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(newValue);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Double> getAndDelete() {
|
||||
return reactive(new Supplier<RFuture<Double>>() {
|
||||
@Override
|
||||
public RFuture<Double> get() {
|
||||
return instance.getAndDeleteAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return instance.toString();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,146 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonAtomicLong;
|
||||
import org.redisson.api.RAtomicLongAsync;
|
||||
import org.redisson.api.RAtomicLongReactive;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
|
||||
/**
|
||||
* Distributed alternative to the {@link java.util.concurrent.atomic.AtomicLong}
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonAtomicLongReactive extends RedissonExpirableReactive implements RAtomicLongReactive {
|
||||
|
||||
private final RAtomicLongAsync instance;
|
||||
|
||||
public RedissonAtomicLongReactive(CommandReactiveExecutor commandExecutor, String name) {
|
||||
this(commandExecutor, name, new RedissonAtomicLong(commandExecutor, name));
|
||||
}
|
||||
|
||||
public RedissonAtomicLongReactive(CommandReactiveExecutor commandExecutor, String name, RAtomicLongAsync instance) {
|
||||
super(commandExecutor, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> addAndGet(final long delta) {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.addAndGetAsync(delta);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> compareAndSet(final long expect, final long update) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.compareAndSetAsync(expect, update);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> decrementAndGet() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.decrementAndGetAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> get() {
|
||||
return addAndGet(0);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> getAndAdd(final long delta) {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.getAndAddAsync(delta);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> getAndDelete() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.getAndDeleteAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> getAndSet(final long newValue) {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.getAndSetAsync(newValue);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> incrementAndGet() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.incrementAndGetAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> getAndIncrement() {
|
||||
return getAndAdd(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> getAndDecrement() {
|
||||
return getAndAdd(-1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final long newValue) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(newValue);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return instance.toString();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,226 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.BitSet;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonBitSet;
|
||||
import org.redisson.api.RBitSetAsync;
|
||||
import org.redisson.api.RBitSetReactive;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.client.codec.BitSetCodec;
|
||||
import org.redisson.client.protocol.RedisCommands;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonBitSetReactive extends RedissonExpirableReactive implements RBitSetReactive {
|
||||
|
||||
private final RBitSetAsync instance;
|
||||
|
||||
public RedissonBitSetReactive(CommandReactiveExecutor connectionManager, String name) {
|
||||
this(connectionManager, name, new RedissonBitSet(connectionManager, name));
|
||||
}
|
||||
|
||||
public RedissonBitSetReactive(CommandReactiveExecutor connectionManager, String name, RBitSetAsync instance) {
|
||||
super(connectionManager, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
public Publisher<Boolean> get(final long bitIndex) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.getAsync(bitIndex);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public Publisher<Boolean> set(final long bitIndex, final boolean value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.setAsync(bitIndex, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public Publisher<byte[]> toByteArray() {
|
||||
return reactive(new Supplier<RFuture<byte[]>>() {
|
||||
@Override
|
||||
public RFuture<byte[]> get() {
|
||||
return instance.toByteArrayAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public Publisher<BitSet> asBitSet() {
|
||||
return commandExecutor.readReactive(getName(), BitSetCodec.INSTANCE, RedisCommands.GET, getName());
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> length() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.lengthAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final long fromIndex, final long toIndex, final boolean value) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(fromIndex, toIndex, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> clear(final long fromIndex, final long toIndex) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.clearAsync(fromIndex, toIndex);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final BitSet bs) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(bs);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> not() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.notAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final long fromIndex, final long toIndex) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(fromIndex, toIndex);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> size() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.sizeAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> set(final long bitIndex) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.setAsync(bitIndex);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> cardinality() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.cardinalityAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> clear(final long bitIndex) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.clearAsync(bitIndex);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> clear() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.clearAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> or(final String... bitSetNames) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.orAsync(bitSetNames);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> and(final String... bitSetNames) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.andAsync(bitSetNames);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> xor(final String... bitSetNames) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.xorAsync(bitSetNames);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return Mono.from(asBitSet()).block().toString();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,147 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonBucket;
|
||||
import org.redisson.api.RBucketAsync;
|
||||
import org.redisson.api.RBucketReactive;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <V> value type
|
||||
*/
|
||||
public class RedissonBucketReactive<V> extends RedissonExpirableReactive implements RBucketReactive<V> {
|
||||
|
||||
private final RBucketAsync<V> instance;
|
||||
|
||||
public RedissonBucketReactive(CommandReactiveExecutor connectionManager, String name) {
|
||||
this(connectionManager, name, new RedissonBucket<V>(connectionManager, name));
|
||||
}
|
||||
|
||||
public RedissonBucketReactive(CommandReactiveExecutor connectionManager, String name, RBucketAsync<V> instance) {
|
||||
super(connectionManager, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
public RedissonBucketReactive(Codec codec, CommandReactiveExecutor connectionManager, String name) {
|
||||
this(codec, connectionManager, name, new RedissonBucket<V>(codec, connectionManager, name));
|
||||
}
|
||||
|
||||
public RedissonBucketReactive(Codec codec, CommandReactiveExecutor connectionManager, String name, RBucketAsync<V> instance) {
|
||||
super(codec, connectionManager, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<V> get() {
|
||||
return reactive(new Supplier<RFuture<V>>() {
|
||||
@Override
|
||||
public RFuture<V> get() {
|
||||
return instance.getAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<V> getAndDelete() {
|
||||
return reactive(new Supplier<RFuture<V>>() {
|
||||
@Override
|
||||
public RFuture<V> get() {
|
||||
return instance.getAndDeleteAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final V value) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> set(final V value, final long timeToLive, final TimeUnit timeUnit) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.setAsync(value, timeToLive, timeUnit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> size() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.sizeAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> trySet(final V value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.trySetAsync(value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> trySet(final V value, final long timeToLive, final TimeUnit timeUnit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.trySetAsync(value, timeToLive, timeUnit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> compareAndSet(final V expect, final V update) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.compareAndSetAsync(expect, update);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<V> getAndSet(final V newValue) {
|
||||
return reactive(new Supplier<RFuture<V>>() {
|
||||
@Override
|
||||
public RFuture<V> get() {
|
||||
return instance.getAndSetAsync(newValue);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,99 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonHyperLogLog;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RHyperLogLogAsync;
|
||||
import org.redisson.api.RHyperLogLogReactive;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <V> value type
|
||||
*/
|
||||
public class RedissonHyperLogLogReactive<V> extends RedissonExpirableReactive implements RHyperLogLogReactive<V> {
|
||||
|
||||
private final RHyperLogLogAsync<V> instance;
|
||||
|
||||
public RedissonHyperLogLogReactive(CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(commandExecutor, name, new RedissonHyperLogLog<V>(commandExecutor, name));
|
||||
this.instance = (RHyperLogLogAsync<V>) super.instance;
|
||||
}
|
||||
|
||||
public RedissonHyperLogLogReactive(Codec codec, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(codec, commandExecutor, name, new RedissonHyperLogLog<V>(commandExecutor, name));
|
||||
this.instance = (RHyperLogLogAsync<V>) super.instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> add(final V obj) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.addAsync(obj);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> addAll(final Collection<V> objects) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.addAllAsync(objects);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> count() {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.countAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> countWith(final String... otherLogNames) {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.countWithAsync(otherLogNames);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> mergeWith(final String... otherLogNames) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.mergeWithAsync(otherLogNames);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,166 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonLock;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RLockAsync;
|
||||
import org.redisson.api.RLockReactive;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonLockReactive extends RedissonExpirableReactive implements RLockReactive {
|
||||
|
||||
private final RLockAsync instance;
|
||||
|
||||
public RedissonLockReactive(CommandReactiveExecutor connectionManager, String name) {
|
||||
this(connectionManager, name, new RedissonLock(connectionManager, name));
|
||||
}
|
||||
|
||||
public RedissonLockReactive(CommandReactiveExecutor connectionManager, String name, RLockAsync instance) {
|
||||
super(connectionManager, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> forceUnlock() {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.forceUnlockAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> unlock() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.unlockAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> unlock(final long threadId) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.unlockAsync(threadId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryLock() {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryLockAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> lock() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.lockAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> lock(final long threadId) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.lockAsync(threadId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> lock(final long leaseTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.lockAsync(leaseTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> lock(final long leaseTime, final TimeUnit unit, final long threadId) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.lockAsync(leaseTime, unit, threadId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryLock(final long threadId) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryLockAsync(threadId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryLock(final long waitTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryLockAsync(waitTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryLock(final long waitTime, final long leaseTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryLockAsync(waitTime, leaseTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryLock(final long waitTime, final long leaseTime, final TimeUnit unit, final long threadId) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryLockAsync(waitTime, leaseTime, unit, threadId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,160 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonLock;
|
||||
import org.redisson.RedissonPermitExpirableSemaphore;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RLockAsync;
|
||||
import org.redisson.api.RPermitExpirableSemaphoreAsync;
|
||||
import org.redisson.api.RPermitExpirableSemaphoreReactive;
|
||||
import org.redisson.command.CommandAsyncExecutor;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
import org.redisson.pubsub.SemaphorePubSub;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonPermitExpirableSemaphoreReactive extends RedissonExpirableReactive implements RPermitExpirableSemaphoreReactive {
|
||||
|
||||
private final RPermitExpirableSemaphoreAsync instance;
|
||||
|
||||
public RedissonPermitExpirableSemaphoreReactive(CommandReactiveExecutor connectionManager, String name, SemaphorePubSub semaphorePubSub) {
|
||||
super(connectionManager, name, new RedissonPermitExpirableSemaphore(connectionManager, name, semaphorePubSub));
|
||||
instance = (RPermitExpirableSemaphoreAsync) super.instance;
|
||||
}
|
||||
|
||||
protected RLockAsync createLock(CommandAsyncExecutor connectionManager, String name) {
|
||||
return new RedissonLock(commandExecutor, name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<String> acquire() {
|
||||
return reactive(new Supplier<RFuture<String>>() {
|
||||
@Override
|
||||
public RFuture<String> get() {
|
||||
return instance.acquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<String> acquire(final long leaseTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<String>>() {
|
||||
@Override
|
||||
public RFuture<String> get() {
|
||||
return instance.acquireAsync(leaseTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<String> tryAcquire() {
|
||||
return reactive(new Supplier<RFuture<String>>() {
|
||||
@Override
|
||||
public RFuture<String> get() {
|
||||
return instance.tryAcquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<String> tryAcquire(final long waitTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<String>>() {
|
||||
@Override
|
||||
public RFuture<String> get() {
|
||||
return instance.tryAcquireAsync(waitTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<String> tryAcquire(final long waitTime, final long leaseTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<String>>() {
|
||||
@Override
|
||||
public RFuture<String> get() {
|
||||
return instance.tryAcquireAsync(waitTime, leaseTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryRelease(final String permitId) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryReleaseAsync(permitId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> release(final String permitId) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.releaseAsync(permitId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Integer> availablePermits() {
|
||||
return reactive(new Supplier<RFuture<Integer>>() {
|
||||
@Override
|
||||
public RFuture<Integer> get() {
|
||||
return instance.availablePermitsAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> trySetPermits(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.trySetPermitsAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> addPermits(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.addPermitsAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> updateLeaseTime(final String permitId, final long leaseTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.updateLeaseTimeAsync(permitId, leaseTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,119 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonRateLimiter;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RRateLimiterAsync;
|
||||
import org.redisson.api.RRateLimiterReactive;
|
||||
import org.redisson.api.RateIntervalUnit;
|
||||
import org.redisson.api.RateType;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonRateLimiterReactive extends RedissonObjectReactive implements RRateLimiterReactive {
|
||||
|
||||
private final RRateLimiterAsync instance;
|
||||
|
||||
public RedissonRateLimiterReactive(CommandReactiveExecutor connectionManager, String name) {
|
||||
this(connectionManager, name, new RedissonRateLimiter(connectionManager, name));
|
||||
}
|
||||
|
||||
private RedissonRateLimiterReactive(CommandReactiveExecutor connectionManager, String name, RRateLimiterAsync instance) {
|
||||
super(connectionManager, name, instance);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> trySetRate(final RateType mode, final long rate, final long rateInterval,
|
||||
final RateIntervalUnit rateIntervalUnit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.trySetRateAsync(mode, rate, rateInterval, rateIntervalUnit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire() {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final long permits) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> acquire() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.acquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> acquire(final long permits) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.acquireAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final long timeout, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(timeout, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final long permits, final long timeout, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(permits, timeout, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,143 +0,0 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.redisson.reactive;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonSemaphore;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RSemaphoreAsync;
|
||||
import org.redisson.api.RSemaphoreReactive;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
import org.redisson.pubsub.SemaphorePubSub;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonSemaphoreReactive extends RedissonExpirableReactive implements RSemaphoreReactive {
|
||||
|
||||
private final RSemaphoreAsync instance;
|
||||
|
||||
public RedissonSemaphoreReactive(CommandReactiveExecutor connectionManager, String name, SemaphorePubSub semaphorePubSub) {
|
||||
super(connectionManager, name, new RedissonSemaphore(connectionManager, name, semaphorePubSub));
|
||||
instance = (RSemaphoreAsync) super.instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire() {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> acquire() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.acquireAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> acquire(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.acquireAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> release() {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.releaseAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> release(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.releaseAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> trySetPermits(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.trySetPermitsAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final long waitTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(waitTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> tryAcquire(final int permits, final long waitTime, final TimeUnit unit) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.tryAcquireAsync(permits, waitTime, unit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> reducePermits(final int permits) {
|
||||
return reactive(new Supplier<RFuture<Void>>() {
|
||||
@Override
|
||||
public RFuture<Void> get() {
|
||||
return instance.reducePermitsAsync(permits);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue