Skip to content

Commit aaf0cc3

Browse files
christophstroblchristophstrobl
committed
DATAREDIS-425 - Refactoring part I
- Introduce Bucket to not operate directly on a Map of byte[]. - Introduce IndexData to not operate directly on Map of byte[] for indexes. - Move index updates to IndexWriter. - Additional tests. - Add hamcrest matcher for newly introduced Bucket.
1 parent 7cf2488 commit aaf0cc3

File tree

13 files changed

+1220
-359
lines changed

13 files changed

+1220
-359
lines changed

‎src/main/java/org/springframework/data/redis/core/IndexDataWriter.java

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
/*
2+
* Copyright 2015 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.data.redis.core;
17+
18+
import java.io.Serializable;
19+
import java.util.Set;
20+
21+
import org.springframework.data.redis.connection.RedisConnection;
22+
import org.springframework.data.redis.core.convert.IndexedData;
23+
import org.springframework.data.redis.core.convert.RedisConverter;
24+
import org.springframework.data.redis.core.convert.SimpleIndexedPropertyValue;
25+
import org.springframework.data.redis.util.ByteUtils;
26+
import org.springframework.util.Assert;
27+
28+
/**
29+
* @author Christoph Strobl
30+
*/
31+
class IndexDataWriter {
32+
33+
private final RedisConnection connection;
34+
private final RedisConverter converter;
35+
private final Serializable keyspace;
36+
37+
public IndexDataWriter(Serializable keyspace, RedisConnection connection, RedisConverter converter) {
38+
39+
Assert.notNull(keyspace, "Keyspace cannot be null!");
40+
Assert.notNull(connection, "RedisConnection cannot be null!");
41+
Assert.notNull(converter, "RedisConverter cannot be null!");
42+
43+
this.connection = connection;
44+
this.converter = converter;
45+
this.keyspace = keyspace;
46+
}
47+
48+
public void updateIndexes(Object key, Iterable<IndexedData> indexValues) {
49+
50+
Assert.notNull(key, "Key must not be null!");
51+
if (indexValues == null) {
52+
return;
53+
}
54+
55+
byte[] binKey = toBytes(key);
56+
57+
removeKeyFromExistingIndexes(binKey, indexValues);
58+
addKeyToIndexes(binKey, indexValues);
59+
}
60+
61+
public void removeKeyFromIndexes(Object key) {
62+
63+
Assert.notNull(key, "Key must not be null!");
64+
byte[] binKey = toBytes(key);
65+
66+
Set<byte[]> potentialIndex = connection.keys(toBytes(keyspace + ".*"));
67+
for (byte[] indexKey : potentialIndex) {
68+
connection.sRem(indexKey, binKey);
69+
}
70+
71+
}
72+
73+
public void removeAllIndexes() {
74+
75+
Set<byte[]> potentialIndex = connection.keys(toBytes(keyspace + ".*"));
76+
77+
if (!potentialIndex.isEmpty()) {
78+
connection.del(potentialIndex.toArray(new byte[potentialIndex.size()][]));
79+
}
80+
}
81+
82+
private void removeKeyFromExistingIndexes(byte[] key, Iterable<IndexedData> indexValues) {
83+
84+
for (IndexedData indexData : indexValues) {
85+
removeKeyFromExistingIndexes(key, indexData);
86+
}
87+
}
88+
89+
private void removeKeyFromExistingIndexes(byte[] key, IndexedData indexedData) {
90+
91+
Set<byte[]> existingKeys = connection.keys(toBytes(keyspace + "." + indexedData.getPath() + ":*"));
92+
for (byte[] existingKey : existingKeys) {
93+
connection.sRem(existingKey, key);
94+
}
95+
}
96+
97+
private void addKeyToIndexes(byte[] key, Iterable<IndexedData> indexValues) {
98+
99+
for (IndexedData indexData : indexValues) {
100+
addKeyToIndex(key, indexData);
101+
}
102+
}
103+
104+
private void addKeyToIndex(byte[] key, IndexedData indexedData) {
105+
106+
if (indexedData instanceof SimpleIndexedPropertyValue) {
107+
108+
Object value = ((SimpleIndexedPropertyValue) indexedData).getValue();
109+
110+
if (value == null) {
111+
return;
112+
}
113+
114+
byte[] indexKey = toBytes(keyspace + "." + indexedData.getPath() + ":");
115+
indexKey = ByteUtils.concat(indexKey, toBytes(value));
116+
connection.sAdd(indexKey, key);
117+
} else {
118+
throw new IllegalArgumentException("Cannot index data");
119+
}
120+
}
121+
122+
private byte[] toBytes(Object source) {
123+
124+
if (source == null) {
125+
return new byte[] {};
126+
}
127+
128+
if (source instanceof byte[]) {
129+
return (byte[]) source;
130+
}
131+
132+
return converter.getConversionService().convert(source, byte[].class);
133+
}
134+
135+
}

‎src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java

Lines changed: 30 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.springframework.dao.DataAccessException;
2727
import org.springframework.data.keyvalue.core.AbstractKeyValueAdapter;
2828
import org.springframework.data.keyvalue.core.KeyValueAdapter;
29+
import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentProperty;
2930
import org.springframework.data.redis.connection.RedisConnection;
3031
import org.springframework.data.redis.connection.RedisConnectionFactory;
3132
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
@@ -34,7 +35,6 @@
3435
import org.springframework.data.redis.core.convert.RedisData;
3536
import org.springframework.data.redis.core.convert.ReferenceResolverImpl;
3637
import org.springframework.data.redis.core.index.IndexConfiguration;
37-
import org.springframework.data.redis.util.ByteUtils;
3838
import org.springframework.data.util.CloseableIterator;
3939

4040
/**
@@ -48,11 +48,11 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter {
4848

4949
private MappingRedisConverter converter;
5050

51-
public RedisKeyValueAdapter() {
51+
RedisKeyValueAdapter() {
5252
this((IndexConfiguration) null);
5353
}
5454

55-
public RedisKeyValueAdapter(IndexConfiguration indexConfiguration) {
55+
RedisKeyValueAdapter(IndexConfiguration indexConfiguration) {
5656

5757
super(new RedisQueryEngine());
5858

@@ -86,35 +86,33 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, IndexConfiguration i
8686
*/
8787
public Object put(final Serializable id, final Object item, final Serializable keyspace) {
8888

89-
final RedisData rdo = new RedisData();
90-
converter.write(item, rdo);
89+
final RedisData rdo = item instanceof RedisData ? (RedisData) item : new RedisData();
90+
if (!(item instanceof RedisData)) {
91+
converter.write(item, rdo);
92+
}
93+
94+
if (rdo.getId() == null) {
9195

92-
final byte[] indexPostFixPattern = converter.toBytes(":*");
96+
rdo.setId(id);
97+
KeyValuePersistentProperty idProperty = converter.getMappingContext().getPersistentEntity(item.getClass())
98+
.getIdProperty();
99+
converter.getMappingContext().getPersistentEntity(item.getClass()).getPropertyAccessor(item)
100+
.setProperty(idProperty, id);
101+
102+
}
93103

94104
redisOps.execute(new RedisCallback<Object>() {
95105

96106
@Override
97107
public Object doInRedis(RedisConnection connection) throws DataAccessException {
98108

99-
connection.hMSet(rdo.getKey(), rdo.getData());
100-
connection.sAdd(rdo.getKeyspace(), rdo.getId());
101-
102-
// remove id from potential indexes since those might be invalid with the new data
103-
for (byte[] potentialIndex : rdo.getIndexPaths()) {
104-
105-
Set<byte[]> existingKeys = connection.keys(ByteUtils.concat(potentialIndex, indexPostFixPattern));
106-
107-
for (byte[] existingKey : existingKeys) {
108-
connection.sRem(existingKey, rdo.getId());
109-
}
110-
}
109+
byte[] key = converter.toBytes(rdo.getId());
111110

112-
if (!rdo.getSimpleIndexKeys().isEmpty()) {
111+
connection.del(createKey(rdo.getKeyspace(), rdo.getId()));
112+
connection.hMSet(createKey(rdo.getKeyspace(), rdo.getId()), rdo.getBucket().rawMap());
113+
connection.sAdd(converter.toBytes(rdo.getKeyspace()), key);
113114

114-
for (byte[] index : rdo.getSimpleIndexKeys()) {
115-
connection.sAdd(index, rdo.getId());
116-
}
117-
}
115+
new IndexDataWriter(rdo.getKeyspace(), connection, converter).updateIndexes(key, rdo.getIndexedData());
118116
return null;
119117
}
120118
});
@@ -145,7 +143,7 @@ public Boolean doInRedis(RedisConnection connection) throws DataAccessException
145143
*/
146144
public Object get(Serializable id, Serializable keyspace) {
147145

148-
final byte[] binId = converter.convertToId(keyspace, id);
146+
final byte[] binId = createKey(keyspace, id);
149147

150148
Map<byte[], byte[]> raw = redisOps.execute(new RedisCallback<Map<byte[], byte[]>>() {
151149

@@ -176,18 +174,10 @@ public Object delete(final Serializable id, final Serializable keyspace) {
176174
@Override
177175
public Void doInRedis(RedisConnection connection) throws DataAccessException {
178176

179-
connection.del(converter.convertToId(binKeyspace, binId));
177+
connection.del(createKey(keyspace, id));
180178
connection.sRem(binKeyspace, binId);
181179

182-
Set<byte[]> potentialIndex = connection.keys(converter.toBytes(keyspace + ".*"));
183-
184-
for (byte[] indexKey : potentialIndex) {
185-
try {
186-
connection.sRem(indexKey, binId);
187-
} catch (Exception e) {
188-
System.err.println(e);
189-
}
190-
}
180+
new IndexDataWriter(keyspace, connection, converter).removeKeyFromIndexes(binId);
191181
return null;
192182
}
193183
});
@@ -214,7 +204,7 @@ public List<Map<byte[], byte[]>> doInRedis(RedisConnection connection) throws Da
214204
Set<byte[]> members = connection.sMembers(binKeyspace);
215205

216206
for (byte[] id : members) {
217-
rawData.add(connection.hGetAll(converter.convertToId(binKeyspace, id)));
207+
rawData.add(connection.hGetAll(createKey(binKeyspace, id)));
218208
}
219209

220210
return rawData;
@@ -241,12 +231,7 @@ public void deleteAllOf(final Serializable keyspace) {
241231
public Void doInRedis(RedisConnection connection) throws DataAccessException {
242232

243233
connection.del(converter.toBytes(keyspace));
244-
245-
Set<byte[]> potentialIndex = connection.keys(converter.toBytes(keyspace + ".*"));
246-
247-
for (byte[] indexKey : potentialIndex) {
248-
connection.del(indexKey);
249-
}
234+
new IndexDataWriter(keyspace, connection, converter).removeAllIndexes();
250235
return null;
251236
}
252237
});
@@ -301,6 +286,10 @@ public void clear() {
301286
// nothing to do
302287
}
303288

289+
public byte[] createKey(Serializable keyspace, Serializable id) {
290+
return this.converter.toBytes(keyspace + ":" + id);
291+
}
292+
304293
/*
305294
* (non-Javadoc)
306295
* @see org.springframework.beans.factory.DisposableBean#destroy()

0 commit comments

Comments
 (0)