RQueueAsync interface added. #186

pull/243/head
Nikita 10 years ago
parent 6ec833973a
commit 5f2185beec

@ -41,6 +41,11 @@ public class RedissonQueue<V> extends RedissonList<V> implements RQueue<V> {
return add(e);
}
@Override
public Future<Boolean> offerAsync(V e) {
return addAsync(e);
}
public V getFirst() {
V value = connectionManager.read(getName(), RedisCommands.LINDEX, getName(), 0);
if (value == null) {
@ -62,9 +67,14 @@ public class RedissonQueue<V> extends RedissonList<V> implements RQueue<V> {
return removeFirst();
}
@Override
public Future<V> pollAsync() {
return connectionManager.writeAsync(getName(), RedisCommands.LPOP, getName());
}
@Override
public V poll() {
return connectionManager.write(getName(), RedisCommands.LPOP, getName());
return connectionManager.get(pollAsync());
}
@Override
@ -72,6 +82,11 @@ public class RedissonQueue<V> extends RedissonList<V> implements RQueue<V> {
return getFirst();
}
@Override
public Future<V> peekAsync() {
return getAsync(0);
}
@Override
public V peek() {
if (isEmpty()) {

@ -0,0 +1,26 @@
package org.redisson.core;
import java.util.Collection;
import java.util.List;
import io.netty.util.concurrent.Future;
public interface RCollectionAsync<V> extends RExpirableAsync {
Future<Boolean> retainAllAsync(Collection<?> c);
Future<Boolean> removeAllAsync(Collection<?> c);
Future<Boolean> containsAllAsync(Collection<?> c);
Future<Boolean> removeAsync(Object o);
Future<List<V>> readAllAsync();
Future<Integer> sizeAsync();
Future<Boolean> addAsync(V e);
Future<Boolean> addAllAsync(Collection<? extends V> c);
}

@ -17,9 +17,6 @@ package org.redisson.core;
import io.netty.util.concurrent.Future;
import java.util.Collection;
import java.util.List;
/**
* Async list functions
*
@ -27,7 +24,7 @@ import java.util.List;
*
* @param <V> the type of elements held in this collection
*/
public interface RListAsync<V> extends RExpirableAsync {
public interface RListAsync<V> extends RCollectionAsync<V> {
Future<Integer> lastIndexOfAsync(Object o);
@ -37,22 +34,6 @@ public interface RListAsync<V> extends RExpirableAsync {
Future<V> setAsync(int index, V element);
Future<Boolean> retainAllAsync(Collection<?> c);
Future<Boolean> removeAllAsync(Collection<?> c);
Future<Boolean> containsAllAsync(Collection<?> c);
Future<Boolean> removeAsync(Object o);
Future<List<V>> readAllAsync();
Future<Integer> sizeAsync();
Future<V> getAsync(int index);
Future<Boolean> addAsync(V e);
Future<Boolean> addAllAsync(Collection<? extends V> c);
}

@ -17,8 +17,6 @@ package org.redisson.core;
import java.util.Queue;
import io.netty.util.concurrent.Future;
/**
* {@link java.util.Queue} backed by Redis
*
@ -26,11 +24,7 @@ import io.netty.util.concurrent.Future;
*
* @param <V> the type of elements held in this collection
*/
public interface RQueue<V> extends Queue<V>, RExpirable {
Future<V> pollLastAndOfferFirstToAsync(RQueue<V> queue);
Future<V> pollLastAndOfferFirstToAsync(String queueName);
public interface RQueue<V> extends Queue<V>, RExpirable, RQueueAsync<V> {
V pollLastAndOfferFirstTo(String dequeName);

@ -0,0 +1,39 @@
/**
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
*
* 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.core;
import io.netty.util.concurrent.Future;
/**
* {@link java.util.Queue} backed by Redis
*
* @author Nikita Koksharov
*
* @param <V> the type of elements held in this collection
*/
public interface RQueueAsync<V> extends RCollectionAsync<V> {
Future<V> peekAsync();
Future<V> pollAsync();
Future<Boolean> offerAsync(V e);
Future<V> pollLastAndOfferFirstToAsync(RQueue<V> queue);
Future<V> pollLastAndOfferFirstToAsync(String queueName);
}
Loading…
Cancel
Save