Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-11 moved from CqlUtils insert and update to the

Browse Source 
CassandraConverter
This commit is contained in:
Alex Shvid
2013-11-16 19:32:44 -08:00
parent 5b8c301baf
commit 963be476b6
5 changed files with 135 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java
View File

@@ -157,14 +157,16 @@ public abstract class AbstractCassandraConfiguration implements BeanClassLoaderA
}
/**
* Return the {@link CassandraConverter} instance to convert Rows to Objects.
* Return the {@link CassandraConverter} instance to convert Rows to Objects, Objects to BuiltStatements
*
* @return
* @throws Exception
*/
@Bean
public CassandraConverter converter() {
return new MappingCassandraConverter(mappingContext());
MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext());
converter.setBeanClassLoader(beanClassLoader);
return converter;
}
/**

4
src/main/java/org/springframework/data/cassandra/convert/CassandraConverter.java
View File

@@ -19,14 +19,12 @@ import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
import org.springframework.data.convert.EntityConverter;
import com.datastax.driver.core.Row;
/**
* Central Cassandra specific converter interface from Object to Row.
*
* @author Alex Shvid
*/
public interface CassandraConverter extends
EntityConverter<CassandraPersistentEntity<?>, CassandraPersistentProperty, Object, Row> {
EntityConverter<CassandraPersistentEntity<?>, CassandraPersistentProperty, Object, Object> {
}

117
src/main/java/org/springframework/data/cassandra/convert/MappingCassandraConverter.java
View File

@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.convert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.support.DefaultConversionService;
@@ -34,8 +35,12 @@ import org.springframework.data.mapping.model.PropertyValueProvider;
import org.springframework.data.mapping.model.SpELContext;
import org.springframework.data.util.ClassTypeInformation;
import org.springframework.data.util.TypeInformation;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Update;
/**
* {@link CassandraConverter} that uses a {@link MappingContext} to do sophisticated mapping of domain objects to
@@ -43,7 +48,8 @@ import com.datastax.driver.core.Row;
*
* @author Alex Shvid
*/
public class MappingCassandraConverter extends AbstractCassandraConverter implements ApplicationContextAware {
public class MappingCassandraConverter extends AbstractCassandraConverter implements ApplicationContextAware,
BeanClassLoaderAware {
protected static final Logger log = LoggerFactory.getLogger(MappingCassandraConverter.class);
@@ -52,6 +58,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter implem
private SpELContext spELContext;
private boolean useFieldAccessOnly = true;
private ClassLoader beanClassLoader;
/**
* Creates a new {@link MappingCassandraConverter} given the new {@link MappingContext}.
*
@@ -65,9 +73,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter implem
}
@SuppressWarnings("unchecked")
public <R> R read(Class<R> clazz, Row row) {
public <R> R readRow(Class<R> clazz, Row row) {
TypeInformation<? extends R> type = ClassTypeInformation.from(clazz);
Class<R> beanClassLoaderClass = transformClassToBeanClassLoaderClass(clazz);
TypeInformation<? extends R> type = ClassTypeInformation.from(beanClassLoaderClass);
// TypeInformation<? extends R> typeToUse = typeMapper.readType(row, type);
TypeInformation<? extends R> typeToUse = type;
Class<? extends R> rawType = typeToUse.getType();
@@ -82,7 +92,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter implem
throw new MappingException("No mapping metadata found for " + rawType.getName());
}
return read(persistentEntity, row);
return readRowInternal(persistentEntity, row);
}
/*
@@ -102,7 +112,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter implem
this.spELContext = new SpELContext(this.spELContext, applicationContext);
}
private <S extends Object> S read(final CassandraPersistentEntity<S> entity, final Row row) {
private <S extends Object> S readRowInternal(final CassandraPersistentEntity<S> entity, final Row row) {
final DefaultSpELExpressionEvaluator evaluator = new DefaultSpELExpressionEvaluator(row, spELContext);
@@ -144,14 +154,97 @@ public class MappingCassandraConverter extends AbstractCassandraConverter implem
* @see org.springframework.data.convert.EntityWriter#write(java.lang.Object, java.lang.Object)
*/
@Override
public void write(Object source, Row sink) {
public <R> R read(Class<R> type, Object row) {
if (row instanceof Row) {
return readRow(type, (Row) row);
}
throw new MappingException("Unknown row object " + row.getClass().getName());
}
/*
* There is no concept of passing a Row into Cassandra for Writing.
* This must be done with Query
*
* See the CQLUtils.
*/
/* (non-Javadoc)
* @see org.springframework.data.convert.EntityWriter#write(java.lang.Object, java.lang.Object)
*/
@Override
public void write(Object obj, Object builtStatement) {
if (obj == null) {
return;
}
Class<?> beanClassLoaderClass = transformClassToBeanClassLoaderClass(obj.getClass());
CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(beanClassLoaderClass);
if (entity == null) {
throw new MappingException("No mapping metadata found for " + obj.getClass());
}
if (builtStatement instanceof Insert) {
writeInsertInternal(obj, (Insert) builtStatement, entity);
} else if (builtStatement instanceof Update) {
writeUpdateInternal(obj, (Update) builtStatement, entity);
} else {
throw new MappingException("Unknown buildStatement " + builtStatement.getClass().getName());
}
}
private void writeInsertInternal(final Object objectToSave, final Insert insert, CassandraPersistentEntity<?> entity) {
final BeanWrapper<CassandraPersistentEntity<Object>, Object> wrapper = BeanWrapper.create(objectToSave,
conversionService);
// Write the properties
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
Object propertyObj = wrapper.getProperty(prop, prop.getType(), useFieldAccessOnly);
if (propertyObj != null) {
insert.value(prop.getColumnName(), propertyObj);
}
}
});
}
private void writeUpdateInternal(final Object objectToSave, final Update update, CassandraPersistentEntity<?> entity) {
final BeanWrapper<CassandraPersistentEntity<Object>, Object> wrapper = BeanWrapper.create(objectToSave,
conversionService);
// Write the properties
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
Object propertyObj = wrapper.getProperty(prop, prop.getType(), useFieldAccessOnly);
if (propertyObj != null) {
if (prop.isIdProperty()) {
update.where(QueryBuilder.eq(prop.getColumnName(), propertyObj));
} else {
update.with(QueryBuilder.set(prop.getColumnName(), propertyObj));
}
}
}
});
}
@SuppressWarnings("unchecked")
private <T> Class<T> transformClassToBeanClassLoaderClass(Class<T> entity) {
try {
return (Class<T>) ClassUtils.forName(entity.getName(), beanClassLoader);
} catch (ClassNotFoundException e) {
return entity;
} catch (LinkageError e) {
return entity;
}
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
}

32
src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -69,10 +69,10 @@ public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAw
*/
private static class ReadRowCallback<T> implements RowCallback<T> {
private final EntityReader<? super T, Row> reader;
private final EntityReader<? super T, Object> reader;
private final Class<T> type;
public ReadRowCallback(EntityReader<? super T, Row> reader, Class<T> type) {
public ReadRowCallback(EntityReader<? super T, Object> reader, Class<T> type) {
Assert.notNull(reader);
Assert.notNull(type);
this.reader = reader;
@@ -1030,13 +1030,10 @@ public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAw
Assert.notEmpty(entities);
CassandraPersistentEntity<?> CPEntity = getEntity(entities.get(0));
Assert.notNull(CPEntity);
try {
final Batch b = CqlUtils.toInsertBatchQuery(keyspace.getKeyspace(), tableName, entities, CPEntity, optionsByName);
final Batch b = CqlUtils.toInsertBatchQuery(keyspace.getKeyspace(), tableName, entities, optionsByName,
cassandraConverter);
log.info(b.getQueryString());
return execute(new SessionCallback<List<T>>() {
@@ -1075,13 +1072,10 @@ public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAw
Assert.notEmpty(entities);
CassandraPersistentEntity<?> CPEntity = getEntity(entities.get(0));
Assert.notNull(CPEntity);
try {
final Batch b = CqlUtils.toUpdateBatchQuery(keyspace.getKeyspace(), tableName, entities, CPEntity, optionsByName);
final Batch b = CqlUtils.toUpdateBatchQuery(keyspace.getKeyspace(), tableName, entities, optionsByName,
cassandraConverter);
log.info(b.toString());
return execute(new SessionCallback<List<T>>() {
@@ -1155,13 +1149,10 @@ public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAw
protected <T> T doInsert(final String tableName, final T entity, final Map<String, Object> optionsByName,
final boolean insertAsychronously) {
CassandraPersistentEntity<?> CPEntity = getEntity(entity);
Assert.notNull(CPEntity);
try {
final Query q = CqlUtils.toInsertQuery(keyspace.getKeyspace(), tableName, entity, CPEntity, optionsByName);
final Query q = CqlUtils.toInsertQuery(keyspace.getKeyspace(), tableName, entity, optionsByName,
cassandraConverter);
log.info(q.toString());
if (q.getConsistencyLevel() != null) {
log.info(q.getConsistencyLevel().name());
@@ -1205,13 +1196,10 @@ public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAw
protected <T> T doUpdate(final String tableName, final T entity, final Map<String, Object> optionsByName,
final boolean updateAsychronously) {
CassandraPersistentEntity<?> CPEntity = getEntity(entity);
Assert.notNull(CPEntity);
try {
final Query q = CqlUtils.toUpdateQuery(keyspace.getKeyspace(), tableName, entity, CPEntity, optionsByName);
final Query q = CqlUtils.toUpdateQuery(keyspace.getKeyspace(), tableName, entity, optionsByName,
cassandraConverter);
log.info(q.toString());
return execute(new SessionCallback<T>() {

85
src/main/java/org/springframework/data/cassandra/util/CqlUtils.java
View File

@@ -16,6 +16,7 @@ import org.springframework.data.cassandra.core.RetryPolicyResolver;
import org.springframework.data.cassandra.exception.EntityWriterException;
import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
import org.springframework.data.convert.EntityWriter;
import org.springframework.data.mapping.PropertyHandler;
import com.datastax.driver.core.ColumnMetadata;
@@ -214,40 +215,14 @@ public abstract class CqlUtils {
* @throws EntityWriterException
*/
public static Query toInsertQuery(String keyspaceName, String tableName, final Object objectToSave,
CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
Map<String, Object> optionsByName, EntityWriter<Object, Object> entityWriter) throws EntityWriterException {
final Insert q = QueryBuilder.insertInto(keyspaceName, tableName);
final Exception innerException = new Exception();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
/*
* See if the object has a value for that column, and if so, add it to the Query
*/
try {
Object o = prop.getGetter().invoke(objectToSave, new Object[0]);
log.info("Getter Invoke [" + prop.getColumnName() + " => " + o);
if (o != null) {
q.value(prop.getColumnName(), o);
}
} catch (IllegalAccessException e) {
innerException.initCause(e);
} catch (IllegalArgumentException e) {
innerException.initCause(e);
} catch (InvocationTargetException e) {
innerException.initCause(e);
}
}
});
if (innerException.getCause() != null) {
throw new EntityWriterException("Failed to convert Persistent Entity to CQL/Query", innerException.getCause());
}
/*
* Write properties
*/
entityWriter.write(objectToSave, q);
/*
* Add Query Options
@@ -280,44 +255,14 @@ public abstract class CqlUtils {
* @throws EntityWriterException
*/
public static Query toUpdateQuery(String keyspaceName, String tableName, final Object objectToSave,
CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
Map<String, Object> optionsByName, EntityWriter<Object, Object> entityWriter) throws EntityWriterException {
final Update q = QueryBuilder.update(keyspaceName, tableName);
final Exception innerException = new Exception();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
public void doWithPersistentProperty(CassandraPersistentProperty prop) {
/*
* See if the object has a value for that column, and if so, add it to the Query
*/
try {
Object o = prop.getGetter().invoke(objectToSave, new Object[0]);
log.info("Getter Invoke [" + prop.getColumnName() + " => " + o);
if (o != null) {
if (prop.isIdProperty()) {
q.where(QueryBuilder.eq(prop.getColumnName(), o));
} else {
q.with(QueryBuilder.set(prop.getColumnName(), o));
}
}
} catch (IllegalAccessException e) {
innerException.initCause(e);
} catch (IllegalArgumentException e) {
innerException.initCause(e);
} catch (InvocationTargetException e) {
innerException.initCause(e);
}
}
});
if (innerException.getCause() != null) {
throw new EntityWriterException("Failed to convert Persistent Entity to CQL/Query", innerException.getCause());
}
/*
* Write properties
*/
entityWriter.write(objectToSave, q);
/*
* Add Query Options
@@ -349,7 +294,7 @@ public abstract class CqlUtils {
* @throws EntityWriterException
*/
public static <T> Batch toUpdateBatchQuery(final String keyspaceName, final String tableName,
final List<T> objectsToSave, CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName)
final List<T> objectsToSave, Map<String, Object> optionsByName, EntityWriter<Object, Object> entityWriter)
throws EntityWriterException {
/*
@@ -359,7 +304,7 @@ public abstract class CqlUtils {
for (final T objectToSave : objectsToSave) {
b.add((Statement) toUpdateQuery(keyspaceName, tableName, objectToSave, entity, optionsByName));
b.add((Statement) toUpdateQuery(keyspaceName, tableName, objectToSave, optionsByName, entityWriter));
}
@@ -383,7 +328,7 @@ public abstract class CqlUtils {
* @throws EntityWriterException
*/
public static <T> Batch toInsertBatchQuery(final String keyspaceName, final String tableName,
final List<T> objectsToSave, CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName)
final List<T> objectsToSave, Map<String, Object> optionsByName, EntityWriter<Object, Object> entityWriter)
throws EntityWriterException {
/*
@@ -393,7 +338,7 @@ public abstract class CqlUtils {
for (final T objectToSave : objectsToSave) {
b.add((Statement) toInsertQuery(keyspaceName, tableName, objectToSave, entity, optionsByName));
b.add((Statement) toInsertQuery(keyspaceName, tableName, objectToSave, optionsByName, entityWriter));
}