Merge branch 'master' into 3.0.0
# Conflicts: # pom.xml # redisson-all/pom.xml # redisson-tomcat/pom.xml # redisson-tomcat/redisson-tomcat-6/pom.xml # redisson-tomcat/redisson-tomcat-7/pom.xml # redisson-tomcat/redisson-tomcat-8/pom.xml # redisson/pom.xmlpull/1303/head
Nikita
7 years ago
commit
9ccdb318a5
@ -0,0 +1,82 @@
|
||||
/**
|
||||
* 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.List;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public interface RListMultimapReactive<K, V> extends RMultimapReactive<K, V> {
|
||||
|
||||
/**
|
||||
* Returns a view List of the values associated with {@code key} in this
|
||||
* multimap, if any. Note that when {@code containsKey(key)} is false, this
|
||||
* returns an empty collection, not {@code null}.
|
||||
*
|
||||
* <p>Changes to the returned collection will update the underlying multimap,
|
||||
* and vice versa.
|
||||
*
|
||||
* @param key - map key
|
||||
* @return list of values
|
||||
*/
|
||||
RListReactive<V> get(K key);
|
||||
|
||||
/**
|
||||
* Returns all elements at once. Result Set is <b>NOT</b> backed by map,
|
||||
* so changes are not reflected in map.
|
||||
*
|
||||
* @param key - map key
|
||||
* @return list of values
|
||||
*/
|
||||
Publisher<List<V>> getAll(K key);
|
||||
|
||||
/**
|
||||
* Removes all values associated with the key {@code key}.
|
||||
*
|
||||
* <p>Once this method returns, {@code key} will not be mapped to any values
|
||||
* <p>Use {@link RMultimapReactive#fastRemove} if values are not needed.</p>
|
||||
*
|
||||
* @param key - map key
|
||||
* @return the values that were removed (possibly empty). The returned
|
||||
* list <i>may</i> be modifiable, but updating it will have no
|
||||
* effect on the multimap.
|
||||
*/
|
||||
Publisher<List<V>> removeAll(Object key);
|
||||
|
||||
/**
|
||||
* Stores a collection of values with the same key, replacing any existing
|
||||
* values for that key.
|
||||
*
|
||||
* <p>If {@code values} is empty, this is equivalent to
|
||||
* {@link #removeAll(Object)}.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param values - map values
|
||||
* @return list of replaced values, or an empty collection if no
|
||||
* values were previously associated with the key. List
|
||||
* <i>may</i> be modifiable, but updating it will have no effect on the
|
||||
* multimap.
|
||||
*/
|
||||
Publisher<List<V>> replaceValues(K key, Iterable<? extends V> values);
|
||||
|
||||
}
|
||||
@ -0,0 +1,139 @@
|
||||
/**
|
||||
* 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.Set;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public interface RMultimapReactive<K, V> {
|
||||
|
||||
/**
|
||||
* Returns the number of key-value pairs in this multimap.
|
||||
*
|
||||
* @return size of multimap
|
||||
*/
|
||||
Publisher<Integer> size();
|
||||
|
||||
/**
|
||||
* Returns {@code true} if this multimap contains at least one key-value pair
|
||||
* with the key {@code key}.
|
||||
*
|
||||
* @param key - map key
|
||||
* @return <code>true</code> if contains a key
|
||||
*/
|
||||
Publisher<Boolean> containsKey(Object key);
|
||||
|
||||
/**
|
||||
* Returns {@code true} if this multimap contains at least one key-value pair
|
||||
* with the value {@code value}.
|
||||
*
|
||||
* @param value - map value
|
||||
* @return <code>true</code> if contains a value
|
||||
*/
|
||||
Publisher<Boolean> containsValue(Object value);
|
||||
|
||||
/**
|
||||
* Returns {@code true} if this multimap contains at least one key-value pair
|
||||
* with the key {@code key} and the value {@code value}.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param value - map value
|
||||
* @return <code>true</code> if contains an entry
|
||||
*/
|
||||
Publisher<Boolean> containsEntry(Object key, Object value);
|
||||
|
||||
/**
|
||||
* Stores a key-value pair in this multimap.
|
||||
*
|
||||
* <p>Some multimap implementations allow duplicate key-value pairs, in which
|
||||
* case {@code put} always adds a new key-value pair and increases the
|
||||
* multimap size by 1. Other implementations prohibit duplicates, and storing
|
||||
* a key-value pair that's already in the multimap has no effect.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param value - map value
|
||||
* @return {@code true} if the method increased the size of the multimap, or
|
||||
* {@code false} if the multimap already contained the key-value pair and
|
||||
* doesn't allow duplicates
|
||||
*/
|
||||
Publisher<Boolean> put(K key, V value);
|
||||
|
||||
/**
|
||||
* Removes a single key-value pair with the key {@code key} and the value
|
||||
* {@code value} from this multimap, if such exists. If multiple key-value
|
||||
* pairs in the multimap fit this description, which one is removed is
|
||||
* unspecified.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param value - map value
|
||||
* @return {@code true} if the multimap changed
|
||||
*/
|
||||
Publisher<Boolean> remove(Object key, Object value);
|
||||
|
||||
// Bulk Operations
|
||||
|
||||
/**
|
||||
* Stores a key-value pair in this multimap for each of {@code values}, all
|
||||
* using the same key, {@code key}. Equivalent to (but expected to be more
|
||||
* efficient than): <pre> {@code
|
||||
*
|
||||
* for (V value : values) {
|
||||
* put(key, value);
|
||||
* }}</pre>
|
||||
*
|
||||
* <p>In particular, this is a no-op if {@code values} is empty.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param values - map values
|
||||
* @return {@code true} if the multimap changed
|
||||
*/
|
||||
Publisher<Boolean> putAll(K key, Iterable<? extends V> values);
|
||||
|
||||
/**
|
||||
* Returns the number of key-value pairs in this multimap.
|
||||
*
|
||||
* @return keys amount
|
||||
*/
|
||||
Publisher<Integer> keySize();
|
||||
|
||||
/**
|
||||
* Removes <code>keys</code> from map by one operation
|
||||
*
|
||||
* Works faster than <code>RMultimap.remove</code> but not returning
|
||||
* the value associated with <code>key</code>
|
||||
*
|
||||
* @param keys - map keys
|
||||
* @return the number of keys that were removed from the hash, not including specified but non existing keys
|
||||
*/
|
||||
Publisher<Long> fastRemove(K ... keys);
|
||||
|
||||
/**
|
||||
* Read all keys at once
|
||||
*
|
||||
* @return keys
|
||||
*/
|
||||
Publisher<Set<K>> readAllKeySet();
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,82 @@
|
||||
/**
|
||||
* 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.Set;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public interface RSetMultimapReactive<K, V> extends RMultimapReactive<K, V> {
|
||||
|
||||
/**
|
||||
* Returns a view Set of the values associated with {@code key} in this
|
||||
* multimap, if any. Note that when {@code containsKey(key)} is false, this
|
||||
* returns an empty collection, not {@code null}.
|
||||
*
|
||||
* <p>Changes to the returned collection will update the underlying multimap,
|
||||
* and vice versa.
|
||||
*
|
||||
* @param key - map key
|
||||
* @return set of values
|
||||
*/
|
||||
RSetReactive<V> get(K key);
|
||||
|
||||
/**
|
||||
* Returns all elements at once. Result Set is <b>NOT</b> backed by map,
|
||||
* so changes are not reflected in map.
|
||||
*
|
||||
* @param key - map key
|
||||
* @return set of values
|
||||
*/
|
||||
Publisher<Set<V>> getAll(K key);
|
||||
|
||||
/**
|
||||
* Removes all values associated with the key {@code key}.
|
||||
*
|
||||
* <p>Once this method returns, {@code key} will not be mapped to any values
|
||||
* <p>Use {@link RMultimapReactive#fastRemove} if values are not needed.</p>
|
||||
*
|
||||
* @param key - map key
|
||||
* @return the values that were removed (possibly empty). The returned
|
||||
* set <i>may</i> be modifiable, but updating it will have no
|
||||
* effect on the multimap.
|
||||
*/
|
||||
Publisher<Set<V>> removeAll(Object key);
|
||||
|
||||
/**
|
||||
* Stores a collection of values with the same key, replacing any existing
|
||||
* values for that key.
|
||||
*
|
||||
* <p>If {@code values} is empty, this is equivalent to
|
||||
* {@link #removeAll(Object)}.
|
||||
*
|
||||
* @param key - map key
|
||||
* @param values - map values
|
||||
* @return set of replaced values, or an empty collection if no
|
||||
* values were previously associated with the key. Set
|
||||
* <i>may</i> be modifiable, but updating it will have no effect on the
|
||||
* multimap.
|
||||
*/
|
||||
Publisher<Set<V>> replaceValues(K key, Iterable<? extends V> values);
|
||||
|
||||
}
|
||||
@ -1,182 +0,0 @@
|
||||
/**
|
||||
* 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.client.protocol.decoder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
|
||||
import org.redisson.client.handler.State;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <T> object type
|
||||
*/
|
||||
public class NestedMultiDecoder<T> implements MultiDecoder<Object> {
|
||||
|
||||
public static class NestedDecoderState implements DecoderState {
|
||||
|
||||
int decoderIndex;
|
||||
|
||||
int flipDecoderIndex;
|
||||
|
||||
public NestedDecoderState() {
|
||||
}
|
||||
|
||||
public NestedDecoderState(int decoderIndex, int flipDecoderIndex) {
|
||||
super();
|
||||
this.decoderIndex = decoderIndex;
|
||||
this.flipDecoderIndex = flipDecoderIndex;
|
||||
}
|
||||
|
||||
public int getDecoderIndex() {
|
||||
return decoderIndex;
|
||||
}
|
||||
public void resetDecoderIndex() {
|
||||
decoderIndex = 0;
|
||||
}
|
||||
public void incDecoderIndex() {
|
||||
decoderIndex++;
|
||||
}
|
||||
|
||||
public int getFlipDecoderIndex() {
|
||||
return flipDecoderIndex;
|
||||
}
|
||||
public void resetFlipDecoderIndex() {
|
||||
flipDecoderIndex = 0;
|
||||
}
|
||||
public void incFlipDecoderIndex() {
|
||||
flipDecoderIndex++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DecoderState copy() {
|
||||
return new NestedDecoderState(decoderIndex, flipDecoderIndex);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "NestedDecoderState [decoderIndex=" + decoderIndex + ", flipDecoderIndex=" + flipDecoderIndex + "]";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected final MultiDecoder<Object> firstDecoder;
|
||||
protected final MultiDecoder<Object> secondDecoder;
|
||||
private MultiDecoder<Object> thirdDecoder;
|
||||
private boolean handleEmpty;
|
||||
|
||||
public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder) {
|
||||
this(firstDecoder, secondDecoder, false);
|
||||
}
|
||||
|
||||
public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, boolean handleEmpty) {
|
||||
this(firstDecoder, secondDecoder, null, handleEmpty);
|
||||
}
|
||||
|
||||
public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, MultiDecoder<Object> thirdDecoder) {
|
||||
this(firstDecoder, secondDecoder, thirdDecoder, false);
|
||||
}
|
||||
|
||||
public NestedMultiDecoder(MultiDecoder<Object> firstDecoder, MultiDecoder<Object> secondDecoder, MultiDecoder<Object> thirdDecoder, boolean handleEmpty) {
|
||||
this.firstDecoder = firstDecoder;
|
||||
this.secondDecoder = secondDecoder;
|
||||
this.thirdDecoder = thirdDecoder;
|
||||
this.handleEmpty = handleEmpty;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object decode(ByteBuf buf, State state) throws IOException {
|
||||
NestedDecoderState ds = getDecoder(state);
|
||||
|
||||
MultiDecoder<?> decoder = null;
|
||||
if (ds.getFlipDecoderIndex() == 2) {
|
||||
decoder = firstDecoder;
|
||||
}
|
||||
if (ds.getFlipDecoderIndex() == 1) {
|
||||
decoder = secondDecoder;
|
||||
}
|
||||
|
||||
return decoder.decode(buf, state);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isApplicable(int paramNum, State state) {
|
||||
NestedDecoderState ds = getDecoder(state);
|
||||
if (paramNum == 0) {
|
||||
ds.incFlipDecoderIndex();
|
||||
ds.resetDecoderIndex();
|
||||
}
|
||||
// used only with thirdDecoder
|
||||
if (ds.getFlipDecoderIndex() == 3) {
|
||||
ds.resetFlipDecoderIndex();
|
||||
ds.incFlipDecoderIndex();
|
||||
}
|
||||
|
||||
MultiDecoder<?> decoder = null;
|
||||
if (ds.getFlipDecoderIndex() == 2) {
|
||||
decoder = firstDecoder;
|
||||
}
|
||||
if (ds.getFlipDecoderIndex() == 1) {
|
||||
decoder = secondDecoder;
|
||||
}
|
||||
|
||||
return decoder.isApplicable(paramNum, state);
|
||||
}
|
||||
|
||||
protected final NestedDecoderState getDecoder(State state) {
|
||||
NestedDecoderState ds = state.getDecoderState();
|
||||
if (ds == null) {
|
||||
ds = new NestedDecoderState();
|
||||
state.setDecoderState(ds);
|
||||
}
|
||||
return ds;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object decode(List<Object> parts, State state) {
|
||||
if (parts.isEmpty() && state.getDecoderState() == null && handleEmpty) {
|
||||
MultiDecoder<?> decoder = secondDecoder;
|
||||
if (thirdDecoder != null) {
|
||||
decoder = thirdDecoder;
|
||||
}
|
||||
return decoder.decode(parts, state);
|
||||
}
|
||||
|
||||
NestedDecoderState ds = getDecoder(state);
|
||||
if (parts.isEmpty()) {
|
||||
ds.resetDecoderIndex();
|
||||
}
|
||||
|
||||
ds.incDecoderIndex();
|
||||
MultiDecoder<?> decoder = null;
|
||||
if (ds.getDecoderIndex() == 1) {
|
||||
decoder = firstDecoder;
|
||||
}
|
||||
if (ds.getDecoderIndex() == 2) {
|
||||
decoder = secondDecoder;
|
||||
}
|
||||
if (ds.getDecoderIndex() == 3) {
|
||||
decoder = thirdDecoder;
|
||||
}
|
||||
|
||||
return decoder.decode(parts, state);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,180 @@
|
||||
/**
|
||||
* 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.reactive;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.api.RFuture;
|
||||
import org.redisson.api.RMultimap;
|
||||
import org.redisson.api.RMultimapReactive;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.client.protocol.RedisCommand;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
import org.redisson.misc.Hash;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import reactor.fn.Supplier;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
abstract class RedissonBaseMultimapReactive<K, V> extends RedissonExpirableReactive implements RMultimapReactive<K, V> {
|
||||
|
||||
private final RMultimap<K, V> instance;
|
||||
|
||||
public RedissonBaseMultimapReactive(RMultimap<K, V> instance, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(commandExecutor, name);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
public RedissonBaseMultimapReactive(RMultimap<K, V> instance, Codec codec, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(codec, commandExecutor, name);
|
||||
this.instance = instance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Integer> size() {
|
||||
return reactive(new Supplier<RFuture<Integer>>() {
|
||||
@Override
|
||||
public RFuture<Integer> get() {
|
||||
return instance.sizeAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> containsKey(final Object key) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.containsKeyAsync(key);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> containsValue(final Object value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.containsValueAsync(value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> containsEntry(final Object key, final Object value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.containsEntryAsync(key, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> put(final K key, final V value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.putAsync(key, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> remove(final Object key, final Object value) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.removeAsync(key, value);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> putAll(final K key, final Iterable<? extends V> values) {
|
||||
return reactive(new Supplier<RFuture<Boolean>>() {
|
||||
@Override
|
||||
public RFuture<Boolean> get() {
|
||||
return instance.putAllAsync(key, values);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Integer> keySize() {
|
||||
return reactive(new Supplier<RFuture<Integer>>() {
|
||||
@Override
|
||||
public RFuture<Integer> get() {
|
||||
return instance.keySizeAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> fastRemove(final K... keys) {
|
||||
return reactive(new Supplier<RFuture<Long>>() {
|
||||
@Override
|
||||
public RFuture<Long> get() {
|
||||
return instance.fastRemoveAsync(keys);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Set<K>> readAllKeySet() {
|
||||
return reactive(new Supplier<RFuture<Set<K>>>() {
|
||||
@Override
|
||||
public RFuture<Set<K>> get() {
|
||||
return instance.readAllKeySetAsync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected String hash(ByteBuf objectState) {
|
||||
return Hash.hashToBase64(objectState);
|
||||
}
|
||||
|
||||
protected String hashAndRelease(ByteBuf objectState) {
|
||||
try {
|
||||
return Hash.hashToBase64(objectState);
|
||||
} finally {
|
||||
objectState.release();
|
||||
}
|
||||
}
|
||||
|
||||
String getValuesName(String hash) {
|
||||
return "{" + getName() + "}:" + hash;
|
||||
}
|
||||
|
||||
protected <T> Publisher<T> fastRemove(List<Object> mapKeys, List<Object> listKeys, RedisCommand<T> evalCommandType) {
|
||||
return commandExecutor.evalWriteReactive(getName(), codec, evalCommandType,
|
||||
"local res = redis.call('hdel', KEYS[1], unpack(ARGV)); " +
|
||||
"if res > 0 then " +
|
||||
"redis.call('del', unpack(KEYS, 2, #KEYS)); " +
|
||||
"end; " +
|
||||
"return res; ",
|
||||
listKeys, mapKeys.toArray());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,143 @@
|
||||
/**
|
||||
* 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.reactive;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonListMultimap;
|
||||
import org.redisson.api.RListMultimapReactive;
|
||||
import org.redisson.api.RListReactive;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.client.protocol.RedisCommands;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public class RedissonListMultimapReactive<K, V> extends RedissonBaseMultimapReactive<K, V> implements RListMultimapReactive<K, V> {
|
||||
|
||||
public RedissonListMultimapReactive(UUID id, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(new RedissonListMultimap<K, V>(id, commandExecutor, name), commandExecutor, name);
|
||||
}
|
||||
|
||||
public RedissonListMultimapReactive(UUID id, Codec codec, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(new RedissonListMultimap<K, V>(id, codec, commandExecutor, name), codec, commandExecutor, name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RListReactive<V> get(final K key) {
|
||||
final ByteBuf keyState = encodeMapKey(key);
|
||||
final String keyHash = hashAndRelease(keyState);
|
||||
final String setName = getValuesName(keyHash);
|
||||
|
||||
return new RedissonListReactive<V>(codec, commandExecutor, setName) {
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> delete() {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
return RedissonListMultimapReactive.this.fastRemove(Arrays.<Object>asList(keyState), Arrays.<Object>asList(setName), RedisCommands.EVAL_BOOLEAN_AMOUNT);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> clearExpire() {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> expire(long timeToLive, TimeUnit timeUnit) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> expireAt(long timestamp) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> remainTimeToLive() {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> rename(String newName) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> renamenx(String newName) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<List<V>> getAll(K key) {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
String keyHash = hashAndRelease(keyState);
|
||||
String setName = getValuesName(keyHash);
|
||||
|
||||
return commandExecutor.readReactive(getName(), codec, RedisCommands.LRANGE, setName, 0, -1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<List<V>> removeAll(Object key) {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
String keyHash = hash(keyState);
|
||||
|
||||
String setName = getValuesName(keyHash);
|
||||
return commandExecutor.evalWriteReactive(getName(), codec, RedisCommands.EVAL_LIST,
|
||||
"redis.call('hdel', KEYS[1], ARGV[1]); " +
|
||||
"local members = redis.call('lrange', KEYS[2], 0, -1); " +
|
||||
"redis.call('del', KEYS[2]); " +
|
||||
"return members; ",
|
||||
Arrays.<Object>asList(getName(), setName), keyState);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<List<V>> replaceValues(K key, Iterable<? extends V> values) {
|
||||
List<Object> params = new ArrayList<Object>();
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
params.add(keyState);
|
||||
String keyHash = hash(keyState);
|
||||
params.add(keyHash);
|
||||
for (Object value : values) {
|
||||
ByteBuf valueState = encodeMapValue(value);
|
||||
params.add(valueState);
|
||||
}
|
||||
|
||||
String setName = getValuesName(keyHash);
|
||||
return commandExecutor.evalWriteReactive(getName(), codec, RedisCommands.EVAL_LIST,
|
||||
"redis.call('hset', KEYS[1], ARGV[1], ARGV[2]); " +
|
||||
"local members = redis.call('lrange', KEYS[2], 0, -1); " +
|
||||
"redis.call('del', KEYS[2]); " +
|
||||
"redis.call('rpush', KEYS[2], unpack(ARGV, 3, #ARGV)); " +
|
||||
"return members; ",
|
||||
Arrays.<Object>asList(getName(), setName), params.toArray());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,144 @@
|
||||
/**
|
||||
* 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.reactive;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.redisson.RedissonSetMultimap;
|
||||
import org.redisson.api.RSetMultimapReactive;
|
||||
import org.redisson.api.RSetReactive;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.client.protocol.RedisCommands;
|
||||
import org.redisson.command.CommandReactiveExecutor;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
* @param <K> key type
|
||||
* @param <V> value type
|
||||
*/
|
||||
public class RedissonSetMultimapReactive<K, V> extends RedissonBaseMultimapReactive<K, V> implements RSetMultimapReactive<K, V> {
|
||||
|
||||
public RedissonSetMultimapReactive(UUID id, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(new RedissonSetMultimap<K, V>(id, commandExecutor, name), commandExecutor, name);
|
||||
}
|
||||
|
||||
public RedissonSetMultimapReactive(UUID id, Codec codec, CommandReactiveExecutor commandExecutor, String name) {
|
||||
super(new RedissonSetMultimap<K, V>(id, codec, commandExecutor, name), codec, commandExecutor, name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RSetReactive<V> get(final K key) {
|
||||
final ByteBuf keyState = encodeMapKey(key);
|
||||
final String keyHash = hashAndRelease(keyState);
|
||||
final String setName = getValuesName(keyHash);
|
||||
|
||||
return new RedissonSetReactive<V>(codec, commandExecutor, setName) {
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> delete() {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
return RedissonSetMultimapReactive.this.fastRemove(Arrays.<Object>asList(keyState), Arrays.<Object>asList(setName), RedisCommands.EVAL_BOOLEAN_AMOUNT);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> clearExpire() {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> expire(long timeToLive, TimeUnit timeUnit) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> expireAt(long timestamp) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Long> remainTimeToLive() {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Void> rename(String newName) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Boolean> renamenx(String newName) {
|
||||
throw new UnsupportedOperationException("This operation is not supported for SetMultimap values Set");
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Set<V>> getAll(K key) {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
String keyHash = hashAndRelease(keyState);
|
||||
String setName = getValuesName(keyHash);
|
||||
|
||||
return commandExecutor.readReactive(getName(), codec, RedisCommands.SMEMBERS, setName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Set<V>> removeAll(Object key) {
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
String keyHash = hash(keyState);
|
||||
|
||||
String setName = getValuesName(keyHash);
|
||||
return commandExecutor.evalWriteReactive(getName(), codec, RedisCommands.EVAL_SET,
|
||||
"redis.call('hdel', KEYS[1], ARGV[1]); " +
|
||||
"local members = redis.call('smembers', KEYS[2]); " +
|
||||
"redis.call('del', KEYS[2]); " +
|
||||
"return members; ",
|
||||
Arrays.<Object>asList(getName(), setName), keyState);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Publisher<Set<V>> replaceValues(K key, Iterable<? extends V> values) {
|
||||
List<Object> params = new ArrayList<Object>();
|
||||
ByteBuf keyState = encodeMapKey(key);
|
||||
params.add(keyState);
|
||||
String keyHash = hash(keyState);
|
||||
params.add(keyHash);
|
||||
for (Object value : values) {
|
||||
ByteBuf valueState = encodeMapValue(value);
|
||||
params.add(valueState);
|
||||
}
|
||||
|
||||
String setName = getValuesName(keyHash);
|
||||
return commandExecutor.evalWriteReactive(getName(), codec, RedisCommands.EVAL_SET,
|
||||
"redis.call('hset', KEYS[1], ARGV[1], ARGV[2]); " +
|
||||
"local members = redis.call('smembers', KEYS[2]); " +
|
||||
"redis.call('del', KEYS[2]); " +
|
||||
"redis.call('sadd', KEYS[2], unpack(ARGV, 3, #ARGV)); " +
|
||||
"return members; ",
|
||||
Arrays.<Object>asList(getName(), setName), params.toArray());
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue