Skip to content

Commit b0e33a4

Browse files
christophstroblchristophstrobl
committed
DATAREDIS-471 - Add support for partial updates.
We now allow partial update of domain types via PartialUpdate. The according expiration times and secondary index structures are updated accordingly. Still need to do some documentation and clean up.
1 parent 881fdc1 commit b0e33a4

17 files changed

+1662
-61
lines changed

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

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import org.springframework.data.redis.connection.RedisConnection;
2222
import org.springframework.data.redis.core.convert.IndexedData;
2323
import org.springframework.data.redis.core.convert.RedisConverter;
24+
import org.springframework.data.redis.core.convert.RemoveIndexedData;
2425
import org.springframework.data.redis.core.convert.SimpleIndexedPropertyValue;
2526
import org.springframework.data.redis.util.ByteUtils;
2627
import org.springframework.util.Assert;
@@ -123,8 +124,8 @@ private void removeKeyFromExistingIndexes(byte[] key, Iterable<IndexedData> inde
123124
protected void removeKeyFromExistingIndexes(byte[] key, IndexedData indexedData) {
124125

125126
Assert.notNull(indexedData, "IndexedData must not be null!");
126-
Set<byte[]> existingKeys = connection.keys(toBytes(indexedData.getKeyspace() + ":" + indexedData.getIndexName()
127-
+ ":*"));
127+
Set<byte[]> existingKeys = connection
128+
.keys(toBytes(indexedData.getKeyspace() + ":" + indexedData.getIndexName() + ":*"));
128129

129130
if (!CollectionUtils.isEmpty(existingKeys)) {
130131
for (byte[] existingKey : existingKeys) {
@@ -151,7 +152,11 @@ protected void addKeyToIndex(byte[] key, IndexedData indexedData) {
151152
Assert.notNull(key, "Key must not be null!");
152153
Assert.notNull(indexedData, "IndexedData must not be null!");
153154

154-
if (indexedData instanceof SimpleIndexedPropertyValue) {
155+
if (indexedData instanceof RemoveIndexedData) {
156+
return;
157+
}
158+
159+
else if (indexedData instanceof SimpleIndexedPropertyValue) {
155160

156161
Object value = ((SimpleIndexedPropertyValue) indexedData).getValue();
157162

@@ -166,8 +171,8 @@ protected void addKeyToIndex(byte[] key, IndexedData indexedData) {
166171
// keep track of indexes used for the object
167172
connection.sAdd(ByteUtils.concatAll(toBytes(indexedData.getKeyspace() + ":"), key, toBytes(":idx")), indexKey);
168173
} else {
169-
throw new IllegalArgumentException(String.format("Cannot write index data for unknown index type %s",
170-
indexedData.getClass()));
174+
throw new IllegalArgumentException(
175+
String.format("Cannot write index data for unknown index type %s", indexedData.getClass()));
171176
}
172177
}
173178

@@ -185,10 +190,8 @@ private byte[] toBytes(Object source) {
185190
return converter.getConversionService().convert(source, byte[].class);
186191
}
187192

188-
throw new InvalidDataAccessApiUsageException(
189-
String
190-
.format(
191-
"Cannot convert %s to binary representation for index key generation. Are you missing a Converter? Did you register a non PathBasedRedisIndexDefinition that might apply to a complex type?",
192-
source.getClass()));
193+
throw new InvalidDataAccessApiUsageException(String.format(
194+
"Cannot convert %s to binary representation for index key generation. Are you missing a Converter? Did you register a non PathBasedRedisIndexDefinition that might apply to a complex type?",
195+
source.getClass()));
193196
}
194197
}

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

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
/*
2+
* Copyright 2016 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.util.ArrayList;
19+
import java.util.Collections;
20+
import java.util.List;
21+
22+
import org.springframework.util.Assert;
23+
import org.springframework.util.ClassUtils;
24+
25+
/**
26+
* @author Christoph Strobl
27+
* @param <T>
28+
*/
29+
public class PartialUpdate<T> {
30+
31+
private final Object id;
32+
private final Class<T> target;
33+
private final T value;
34+
private boolean refreshTtl = false;
35+
36+
private final List<PropertyUpdate> propertyUpdates = new ArrayList<PropertyUpdate>();
37+
38+
private PartialUpdate(Object id, Class<T> target, T value, boolean refreshTtl, List<PropertyUpdate> propertyUpdates) {
39+
40+
this.id = id;
41+
this.target = target;
42+
this.value = value;
43+
this.refreshTtl = refreshTtl;
44+
this.propertyUpdates.addAll(propertyUpdates);
45+
}
46+
47+
/**
48+
* @param id must not be {@literal null}.
49+
* @param target must not be {@literal null}.
50+
*/
51+
@SuppressWarnings("unchecked")
52+
public PartialUpdate(Object id, Class<T> target) {
53+
54+
Assert.notNull(id, "Id must not be null!");
55+
Assert.notNull(target, "Target must not be null!");
56+
57+
this.id = id;
58+
this.target = (Class<T>) ClassUtils.getUserClass(target);
59+
this.value = null;
60+
}
61+
62+
/**
63+
* @param id must not be {@literal null}.
64+
* @param value must not be {@literal null}.
65+
*/
66+
@SuppressWarnings("unchecked")
67+
public PartialUpdate(Object id, T value) {
68+
69+
Assert.notNull(id, "Id must not be null!");
70+
Assert.notNull(value, "Value must not be null!");
71+
72+
this.id = id;
73+
this.target = (Class<T>) ClassUtils.getUserClass(value.getClass());
74+
this.value = value;
75+
}
76+
77+
/**
78+
* @return can be {@literal null}.
79+
*/
80+
public T getValue() {
81+
return value;
82+
}
83+
84+
public PartialUpdate<T> set(String path, Object value) {
85+
86+
propertyUpdates.add(new PropertyUpdate(UpdateCommand.SET, path, value));
87+
return new PartialUpdate<T>(this.id, this.target, this.value, this.refreshTtl, this.propertyUpdates);
88+
}
89+
90+
public PartialUpdate<T> del(String path) {
91+
92+
propertyUpdates.add(new PropertyUpdate(UpdateCommand.DEL, path));
93+
return new PartialUpdate<T>(this.id, this.target, this.value, this.refreshTtl, this.propertyUpdates);
94+
}
95+
96+
/**
97+
* @return never {@literal null}.
98+
*/
99+
public Class<T> getTarget() {
100+
return target;
101+
}
102+
103+
/**
104+
* @return never {@literal null}.
105+
*/
106+
public Object getId() {
107+
return id;
108+
}
109+
110+
public List<PropertyUpdate> getPropertyUpdates() {
111+
return Collections.unmodifiableList(propertyUpdates);
112+
}
113+
114+
public boolean isRefreshTtl() {
115+
return refreshTtl;
116+
}
117+
118+
public PartialUpdate<T> refreshTtl(boolean refreshTtl) {
119+
120+
this.refreshTtl = refreshTtl;
121+
return new PartialUpdate<T>(this.id, this.target, this.value, this.refreshTtl, this.propertyUpdates);
122+
}
123+
124+
/**
125+
* @author Christoph Strobl
126+
*/
127+
public static class PropertyUpdate {
128+
129+
private final UpdateCommand cmd;
130+
private final String propertyPath;
131+
private final Object value;
132+
133+
private PropertyUpdate(UpdateCommand cmd, String propertyPath) {
134+
this(cmd, propertyPath, null);
135+
}
136+
137+
private PropertyUpdate(UpdateCommand cmd, String propertyPath, Object value) {
138+
139+
this.cmd = cmd;
140+
this.propertyPath = propertyPath;
141+
this.value = value;
142+
}
143+
144+
public UpdateCommand getCmd() {
145+
return cmd;
146+
}
147+
148+
public String getPropertyPath() {
149+
return propertyPath;
150+
}
151+
152+
public Object getValue() {
153+
return value;
154+
}
155+
}
156+
157+
/**
158+
* @author Christoph Strobl
159+
*/
160+
public static enum UpdateCommand {
161+
SET, DEL
162+
}
163+
164+
}

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

Lines changed: 108 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import java.io.Serializable;
1919
import java.util.ArrayList;
20+
import java.util.Collection;
2021
import java.util.List;
2122
import java.util.Map;
2223
import java.util.Map.Entry;
@@ -25,7 +26,6 @@
2526
import org.slf4j.Logger;
2627
import org.slf4j.LoggerFactory;
2728
import org.springframework.beans.BeansException;
28-
import org.springframework.beans.factory.DisposableBean;
2929
import org.springframework.context.ApplicationContext;
3030
import org.springframework.context.ApplicationContextAware;
3131
import org.springframework.context.ApplicationListener;
@@ -38,7 +38,8 @@
3838
import org.springframework.data.redis.connection.Message;
3939
import org.springframework.data.redis.connection.MessageListener;
4040
import org.springframework.data.redis.connection.RedisConnection;
41-
import org.springframework.data.redis.connection.RedisConnectionFactory;
41+
import org.springframework.data.redis.core.PartialUpdate.PropertyUpdate;
42+
import org.springframework.data.redis.core.PartialUpdate.UpdateCommand;
4243
import org.springframework.data.redis.core.convert.CustomConversions;
4344
import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
4445
import org.springframework.data.redis.core.convert.MappingRedisConverter;
@@ -47,6 +48,7 @@
4748
import org.springframework.data.redis.core.convert.RedisData;
4849
import org.springframework.data.redis.core.convert.ReferenceResolverImpl;
4950
import org.springframework.data.redis.core.mapping.RedisMappingContext;
51+
import org.springframework.data.redis.core.mapping.RedisPersistentEntity;
5052
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
5153
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
5254
import org.springframework.data.redis.util.ByteUtils;
@@ -163,8 +165,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisConverter redis
163165
/**
164166
* Default constructor.
165167
*/
166-
protected RedisKeyValueAdapter() {
167-
}
168+
protected RedisKeyValueAdapter() {}
168169

169170
/*
170171
* (non-Javadoc)
@@ -392,6 +393,109 @@ public Long doInRedis(RedisConnection connection) throws DataAccessException {
392393
return count != null ? count.longValue() : 0;
393394
}
394395

396+
public void update(final PartialUpdate<?> update) {
397+
398+
final RedisPersistentEntity<?> entity = this.converter.getMappingContext().getPersistentEntity(update.getTarget());
399+
400+
final String keyspace = entity.getKeySpace();
401+
final Object id = update.getId();
402+
403+
final byte[] redisKey = createKey(keyspace, converter.getConversionService().convert(id, String.class));
404+
405+
final RedisData rdo = new RedisData();
406+
this.converter.write(update, rdo);
407+
408+
redisOps.execute(new RedisCallback<Void>() {
409+
410+
@Override
411+
public Void doInRedis(RedisConnection connection) throws DataAccessException {
412+
413+
List<byte[]> pathsToRemove = new ArrayList<byte[]>(update.getPropertyUpdates().size());
414+
415+
for (PropertyUpdate pUpdate : update.getPropertyUpdates()) {
416+
417+
String propertyPath = pUpdate.getPropertyPath();
418+
419+
if (UpdateCommand.DEL.equals(pUpdate.getCmd())) {
420+
421+
byte[] existingValue = connection.hGet(redisKey, toBytes(propertyPath));
422+
pathsToRemove.add(toBytes(propertyPath));
423+
424+
byte[] existingValueIndexKey = existingValue != null
425+
? ByteUtils.concatAll(toBytes(keyspace), (":" + propertyPath).getBytes(), ":".getBytes(), existingValue)
426+
: null;
427+
428+
if (existingValue != null) {
429+
430+
if (connection.exists(existingValueIndexKey)) {
431+
connection.sRem(existingValueIndexKey, toBytes(id));
432+
}
433+
}
434+
}
435+
436+
if (pUpdate.getValue() instanceof Collection || pUpdate.getValue() instanceof Map
437+
|| (pUpdate.getValue() != null && pUpdate.getValue().getClass().isArray()) || (pUpdate.getValue() != null
438+
&& !converter.getConversionService().canConvert(pUpdate.getValue().getClass(), byte[].class))) {
439+
440+
Set<byte[]> existingFields = connection.hKeys(redisKey);
441+
442+
for (byte[] hkey : existingFields) {
443+
444+
if (asString(hkey).startsWith(pUpdate.getPropertyPath() + ".")) {
445+
pathsToRemove.add(hkey);
446+
447+
byte[] existingValue = connection.hGet(redisKey, toBytes(hkey));
448+
byte[] existingValueIndexKey = existingValue != null ? ByteUtils.concatAll(toBytes(keyspace),
449+
(":" + propertyPath).getBytes(), ":".getBytes(), existingValue) : null;
450+
451+
if (existingValue != null) {
452+
453+
if (connection.exists(existingValueIndexKey)) {
454+
connection.sRem(existingValueIndexKey, toBytes(id));
455+
}
456+
}
457+
}
458+
}
459+
460+
}
461+
}
462+
463+
if (!pathsToRemove.isEmpty()) {
464+
connection.hDel(redisKey, pathsToRemove.toArray(new byte[pathsToRemove.size()][]));
465+
}
466+
467+
if (!rdo.getBucket().isEmpty()) {
468+
if (rdo.getBucket().size() > 1
469+
|| (rdo.getBucket().size() == 1 && !rdo.getBucket().asMap().containsKey("_class"))) {
470+
connection.hMSet(redisKey, rdo.getBucket().rawMap());
471+
}
472+
}
473+
474+
if (update.isRefreshTtl()) {
475+
476+
if (rdo.getTimeToLive() != null && rdo.getTimeToLive().longValue() > 0) {
477+
478+
connection.expire(redisKey, rdo.getTimeToLive().longValue());
479+
480+
// add phantom key so values can be restored
481+
byte[] phantomKey = ByteUtils.concat(redisKey, toBytes(":phantom"));
482+
connection.hMSet(phantomKey, rdo.getBucket().rawMap());
483+
connection.expire(phantomKey, rdo.getTimeToLive().longValue() + 300);
484+
485+
} else {
486+
487+
connection.persist(redisKey);
488+
connection.persist(ByteUtils.concat(redisKey, toBytes(":phantom")));
489+
}
490+
}
491+
492+
new IndexWriter(connection, converter).updateIndexes(toBytes(id), rdo.getIndexedData());
493+
return null;
494+
}
495+
496+
});
497+
}
498+
395499
/**
396500
* Execute {@link RedisCallback} via underlying {@link RedisOperations}.
397501
*
@@ -447,13 +551,6 @@ public byte[] toBytes(Object source) {
447551
*/
448552
public void destroy() throws Exception {
449553

450-
if (redisOps instanceof RedisTemplate) {
451-
RedisConnectionFactory connectionFactory = ((RedisTemplate<?, ?>) redisOps).getConnectionFactory();
452-
if (connectionFactory instanceof DisposableBean) {
453-
((DisposableBean) connectionFactory).destroy();
454-
}
455-
}
456-
457554
this.expirationListener.destroy();
458555
this.messageListenerContainer.destroy();
459556
}

0 commit comments

Comments
 (0)