RPriorityDeque implemented

pull/777/head
Nikita 8 years ago
parent 74d0bfeb19
commit 2ca00fe9a9

@ -52,6 +52,7 @@ import org.redisson.api.RMap;
import org.redisson.api.RMapCache; import org.redisson.api.RMapCache;
import org.redisson.api.RPatternTopic; import org.redisson.api.RPatternTopic;
import org.redisson.api.RPermitExpirableSemaphore; import org.redisson.api.RPermitExpirableSemaphore;
import org.redisson.api.RPriorityDeque;
import org.redisson.api.RPriorityQueue; import org.redisson.api.RPriorityQueue;
import org.redisson.api.RQueue; import org.redisson.api.RQueue;
import org.redisson.api.RReadWriteLock; import org.redisson.api.RReadWriteLock;
@ -615,5 +616,16 @@ public class Redisson implements RedissonClient {
return new RedissonPriorityQueue<V>(codec, connectionManager.getCommandExecutor(), name, this); return new RedissonPriorityQueue<V>(codec, connectionManager.getCommandExecutor(), name, this);
} }
@Override
public <V> RPriorityDeque<V> getPriorityDeque(String name) {
return new RedissonPriorityDeque<V>(connectionManager.getCommandExecutor(), name, this);
}
@Override
public <V> RPriorityDeque<V> getPriorityDeque(String name, Codec codec) {
return new RedissonPriorityDeque<V>(codec, connectionManager.getCommandExecutor(), name, this);
}
} }

@ -0,0 +1,243 @@
/**
* Copyright 2016 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;
import java.util.Iterator;
import java.util.NoSuchElementException;
import org.redisson.api.RFuture;
import org.redisson.api.RPriorityDeque;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.RedisCommand.ValueType;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.convertor.VoidReplayConvertor;
import org.redisson.client.protocol.decoder.ListFirstObjectDecoder;
import org.redisson.command.CommandExecutor;
/**
* Distributed and concurrent implementation of {@link java.util.Queue}
*
* @author Nikita Koksharov
*
* @param <V> the type of elements held in this collection
*/
public class RedissonPriorityDeque<V> extends RedissonPriorityQueue<V> implements RPriorityDeque<V> {
private static final RedisCommand<Void> RPUSH_VOID = new RedisCommand<Void>("RPUSH", new VoidReplayConvertor(), 2, ValueType.OBJECTS);
private static final RedisCommand<Object> LRANGE_SINGLE = new RedisCommand<Object>("LRANGE", new ListFirstObjectDecoder());
protected RedissonPriorityDeque(CommandExecutor commandExecutor, String name, Redisson redisson) {
super(commandExecutor, name, redisson);
}
public RedissonPriorityDeque(Codec codec, CommandExecutor commandExecutor, String name, Redisson redisson) {
super(codec, commandExecutor, name, redisson);
}
@Override
public void addFirst(V e) {
get(addFirstAsync(e));
}
// @Override
public RFuture<Void> addFirstAsync(V e) {
return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_VOID, getName(), e);
}
@Override
public void addLast(V e) {
get(addLastAsync(e));
}
// @Override
public RFuture<Void> addLastAsync(V e) {
return commandExecutor.writeAsync(getName(), codec, RPUSH_VOID, getName(), e);
}
@Override
public Iterator<V> descendingIterator() {
return new Iterator<V>() {
private int currentIndex = size();
private boolean removeExecuted;
@Override
public boolean hasNext() {
int size = size();
return currentIndex > 0 && size > 0;
}
@Override
public V next() {
if (!hasNext()) {
throw new NoSuchElementException("No such element at index " + currentIndex);
}
currentIndex--;
removeExecuted = false;
return RedissonPriorityDeque.this.get(currentIndex);
}
@Override
public void remove() {
if (removeExecuted) {
throw new IllegalStateException("Element been already deleted");
}
RedissonPriorityDeque.this.remove(currentIndex);
currentIndex++;
removeExecuted = true;
}
};
}
// @Override
public RFuture<V> getLastAsync() {
return commandExecutor.readAsync(getName(), codec, LRANGE_SINGLE, getName(), -1, -1);
}
@Override
public V getLast() {
V result = get(getLastAsync());
if (result == null) {
throw new NoSuchElementException();
}
return result;
}
@Override
public boolean offerFirst(V e) {
return get(offerFirstAsync(e));
}
// @Override
public RFuture<Boolean> offerFirstAsync(V e) {
return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_BOOLEAN, getName(), e);
}
// @Override
public RFuture<Boolean> offerLastAsync(V e) {
return offerAsync(e);
}
@Override
public boolean offerLast(V e) {
return get(offerLastAsync(e));
}
// @Override
public RFuture<V> peekFirstAsync() {
return getAsync(0);
}
@Override
public V peekFirst() {
return get(peekFirstAsync());
}
// @Override
public RFuture<V> peekLastAsync() {
return getLastAsync();
}
@Override
public V peekLast() {
return get(getLastAsync());
}
// @Override
public RFuture<V> pollFirstAsync() {
return pollAsync();
}
@Override
public V pollFirst() {
return poll();
}
// @Override
public RFuture<V> pollLastAsync() {
return commandExecutor.writeAsync(getName(), codec, RedisCommands.RPOP, getName());
}
@Override
public V pollLast() {
return get(pollLastAsync());
}
// @Override
public RFuture<V> popAsync() {
return pollAsync();
}
@Override
public V pop() {
return removeFirst();
}
// @Override
public RFuture<Void> pushAsync(V e) {
return addFirstAsync(e);
}
@Override
public void push(V e) {
addFirst(e);
}
// @Override
public RFuture<Boolean> removeFirstOccurrenceAsync(Object o) {
return removeAsync(o, 1);
}
@Override
public boolean removeFirstOccurrence(Object o) {
return remove(o, 1);
}
// @Override
public RFuture<V> removeFirstAsync() {
return pollAsync();
}
// @Override
public RFuture<V> removeLastAsync() {
return commandExecutor.writeAsync(getName(), codec, RedisCommands.RPOP, getName());
}
@Override
public V removeLast() {
V value = get(removeLastAsync());
if (value == null) {
throw new NoSuchElementException();
}
return value;
}
// @Override
public RFuture<Boolean> removeLastOccurrenceAsync(Object o) {
return removeAsync(o, -1);
}
@Override
public boolean removeLastOccurrence(Object o) {
return remove(o, -1);
}
}

@ -0,0 +1,28 @@
/**
* Copyright 2016 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.api;
import java.util.Deque;
/**
*
* @author Nikita Koksharov
*
* @param <V> value type
*/
public interface RPriorityDeque<V> extends Deque<V>, RPriorityQueue<V> {
}

@ -547,7 +547,6 @@ public interface RedissonClient {
*/ */
<V> RQueue<V> getQueue(String name, Codec codec); <V> RQueue<V> getQueue(String name, Codec codec);
/** /**
* Returns priority unbounded queue instance by name. * Returns priority unbounded queue instance by name.
* It uses comparator to sort objects. * It uses comparator to sort objects.
@ -570,6 +569,28 @@ public interface RedissonClient {
*/ */
<V> RPriorityQueue<V> getPriorityQueue(String name, Codec codec); <V> RPriorityQueue<V> getPriorityQueue(String name, Codec codec);
/**
* Returns priority unbounded deque instance by name.
* It uses comparator to sort objects.
*
* @param <V> type of value
* @param name of object
* @return Queue object
*/
<V> RPriorityDeque<V> getPriorityDeque(String name);
/**
* Returns priority unbounded deque instance by name
* using provided codec for queue objects.
* It uses comparator to sort objects.
*
* @param <V> type of value
* @param name - name of object
* @param codec - codec for message
* @return Queue object
*/
<V> RPriorityDeque<V> getPriorityDeque(String name, Codec codec);
/** /**
* Returns unbounded blocking queue instance by name. * Returns unbounded blocking queue instance by name.
* *

Loading…
Cancel
Save