more efficient way of cleanup of SerializedMappedSet

Author: drm

src/nl/melp/redis/collections/ScanIterator.java

@@ -4,22 +4,30 @@
 
 import java.io.IOException;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 
 public class ScanIterator implements Iterator<byte[]> {
 	private final Redis redis;
 	private final byte[] operation;
 	private final byte[] keyName;
+	private final byte[] match;
 	private int cursor;
 	protected int localCursor;
 	protected List<byte[]> buffer;
 
 	public ScanIterator(Redis redis, byte[]operation, byte[] keyName) {
+		this(redis, operation, null, keyName);
+	}
+
+
+	public ScanIterator(Redis redis, byte[]operation, byte[] match, byte[] keyName) {
 		this.redis = redis;
 		this.operation = operation;
 		this.keyName = keyName;
 		this.cursor = 0;
 		this.localCursor = 0;
+		this.match = match;
 	}
 
 
@@ -32,8 +40,18 @@ public boolean hasNext() {
 			}
 			try {
 				List<Object> result;
+				List<Object> args = new LinkedList<>();
+				args.add(this.operation);
+				if (keyName != null) {
+					args.add(this.keyName);
+				}
+				args.add(Integer.toString(this.cursor));
+				if (match != null) {
+					args.add("MATCH");
+					args.add(match);
+				}
 				synchronized (redis) {
-					result = redis.call(this.operation, this.keyName, Integer.toString(this.cursor));
+					result = redis.call(args.toArray());
 				}
 				this.cursor = Integer.valueOf(new String((byte[]) result.get(0)));
 				this.buffer = (List<byte[]>) result.get(1);

src/nl/melp/redis/collections/SerializedMappedSet.java

@@ -19,6 +19,27 @@ public SerializedMemberSet(K key, ISerializer<V> serializer, Redis redis, String
 			}
 		}
 
+		@Override
+		public boolean addAll(Collection<? extends V> collection) {
+			keys.add(key);
+			return super.addAll(collection);
+		}
+
+		@Override
+		public boolean removeAll(Collection<?> collection) {
+			boolean ret = super.removeAll(collection);
+			if (ret && size() == 0) {
+				keys.remove(key);
+			}
+			return ret;
+		}
+
+		@Override
+		public boolean add(V o) {
+			keys.add(key);
+			return super.add(o);
+		}
+
 		@Override
 		public void clear() {
 			super.clear();
@@ -68,17 +89,6 @@ private byte[] withPrefix(byte[] value) {
 		return prefixedValue;
 	}
 
-	private <T> T call(String command, byte[]... args) {
-		byte[][] rawArgs = new byte[args.length + 1][];
-		rawArgs[0] = command.getBytes();
-		System.arraycopy(args, 0, rawArgs, 1, args.length);
-		try {
-			return redis.call((Object[])rawArgs);
-		} catch (IOException e) {
-			throw new UnsupportedOperationException(e);
-		}
-	}
-
 	@Override
 	public int size() {
 		return keys.size();
@@ -140,10 +150,19 @@ public void putAll(Map<? extends K, ? extends Set<V>> map) {
 	@Override
 	public void clear() {
 		synchronized (redis) {
-			for (K key : keys) {
-				this.get(key).clear();
+			byte[] copy = new byte[prefixLength + 2];
+			System.arraycopy(prefix, 0, copy, 0, prefixLength);
+			copy[prefixLength] = ':';
+			copy[prefixLength + 1] = '*';
+			Iterator<byte[]> i = new ScanIterator(redis, "SCAN".getBytes(), copy, null);
+
+			while (i.hasNext()) {
+				try {
+					redis.call("DEL", i.next());
+				} catch (IOException e) {
+					throw new RuntimeException(e);
+				}
 			}
-			keys.clear();
 		}
 	}
 

src/nl/melp/redis/collections/SerializedSet.java

@@ -5,9 +5,11 @@
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 public class SerializedSet<V> implements Set<V>, Collection<V> {
@@ -40,7 +42,7 @@ public boolean add(V o) {
 
 	@Override
 	public boolean addAll(java.util.Collection<? extends V> collection) {
-		return innerSet.addAll(collection.stream().map(serializer::serialize).collect(Collectors.toList()));
+		return innerSet.addAll(collection.stream().map(serializer::serialize).collect(Collectors.toSet()));
 	}
 
 	@Override
@@ -81,7 +83,12 @@ public V next() {
 
 	@Override
 	public boolean containsAll(Collection<?> collection) {
-		throw new UnsupportedOperationException("Not implemented");
+		for (Object item : collection) {
+			if (!contains(item)) {
+				return false;
+			}
+		}
+		return true;
 	}
 
 	@Override
@@ -91,7 +98,13 @@ public boolean retainAll(Collection<?> collection) {
 
 	@Override
 	public boolean removeAll(Collection<?> collection) {
-		throw new UnsupportedOperationException("Not implemented");
+		AtomicBoolean b = new AtomicBoolean(false);
+		collection.forEach(item -> {
+			if (this.remove(item)) {
+				b.set(true);
+			}
+		});
+		return b.get();
 	}
 
 	@Override

test/nl/melp/redis/collections/IntegrationTest.java

@@ -359,9 +359,30 @@ public void testSerializedMappedSetForInForEach() throws IOException {
 	@Test
 	public void testSerializedMappedSetKeySynchronization() throws IOException {
 		Map<String, Set<String>> map = new SerializedMappedSet<>(Serializers.of(String.class), Serializers.of(String.class), redis, keyName);
+		Map<String, Set<String>> secondary = new SerializedMappedSet<>(Serializers.of(String.class), Serializers.of(String.class), redis, keyName);
+
 		Assert.assertEquals(0, map.get("A").size());
+		Assert.assertEquals(0, secondary.get("A").size());
+		Assert.assertEquals(0, map.size());
+		Assert.assertEquals(0, secondary.size());
+		map.get("A").add("foo");
+		Assert.assertEquals(1, map.size());
+		Assert.assertEquals(1, secondary.size());
+		map.get("A").remove("foo");
+		Assert.assertEquals(0, map.size());
+		Assert.assertEquals(0, secondary.size());
+		map.get("A").addAll(new HashSet<String>(){{ add ("foo"); add("bar"); }});
+		Assert.assertEquals(1, map.size());
+		Assert.assertEquals(1, secondary.size());
+		map.get("A").removeAll(new HashSet<String>(){{ add ("foo"); add("bar"); }});
 		Assert.assertEquals(0, map.size());
-		map.remove("A");
+		Assert.assertEquals(0, secondary.size());
+
+		map.get("A").addAll(new HashSet<String>(){{ add ("foo"); add("bar"); }});
+		map.clear();
+
 		Assert.assertEquals(0, map.size());
+		Assert.assertEquals(0, secondary.size());
+
 	}
 }