DATACOUCH-12: Refactor mapping layer to accomodate complex types.

This changeset is a complete refactor of the mapping layer to make
sure that _class complex type mapping works. It also decouples
the actual JSON mapping and abstracts it into a CouchbaseDocument
mapping. This ensures forward compatibility by potentially swapping
out JSON to a different format.

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java

@@ -26,9 +26,12 @@ import net.spy.memcached.internal.OperationFuture;
 
 import org.springframework.data.couchbase.core.convert.CouchbaseConverter;
 import org.springframework.data.couchbase.core.convert.MappingCouchbaseConverter;
-import org.springframework.data.couchbase.core.mapping.ConvertedCouchbaseDocument;
+import org.springframework.data.couchbase.core.convert.translation.JacksonTranslationService;
+import org.springframework.data.couchbase.core.convert.translation.TranslationService;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+import org.springframework.data.couchbase.core.mapping.CouchbaseStorable;
 import org.springframework.data.mapping.context.MappingContext;
 
 import com.couchbase.client.CouchbaseClient;
@@ -45,6 +48,7 @@ public class CouchbaseTemplate implements CouchbaseOperations {
   private static final Collection<String> ITERABLE_CLASSES;
   private final CouchbaseExceptionTranslator exceptionTranslator = 
   		new CouchbaseExceptionTranslator();
+  private final TranslationService<Object> translationService;
   
 	static {
 		Set<String> iterableClasses = new HashSet<String>();
@@ -61,8 +65,9 @@ public class CouchbaseTemplate implements CouchbaseOperations {
   public CouchbaseTemplate(final CouchbaseClient client,
     final CouchbaseConverter converter) {
     this.client = client;
-    this.couchbaseConverter = converter == null ? getDefaultConverter(client) : converter;
-    this.mappingContext = this.couchbaseConverter.getMappingContext();
+    couchbaseConverter = converter == null ? getDefaultConverter(client) : converter;
+    mappingContext = couchbaseConverter.getMappingContext();
+    translationService = new JacksonTranslationService();
   }
 
   private CouchbaseConverter getDefaultConverter(final CouchbaseClient client) {
@@ -72,24 +77,32 @@ public class CouchbaseTemplate implements CouchbaseOperations {
     return converter;
   }
 
+  private Object translateEncode(final CouchbaseStorable source) {
+    return translationService.encode(source);
+  }
+
+  private CouchbaseStorable translateDecode(final String source, final CouchbaseStorable target) {
+    return translationService.decode(source, target);
+  }
+
   public final void insert(final Object objectToSave) {
   	ensureNotIterable(objectToSave);
 
-    final ConvertedCouchbaseDocument converted =
-      new ConvertedCouchbaseDocument();
+    final CouchbaseDocument converted = new CouchbaseDocument();
     couchbaseConverter.write(objectToSave, converted);
+
     execute(new BucketCallback<OperationFuture<Boolean>>() {
       @Override
       public OperationFuture<Boolean> doInBucket() {
         return client.add(
-          converted.getId(), converted.getExpiry(), converted.getRawValue());
+          converted.getId(), converted.getExpiration(), translateEncode(converted));
       }
     });
   }
   
   public final void insert(final Collection<? extends Object> batchToSave) {
   	Iterator<? extends Object> iter = batchToSave.iterator();
-  	while(iter.hasNext()) {
+  	while (iter.hasNext()) {
   		insert(iter.next());
   	}
   }
@@ -97,15 +110,14 @@ public class CouchbaseTemplate implements CouchbaseOperations {
   public void save(final Object objectToSave) {
   	ensureNotIterable(objectToSave);
 
-    final ConvertedCouchbaseDocument converted =
-      new ConvertedCouchbaseDocument();
+    final CouchbaseDocument converted = new CouchbaseDocument();
     couchbaseConverter.write(objectToSave, converted);
 
     execute(new BucketCallback<OperationFuture<Boolean>>() {
       @Override
       public OperationFuture<Boolean> doInBucket() {
         return client.set(
-          converted.getId(), converted.getExpiry(), converted.getRawValue());
+          converted.getId(), converted.getExpiration(), translateEncode(converted));
       }
     });
   }
@@ -120,15 +132,14 @@ public class CouchbaseTemplate implements CouchbaseOperations {
   public void update(final Object objectToSave) {
   	ensureNotIterable(objectToSave);
 
-    final ConvertedCouchbaseDocument converted =
-      new ConvertedCouchbaseDocument();
+    final CouchbaseDocument converted = new CouchbaseDocument();
     couchbaseConverter.write(objectToSave, converted);
 
     execute(new BucketCallback<OperationFuture<Boolean>>() {
       @Override
       public OperationFuture<Boolean> doInBucket() {
         return client.replace(
-          converted.getId(), converted.getExpiry(), converted.getRawValue());
+          converted.getId(), converted.getExpiration(), translateEncode(converted));
       }
     });
 
@@ -152,9 +163,9 @@ public class CouchbaseTemplate implements CouchbaseOperations {
   	if (result == null) {
   		return null;
   	}
-  	
-  	ConvertedCouchbaseDocument converted = new ConvertedCouchbaseDocument(id, result);
-  	return couchbaseConverter.read(entityClass, converted);
+
+    CouchbaseDocument converted = new CouchbaseDocument(id);
+  	return couchbaseConverter.read(entityClass, (CouchbaseDocument) translateDecode(result, converted));
   }
 
 
@@ -173,9 +184,9 @@ public class CouchbaseTemplate implements CouchbaseOperations {
 
     List<T> result = new ArrayList<T>(response.size());
     for (ViewRow row : response) {
-      ConvertedCouchbaseDocument converted =
-        new ConvertedCouchbaseDocument(row.getId(), (String) row.getDocument());
-      result.add(couchbaseConverter.read(entityClass, converted));
+      CouchbaseDocument converted = new CouchbaseDocument(row.getId());
+      result.add(couchbaseConverter.read(entityClass,
+        (CouchbaseDocument) translateDecode((String) row.getDocument(), converted)));
     }
 
     return result;
@@ -206,7 +217,7 @@ public class CouchbaseTemplate implements CouchbaseOperations {
       return;
     }
 
-    final ConvertedCouchbaseDocument converted = new ConvertedCouchbaseDocument();
+    final CouchbaseDocument converted = new CouchbaseDocument();
     couchbaseConverter.write(objectToRemove, converted);
 
     execute(new BucketCallback<OperationFuture<Boolean>>() {
@@ -257,7 +268,7 @@ public class CouchbaseTemplate implements CouchbaseOperations {
 	}
 
 	private RuntimeException potentiallyConvertRuntimeException(final RuntimeException ex) {
-		RuntimeException resolved = this.exceptionTranslator.translateExceptionIfPossible(ex);
+		RuntimeException resolved = exceptionTranslator.translateExceptionIfPossible(ex);
 		return resolved == null ? ex : resolved;
 	}
 

src/main/java/org/springframework/data/couchbase/core/convert/AbstractCouchbaseConverter.java

@@ -29,6 +29,7 @@ public abstract class AbstractCouchbaseConverter implements CouchbaseConverter,
 
   protected final GenericConversionService conversionService;
   protected EntityInstantiators instantiators = new EntityInstantiators();
+  protected CustomConversions conversions = new CustomConversions();
 
   public AbstractCouchbaseConverter(
     GenericConversionService conversionService) {

src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseConverter.java

@@ -18,7 +18,7 @@ package org.springframework.data.couchbase.core.convert;
 
 import org.springframework.data.convert.EntityConverter;
 import org.springframework.data.convert.EntityReader;
-import org.springframework.data.couchbase.core.mapping.ConvertedCouchbaseDocument;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
 
@@ -27,7 +27,7 @@ import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProper
  */
 public interface CouchbaseConverter extends
   EntityConverter<CouchbasePersistentEntity<?>,
-    CouchbasePersistentProperty, Object, ConvertedCouchbaseDocument>,
-  CouchbaseWriter<Object, ConvertedCouchbaseDocument>,
-  EntityReader<Object, ConvertedCouchbaseDocument> {
+    CouchbasePersistentProperty, Object, CouchbaseDocument>,
+  CouchbaseWriter<Object, CouchbaseDocument>,
+  EntityReader<Object, CouchbaseDocument> {
 }

src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseDocumentPropertyAccessor.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert;
+
+import org.springframework.context.expression.MapAccessor;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.TypedValue;
+
+import java.util.Map;
+
+/**
+ * @author Michael Nitschinger
+ */
+public class CouchbaseDocumentPropertyAccessor extends MapAccessor {
+
+  static MapAccessor INSTANCE = new CouchbaseDocumentPropertyAccessor();
+
+  @Override
+  public Class<?>[] getSpecificTargetClasses() {
+    return new Class[] {CouchbaseDocument.class};
+  }
+
+  @Override
+  public boolean canRead(EvaluationContext context, Object target, String name) {
+    return true;
+  }
+
+  @Override
+  public TypedValue read(EvaluationContext contect, Object target, String name) {
+    Map<String, Object> source = (Map<String, Object>) target;
+
+    Object value = source.get(name);
+    return value == null ? TypedValue.NULL : new TypedValue(value);
+  }
+}

src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseTypeMapper.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert;
+
+import org.springframework.data.convert.TypeMapper;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+
+/**
+ * @author Michael Nitschinger
+ */
+public interface CouchbaseTypeMapper extends TypeMapper<CouchbaseDocument> {
+
+}

src/main/java/org/springframework/data/couchbase/core/convert/CustomConversions.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert;
+
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.util.Assert;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Value object to capture custom conversion.
+ *
+ * Types that can be mapped directly onto JSON are considered simple ones,
+ * because they neither need deeper inspection nor nested conversion.
+ *
+ * @author Michael Nitschinger
+ */
+public class CustomConversions {
+
+  private final SimpleTypeHolder simpleTypeHolder;
+
+  CustomConversions() {
+    this(new ArrayList<Object>());
+  }
+
+  public CustomConversions(final List<?> converters) {
+    Assert.notNull(converters);
+
+    simpleTypeHolder = new SimpleTypeHolder();
+  }
+
+  public boolean isSimpleType(Class<?> type) {
+    return simpleTypeHolder.isSimpleType(type);
+  }
+
+}

src/main/java/org/springframework/data/couchbase/core/convert/DefaultCouchbaseTypeMapper.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert;
+
+import org.springframework.data.convert.DefaultTypeMapper;
+import org.springframework.data.convert.TypeAliasAccessor;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+
+/**
+ * @author Michael Nitschinger
+ */
+public class DefaultCouchbaseTypeMapper extends DefaultTypeMapper<CouchbaseDocument> implements CouchbaseTypeMapper {
+
+  public static final String DEFAULT_TYPE_KEY = "_class";
+
+  public DefaultCouchbaseTypeMapper(final String typeKey) {
+    super(new CouchbaseDocumentTypeAliasAccessor(typeKey));
+  }
+
+  public static final class CouchbaseDocumentTypeAliasAccessor implements TypeAliasAccessor<CouchbaseDocument> {
+
+    private final String typeKey;
+
+    public CouchbaseDocumentTypeAliasAccessor(final String typeKey) {
+      this.typeKey = typeKey;
+    }
+
+    @Override
+    public Object readAliasFrom(final CouchbaseDocument source) {
+      return source.get(typeKey);
+    }
+
+    @Override
+    public void writeTypeTo(final CouchbaseDocument sink, final Object alias) {
+      if (typeKey != null) {
+        sink.put(typeKey, alias);
+      }
+    }
+  }
+
+}

src/main/java/org/springframework/data/couchbase/core/convert/MappingCouchbaseConverter.java

@@ -17,31 +17,26 @@
 package org.springframework.data.couchbase.core.convert;
 
 
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.core.JsonFactory;
-import com.fasterxml.jackson.core.JsonEncoding;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.*;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.CollectionFactory;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.ConversionServiceFactory;
 import org.springframework.data.convert.EntityInstantiator;
-import org.springframework.data.couchbase.core.mapping.ConvertedCouchbaseDocument;
-import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
-import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
+import org.springframework.data.couchbase.core.mapping.*;
+import org.springframework.data.mapping.Association;
+import org.springframework.data.mapping.AssociationHandler;
+import org.springframework.data.mapping.PreferredConstructor.Parameter;
 import org.springframework.data.mapping.context.MappingContext;
-import org.springframework.data.mapping.model.BeanWrapper;
-import org.springframework.data.mapping.model.MappingException;
-import org.springframework.data.mapping.model.ParameterValueProvider;
-import org.springframework.data.mapping.model.PersistentEntityParameterValueProvider;
-import org.springframework.data.mapping.model.PropertyValueProvider;
+import org.springframework.data.mapping.model.*;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * @author Michael Nitschinger
@@ -53,6 +48,8 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
   protected final MappingContext<? extends CouchbasePersistentEntity<?>,
       CouchbasePersistentProperty> mappingContext;
   protected boolean useFieldAccessOnly = true;
+  protected CouchbaseTypeMapper typeMapper;
+  private SpELContext spELContext;
 
   @SuppressWarnings("deprecation")
   public MappingCouchbaseConverter(MappingContext<? extends CouchbasePersistentEntity<?>,
@@ -60,6 +57,9 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
     super(ConversionServiceFactory.createDefaultConversionService());
 
     this.mappingContext = mappingContext;
+    typeMapper = new DefaultCouchbaseTypeMapper(DefaultCouchbaseTypeMapper.DEFAULT_TYPE_KEY);
+
+    spELContext = new SpELContext(CouchbaseDocumentPropertyAccessor.INSTANCE);
   }
 
   @Override
@@ -68,131 +68,360 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
     return mappingContext;
   }
 
-  
-	private ParameterValueProvider<CouchbasePersistentProperty> getParameterProvider(
-			CouchbasePersistentEntity<?> entity, ConvertedCouchbaseDocument source, Object parent) {
-		
-		CouchbasePropertyValueProvider provider = new CouchbasePropertyValueProvider(source, parent);
-		PersistentEntityParameterValueProvider<CouchbasePersistentProperty> parameterProvider = 
-				new PersistentEntityParameterValueProvider<CouchbasePersistentProperty>(
-						entity, provider, parent);
-		
-		return parameterProvider;
-	}
-	
   @Override
-  public <R> R read(Class<R> type, ConvertedCouchbaseDocument doc) {
-  	return read(type, doc, null);
-  }
-  
-  public <R> R read(Class<R> type, final ConvertedCouchbaseDocument doc, Object parent) {
-  	final CouchbasePersistentEntity<R> entity  = (CouchbasePersistentEntity<R>) 
-  			mappingContext.getPersistentEntity(type);
-  	
-  	ParameterValueProvider<CouchbasePersistentProperty> provider = 
-  			getParameterProvider(entity, doc, parent);
-  	EntityInstantiator instantiator = instantiators.getInstantiatorFor(entity);
-  	R instance = instantiator.createInstance(entity, provider);
-  	
-  	final BeanWrapper<CouchbasePersistentEntity<R>, R> wrapper = 
-  			BeanWrapper.create(instance, conversionService);
-		final R result = wrapper.getBean();
-		
-		// Set properties not already set in the constructor
-		entity.doWithProperties(new PropertyHandler<CouchbasePersistentProperty>() {
-			public void doWithPersistentProperty(CouchbasePersistentProperty prop) {
-
-				boolean isConstructorProperty = entity.isConstructorArgument(prop);
-				boolean hasValueForProperty = doc.containsField(prop.getFieldName());
-
-				if (!hasValueForProperty || isConstructorProperty) {
-					return;
-				}
-
-				Object obj = null;
-				if(prop.isIdProperty()) {
-					obj = doc.getId();
-				} else {
-					obj = doc.get(prop.getFieldName());
-				}
-				wrapper.setProperty(prop, obj, useFieldAccessOnly);
-			}
-		});
-		
-		return result;
+  public <R> R read(Class<R> clazz, CouchbaseDocument doc) {
+  	return read(ClassTypeInformation.from(clazz), doc, null);
   }
 
+  protected <R> R read(TypeInformation<R> type, CouchbaseDocument doc) {
+    return read(type, doc, null);
+  }
+  
+  protected <R> R read(TypeInformation<R> type, final CouchbaseDocument source, Object parent) {
+
+    if (source == null) {
+      return null;
+    }
+
+    TypeInformation<? extends R> typeToUse = typeMapper.readType(source, type);
+    Class<? extends R> rawType = typeToUse.getType();
+
+    if (typeToUse.isMap()) {
+      return (R) readMap(typeToUse, source, parent);
+    }
+
+    CouchbasePersistentEntity<R> persistentEntity = (CouchbasePersistentEntity<R>)
+      mappingContext.getPersistentEntity(typeToUse);
+
+    if (persistentEntity == null) {
+      throw new MappingException("No mapping metadata found for " + rawType.getName());
+    }
+
+    return read(persistentEntity, source, parent);
+  }
+
+  protected <R extends Object> R read(final CouchbasePersistentEntity<R> entity, final CouchbaseDocument source, final Object parent) {
+    final DefaultSpELExpressionEvaluator evaluator = new DefaultSpELExpressionEvaluator(source, spELContext);
+
+    ParameterValueProvider<CouchbasePersistentProperty> provider = getParameterProvider(entity, source, evaluator, parent);
+    EntityInstantiator instantiator = instantiators.getInstantiatorFor(entity);
+    R instance = instantiator.createInstance(entity, provider);
+
+    final BeanWrapper<CouchbasePersistentEntity<R>, R> wrapper = BeanWrapper.create(instance, conversionService);
+    final R result = wrapper.getBean();
+
+    entity.doWithProperties(new PropertyHandler<CouchbasePersistentProperty>() {
+      public void doWithPersistentProperty(final CouchbasePersistentProperty prop) {
+        if (!source.containsKey(prop.getFieldName()) || entity.isConstructorArgument(prop)) {
+          return;
+        }
+
+        Object obj = prop.isIdProperty() ? source.getId() : getValueInternal(prop, source, evaluator, result);
+        wrapper.setProperty(prop, obj, useFieldAccessOnly);
+      }
+    });
+
+    entity.doWithAssociations(new AssociationHandler<CouchbasePersistentProperty>() {
+      public void doWithAssociation(final Association<CouchbasePersistentProperty> association) {
+        CouchbasePersistentProperty inverseProp = association.getInverse();
+        Object obj = getValueInternal(inverseProp, source, evaluator, result);
+
+        wrapper.setProperty(inverseProp, obj);
+      }
+    });
+
+    return result;
+  }
+
+  protected Object getValueInternal(CouchbasePersistentProperty prop, CouchbaseDocument source, SpELExpressionEvaluator eval,
+                                    Object parent) {
+
+    CouchbasePropertyValueProvider provider = new CouchbasePropertyValueProvider(source, spELContext, parent);
+    return provider.getPropertyValue(prop);
+  }
+
+  private ParameterValueProvider<CouchbasePersistentProperty> getParameterProvider(CouchbasePersistentEntity<?> entity,
+    CouchbaseDocument source, DefaultSpELExpressionEvaluator evaluator, Object parent) {
+
+    CouchbasePropertyValueProvider provider = new CouchbasePropertyValueProvider(source, evaluator, parent);
+    PersistentEntityParameterValueProvider<CouchbasePersistentProperty> parameterProvider = new PersistentEntityParameterValueProvider<CouchbasePersistentProperty>(
+      entity, provider, parent);
+    return new ConverterAwareSpELExpressionParameterValueProvider(evaluator, conversionService, parameterProvider,
+      parent);
+  }
+
+  protected Map<Object, Object> readMap(TypeInformation<?> type, CouchbaseDocument doc, Object parent) {
+    Assert.notNull(doc);
+
+    Class<?> mapType = typeMapper.readType(doc, type).getType();
+    Map<Object, Object> map = CollectionFactory.createMap(mapType, doc.export().keySet().size());
+    Map<String, Object> sourceMap = doc.getPayload();
+
+    for (Map.Entry<String, Object> entry : sourceMap.entrySet()) {
+      Object key = entry.getKey();
+      Object value = entry.getValue();
+
+      TypeInformation<?> keyTypeInformation = type.getComponentType();
+      if (keyTypeInformation != null) {
+        Class<?> keyType = keyTypeInformation.getType();
+        key = conversionService.convert(key, keyType);
+      }
+
+      TypeInformation<?> valueType = type.getMapValueType();
+
+      if (value instanceof CouchbaseDocument) {
+        map.put(key, read(valueType, (CouchbaseDocument) value, parent));
+      } else if (value instanceof CouchbaseList) {
+        map.put(key, readCollection(valueType, (CouchbaseList) value, parent));
+      } else {
+        Class<?> valueClass = valueType == null ? null : valueType.getType();
+        map.put(key, getPotentiallyConvertedSimpleRead(value, valueClass));
+      }
+    }
+
+    return map;
+  }
+
+  private Object getPotentiallyConvertedSimpleRead(Object value, Class<?> target) {
+
+    if (value == null || target == null) {
+      return value;
+    }
+
+    if (Enum.class.isAssignableFrom(target)) {
+      return Enum.valueOf((Class<Enum>) target, value.toString());
+    }
+
+    return target.isAssignableFrom(value.getClass()) ? value : conversionService.convert(value, target);
+  }
+
+
   @Override
-  public void write(Object source, ConvertedCouchbaseDocument target) {
-    if(source == null) {
+  public void write(final Object source, final CouchbaseDocument target) {
+    if (source == null) {
       return;
     }
 
     TypeInformation<? extends Object> type = ClassTypeInformation.from(source.getClass());
-    try {
-      writeInternal(source, target, type);
-    } catch (IOException ex) {
-      throw new MappingException("Could not translate to JSON while converting "
-        + source.getClass().getName());
-    }
+    typeMapper.writeType(type, target);
+    writeInternal(source, target, type);
 
+    if (target.getId() == null) {
+      throw new MappingException("An ID property is needed, but not found on this entity.");
+    }
   }
 
-  protected void writeInternal(final Object source,
-    ConvertedCouchbaseDocument target, TypeInformation<?> type)
-    throws IOException {
-    CouchbasePersistentEntity<?> entity  = mappingContext.getPersistentEntity(
-      source.getClass());
-
-    if(entity == null) {
-      throw new MappingException("No mapping metadata found for entity of type "
-        + source.getClass().getName());
+  protected void writeInternal(final Object source, final CouchbaseDocument target, final TypeInformation<?> typeHint) {
+    if (source == null) {
+      return;
     }
 
+    if (Map.class.isAssignableFrom(source.getClass())) {
+      writeMapInternal((Map<Object, Object>) source, target, ClassTypeInformation.MAP);
+      return;
+    }
+
+    if (Collection.class.isAssignableFrom(source.getClass())) {
+      throw new IllegalArgumentException("Root Document must be either CouchbaseDocument or Map.");
+    }
+
+    CouchbasePersistentEntity<?> entity = mappingContext.getPersistentEntity(source.getClass());
+    writeInternal(source, target, entity);
+    addCustomTypeKeyIfNecessary(typeHint, source, target);
+  }
+
+  protected void writeInternal(final Object source, final CouchbaseDocument target, final CouchbasePersistentEntity<?> entity) {
+    if (source == null) {
+      return;
+    }
+
+    if (entity == null) {
+      throw new MappingException("No mapping metadata found for entity of type " + source.getClass().getName());
+    }
+
+    final BeanWrapper<CouchbasePersistentEntity<Object>, Object> wrapper = BeanWrapper.create(source, conversionService);
     final CouchbasePersistentProperty idProperty = entity.getIdProperty();
-    if(idProperty == null) {
-      throw new MappingException("ID property required for entity of type "
-        + source.getClass().getName());
+    if (idProperty != null && target.getId() == null) {
+      String id = wrapper.getProperty(idProperty, String.class, useFieldAccessOnly);
+      target.setId(id);
     }
+    target.setExpiration(entity.getExpiry());
 
-    final BeanWrapper<CouchbasePersistentEntity<Object>, Object> wrapper =
-      BeanWrapper.create(source, conversionService);
-
-    String id = wrapper.getProperty(idProperty, String.class, false);
-    target.setId(id);
-    target.setExpiry(entity.getExpiry());
-
-    JsonFactory jsonFactory = new JsonFactory();
-    OutputStream jsonStream = new ByteArrayOutputStream();
-    final JsonGenerator jsonGenerator = jsonFactory.createJsonGenerator(
-      jsonStream, JsonEncoding.UTF8);
-    jsonGenerator.setCodec(new ObjectMapper());
-
-    jsonGenerator.writeStartObject();
     entity.doWithProperties(new PropertyHandler<CouchbasePersistentProperty>() {
-      @Override
-      public void doWithPersistentProperty(CouchbasePersistentProperty prop) {
-        if(prop.equals(idProperty)) {
+      public void doWithPersistentProperty(final CouchbasePersistentProperty prop) {
+        if (prop.equals(idProperty)) {
           return;
         }
 
-        Object propertyValue = wrapper.getProperty(prop, prop.getType(), false);
-        if(propertyValue != null) {
-          try {
-            jsonGenerator.writeFieldName(prop.getFieldName());
-            jsonGenerator.writeObject(propertyValue);
-          } catch (IOException ex) {
-            throw new MappingException("Could not translate to JSON while converting "
-              + source.getClass().getName());
+        Object propertyObj = wrapper.getProperty(prop, prop.getType(), useFieldAccessOnly);
+        if (null != propertyObj) {
+          if (!conversions.isSimpleType(propertyObj.getClass())) {
+            writePropertyInternal(propertyObj, target, prop);
+          } else {
+            writeSimpleInternal(propertyObj, target, prop.getFieldName());
           }
         }
-
       }
     });
-    jsonGenerator.writeEndObject();
-    jsonGenerator.close();
 
-    target.setRawValue(jsonStream.toString());
+    entity.doWithAssociations(new AssociationHandler<CouchbasePersistentProperty>() {
+      @Override
+      public void doWithAssociation(final Association<CouchbasePersistentProperty> association) {
+        CouchbasePersistentProperty inverseProp = association.getInverse();
+        Class<?> type = inverseProp.getType();
+        Object propertyObj = wrapper.getProperty(inverseProp, type, useFieldAccessOnly);
+        if (null != propertyObj) {
+          writePropertyInternal(propertyObj, target, inverseProp);
+        }
+      }
+    });
+
+  }
+
+  private void writePropertyInternal(final Object source, final CouchbaseDocument target, final CouchbasePersistentProperty prop) {
+    if (source == null) {
+      return;
+    }
+
+    String name = prop.getFieldName();
+    TypeInformation<?> valueType = ClassTypeInformation.from(source.getClass());
+    TypeInformation<?> type = prop.getTypeInformation();
+
+    if (valueType.isCollectionLike()) {
+      CouchbaseList collectionDoc = createCollection(asCollection(source), prop);
+      target.put(name, collectionDoc);
+      return;
+    }
+
+    if (valueType.isMap()) {
+      CouchbaseDocument mapDoc = createMap((Map<Object, Object>) source, prop);
+      target.put(name, mapDoc);
+      return;
+    }
+
+    CouchbaseDocument propertyDoc = new CouchbaseDocument();
+    addCustomTypeKeyIfNecessary(type, source, propertyDoc);
+
+    CouchbasePersistentEntity<?> entity = isSubtype(prop.getType(), source.getClass()) ? mappingContext
+      .getPersistentEntity(source.getClass()) : mappingContext.getPersistentEntity(type);
+    writeInternal(source, propertyDoc, entity);
+    target.put(name, propertyDoc);
+  }
+
+  private CouchbaseDocument createMap(Map<Object, Object> map, CouchbasePersistentProperty prop) {
+    Assert.notNull(map, "Given map must not be null!");
+    Assert.notNull(prop, "PersistentProperty must not be null!");
+
+    return writeMapInternal(map, new CouchbaseDocument(), prop.getTypeInformation());
+  }
+
+  private CouchbaseDocument writeMapInternal(Map<Object,Object> source, CouchbaseDocument target, TypeInformation<?> type) {
+    for (Map.Entry<Object, Object> entry : source.entrySet()) {
+      Object key = entry.getKey();
+      Object val = entry.getValue();
+
+      if (conversions.isSimpleType(key.getClass())) {
+        String simpleKey = key.toString();
+
+        if (val == null || conversions.isSimpleType(val.getClass())) {
+          writeSimpleInternal(val, target, simpleKey);
+        } else if (val instanceof Collection || val.getClass().isArray()) {
+          target.put(simpleKey, writeCollectionInternal(asCollection(val), new CouchbaseList(), type.getMapValueType()));
+        } else {
+          CouchbaseDocument embeddedDoc = new CouchbaseDocument();
+          TypeInformation<?> valueTypeInfo = type.isMap() ? type.getMapValueType() : ClassTypeInformation.OBJECT;
+          writeInternal(val, embeddedDoc, valueTypeInfo);
+          target.put(simpleKey, embeddedDoc);
+        }
+      } else {
+        throw new MappingException("Cannot use a complex object as a key value.");
+      }
+    }
+
+    return target;
+  }
+
+  private CouchbaseList createCollection(Collection<?> collection, CouchbasePersistentProperty prop) {
+    return writeCollectionInternal(collection, new CouchbaseList(), prop.getTypeInformation());
+  }
+
+  private CouchbaseList writeCollectionInternal(Collection<?> source, CouchbaseList target, TypeInformation<?> type) {
+    TypeInformation<?> componentType = type == null ? null : type.getComponentType();
+
+    for (Object element : source) {
+      Class<?> elementType = element == null ? null : element.getClass();
+
+      if (elementType == null || conversions.isSimpleType(elementType)) {
+        target.put(element);
+      } else if (element instanceof Collection || elementType.isArray()) {
+        target.put(writeCollectionInternal(asCollection(element), new CouchbaseList(), componentType));
+      } else {
+        CouchbaseDocument embeddedDoc = new CouchbaseDocument();
+        writeInternal(element, embeddedDoc, componentType);
+        target.put(embeddedDoc);
+      }
+
+    }
+
+    return target;
+  }
+
+  private Object readCollection(final TypeInformation<?> targetType, final CouchbaseList source, final Object parent) {
+    Assert.notNull(targetType);
+
+    Class<?> collectionType = targetType.getType();
+    if (source.isEmpty()) {
+      return getPotentiallyConvertedSimpleRead(new HashSet<Object>(), collectionType);
+    }
+
+    collectionType = Collection.class.isAssignableFrom(collectionType) ? collectionType : List.class;
+    Collection<Object> items = targetType.getType().isArray() ? new ArrayList<Object>() : CollectionFactory
+      .createCollection(collectionType, source.size(false));
+    TypeInformation<?> componentType = targetType.getComponentType();
+    Class<?> rawComponentType = componentType == null ? null : componentType.getType();
+
+    for (int i = 0; i < source.size(false); i++) {
+
+      Object dbObjItem = source.get(i);
+
+      if (dbObjItem instanceof CouchbaseDocument) {
+        items.add(read(componentType, (CouchbaseDocument) dbObjItem, parent));
+      } else if (dbObjItem instanceof CouchbaseList) {
+        items.add(readCollection(componentType, (CouchbaseList) dbObjItem, parent));
+      } else {
+        items.add(getPotentiallyConvertedSimpleRead(dbObjItem, rawComponentType));
+      }
+    }
+
+    return getPotentiallyConvertedSimpleRead(items, targetType.getType());
+  }
+
+
+  private static Collection<?> asCollection(final Object source) {
+
+    if (source instanceof Collection) {
+      return (Collection<?>) source;
+    }
+
+    return source.getClass().isArray() ? CollectionUtils.arrayToList(source) : Collections.singleton(source);
+  }
+
+  private boolean isSubtype(final Class<?> left, final Class<?> right) {
+    return left.isAssignableFrom(right) && !left.equals(right);
+  }
+
+  private void writeSimpleInternal(Object source, CouchbaseDocument target, String key) {
+    target.put(key, source);
+  }
+
+  protected void addCustomTypeKeyIfNecessary(TypeInformation<?> type, Object source, CouchbaseDocument target) {
+    TypeInformation<?> actualType = type != null ? type.getActualType() : type;
+    Class<?> reference = actualType == null ? Object.class : actualType.getType();
+
+    boolean notTheSameClass = !source.getClass().equals(reference);
+    if (notTheSameClass) {
+      typeMapper.writeType(source.getClass(), target);
+    }
   }
 
   @Override
@@ -200,33 +429,71 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
     throws BeansException {
     this.applicationContext = applicationContext;
   }
-  
-	private class CouchbasePropertyValueProvider implements PropertyValueProvider<CouchbasePersistentProperty> {
 
-		private final ConvertedCouchbaseDocument source;
-		private final Object parent;
+  private class CouchbasePropertyValueProvider implements PropertyValueProvider<CouchbasePersistentProperty> {
 
-		public CouchbasePropertyValueProvider(ConvertedCouchbaseDocument source, Object parent) {
-			Assert.notNull(source);
-			this.source = source;
-			this.parent = parent;
-		}
+    private final CouchbaseDocument source;
+    private final SpELExpressionEvaluator evaluator;
+    private final Object parent;
 
-		public <T> T getPropertyValue(CouchbasePersistentProperty property) {
-			T value = null;
-			
-			if(property.isIdProperty()) {
-				value = (T) source.getId();
-			} else {
-				value = (T) source.get(property.getFieldName());
-			}
-			
-			if (value == null) {
-				return null;
-			}
+    public CouchbasePropertyValueProvider(CouchbaseDocument source, SpELContext factory, Object parent) {
+      this(source, new DefaultSpELExpressionEvaluator(source, factory), parent);
+    }
 
-			return value;
-		}
-	}
+    public CouchbasePropertyValueProvider(CouchbaseDocument source, DefaultSpELExpressionEvaluator evaluator, Object parent) {
+
+      Assert.notNull(source);
+      Assert.notNull(evaluator);
+
+      this.source = source;
+      this.evaluator = evaluator;
+      this.parent = parent;
+    }
+
+    public <R> R getPropertyValue(final CouchbasePersistentProperty property) {
+
+      String expression = property.getSpelExpression();
+      Object value = expression != null ? evaluator.evaluate(expression) : source.get(property.getFieldName());
+
+      if (property.isIdProperty()) {
+        return (R) source.getId();
+      }
+
+      if (value == null) {
+        return null;
+      }
+
+      return readValue(value, property.getTypeInformation(), parent);
+    }
+  }
+
+  private <R> R readValue(Object value, TypeInformation<?> type, Object parent) {
+    Class<?> rawType = type.getType();
+
+    if (value instanceof CouchbaseDocument) {
+      return (R) read(type, (CouchbaseDocument) value, parent);
+    } else if (value instanceof CouchbaseList) {
+      return (R) readCollection(type, (CouchbaseList) value, parent);
+    } else {
+      return (R) getPotentiallyConvertedSimpleRead(value, rawType);
+    }
+  }
+
+  private class ConverterAwareSpELExpressionParameterValueProvider extends
+    SpELExpressionParameterValueProvider<CouchbasePersistentProperty> {
+    private final Object parent;
+
+    public ConverterAwareSpELExpressionParameterValueProvider(SpELExpressionEvaluator evaluator,
+       ConversionService conversionService, ParameterValueProvider<CouchbasePersistentProperty> delegate, Object parent) {
+
+      super(evaluator, conversionService, delegate);
+      this.parent = parent;
+    }
+
+    @Override
+    protected <T> T potentiallyConvertSpelValue(Object object, Parameter<T, CouchbasePersistentProperty> parameter) {
+      return readValue(object, parameter.getType(), parent);
+    }
+  }
 
 }

src/main/java/org/springframework/data/couchbase/core/convert/translation/JacksonTranslationService.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert.translation;
+
+import com.fasterxml.jackson.core.*;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.data.couchbase.core.mapping.CouchbaseList;
+import org.springframework.data.couchbase.core.mapping.CouchbaseStorable;
+import org.springframework.data.mapping.model.MappingException;
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A Jackson JSON Translator.
+ *
+ * @author Michael Nitschinger
+ */
+public class JacksonTranslationService implements TranslationService {
+
+  private SimpleTypeHolder simpleTypeHolder = new SimpleTypeHolder();
+  private JsonFactory factory = new JsonFactory();
+
+  @Override
+  public final Object encode(final CouchbaseStorable source) {
+    OutputStream stream = new ByteArrayOutputStream();
+
+    try {
+      JsonGenerator generator = factory.createGenerator(stream, JsonEncoding.UTF8);
+      encodeRecursive(source, generator);
+      generator.close();
+    } catch (IOException ex) {
+      throw new RuntimeException("Could not encode JSON", ex);
+    }
+
+    return stream.toString();
+  }
+
+  /**
+   * Recursively iterates through the sources and adds it to the JSON generator.
+   *
+   * @param source the source document
+   * @param generator the JSON generator.
+   * @throws IOException
+   */
+  private void encodeRecursive(final CouchbaseStorable source, final JsonGenerator generator) throws IOException {
+    generator.writeStartObject();
+
+    for (Map.Entry<String, Object> entry : ((CouchbaseDocument) source).export().entrySet()) {
+      String key = entry.getKey();
+      Object value = entry.getValue();
+      generator.writeFieldName(key);
+      if (value instanceof CouchbaseDocument) {
+        encodeRecursive((CouchbaseDocument) value, generator);
+        continue;
+      }
+
+      if (simpleTypeHolder.isSimpleType(value.getClass())) {
+        generator.writeObject(value);
+      } else {
+        ObjectMapper mapper = new ObjectMapper();
+        mapper.writeValue(generator, value);
+      }
+
+    }
+
+    generator.writeEndObject();
+  }
+
+  @Override
+  public final CouchbaseStorable decode(final Object source, final CouchbaseStorable target) {
+    try {
+      JsonParser parser = factory.createParser((String) source);
+      while (parser.nextToken() != null) {
+        JsonToken currentToken = parser.getCurrentToken();
+
+        if (currentToken == JsonToken.START_OBJECT) {
+          return decodeObject(parser, (CouchbaseDocument) target);
+        } else if (currentToken == JsonToken.START_ARRAY) {
+          return decodeArray(parser, new CouchbaseList());
+        } else {
+          throw new MappingException("JSON to decode needs to start as array or object!");
+        }
+      }
+      parser.close();
+    } catch (IOException ex) {
+      throw new RuntimeException("Could not decode JSON", ex);
+    }
+    return target;
+  }
+
+  private CouchbaseDocument decodeObject(final JsonParser parser, final CouchbaseDocument target) throws IOException {
+    JsonToken currentToken = parser.nextToken();
+
+    String fieldName = "";
+    while (currentToken != null && currentToken != JsonToken.END_OBJECT) {
+      if (currentToken == JsonToken.START_OBJECT) {
+        target.put(fieldName, decodeObject(parser, new CouchbaseDocument()));
+      } else if (currentToken == JsonToken.START_ARRAY) {
+        target.put(fieldName, decodeArray(parser, new CouchbaseList()));
+      } else if (currentToken == JsonToken.FIELD_NAME) {
+        fieldName = parser.getCurrentName();
+      } else {
+        target.put(fieldName, decodePrimitive(currentToken, parser));
+      }
+
+      currentToken = parser.nextToken();
+    }
+
+    return target;
+  }
+
+  private CouchbaseList decodeArray(final JsonParser parser, final CouchbaseList target) throws IOException {
+    JsonToken currentToken = parser.nextToken();
+
+    while (currentToken != null && currentToken != JsonToken.END_ARRAY) {
+      if (currentToken == JsonToken.START_OBJECT) {
+        target.put(decodeObject(parser, new CouchbaseDocument()));
+      } else if (currentToken == JsonToken.START_ARRAY) {
+        target.put(decodeArray(parser, new CouchbaseList()));
+      } else {
+        target.put(decodePrimitive(currentToken, parser));
+      }
+
+      currentToken = parser.nextToken();
+    }
+
+    return target;
+  }
+
+  private Object decodePrimitive(final JsonToken token, final JsonParser parser) throws IOException {
+    switch (token) {
+      case VALUE_TRUE:
+      case VALUE_FALSE:
+        return parser.getValueAsBoolean();
+      case VALUE_STRING:
+        return parser.getValueAsString();
+      case VALUE_NUMBER_INT:
+        return parser.getValueAsInt();
+      case VALUE_NUMBER_FLOAT:
+        return parser.getValueAsDouble();
+      default:
+        throw new MappingException("Could not decode primitve value " + token);
+    }
+  }
+
+
+}

src/main/java/org/springframework/data/couchbase/core/convert/translation/TranslationService.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.convert.translation;
+
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.data.couchbase.core.mapping.CouchbaseStorable;
+
+/**
+ * @author Michael Nitschinger
+ */
+public interface TranslationService<T> {
+
+  /**
+   * Encodes a {@link CouchbaseDocument} into the target format.
+   *
+   * @param source the source document to encode.
+   * @return the encoded document representation.
+   */
+  T encode(CouchbaseStorable source);
+
+  /**
+   * Decodes the target format into a {@link CouchbaseDocument}
+   *
+   * @param source the source formatted document.
+   * @param target the target of the populated data.
+   * @return a properly populated document to work with.
+   */
+  CouchbaseStorable decode(T source, CouchbaseStorable target);
+}

src/main/java/org/springframework/data/couchbase/core/mapping/ConvertedCouchbaseDocument.java

@@ -1,102 +0,0 @@
-/*
- * Copyright 2013 the original author or authors.
- *
- * 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.springframework.data.couchbase.core.mapping;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.springframework.data.mapping.model.MappingException;
-
-/**
- * @author Michael Nitschinger
- */
-public class ConvertedCouchbaseDocument {
-
-  private String id;
-
-  private String rawValue;
-
-  private int expiry;
-  
-  private Map<String, Object> decoded;
-
-  public ConvertedCouchbaseDocument() {
-    this("", "", 0);
-  }
-
-  public ConvertedCouchbaseDocument(String id, String rawValue) {
-    this(id, rawValue, 0);
-  }
-
-  public ConvertedCouchbaseDocument(String id, String rawValue, int expiry) {
-    this.id = id;
-    this.rawValue = rawValue;
-    this.expiry = expiry;
-    this.decoded = new HashMap<String, Object>();
-    parseJson();
-  }
-
-  public void setId(String id) {
-    this.id = id;
-  }
-
-  public String getId() {
-    return id;
-  }
-
-  public String getRawValue() {
-    return rawValue;
-  }
-
-  public void setRawValue(String value) {
-    this.rawValue = value;
-    parseJson();
-    
-  }
-
-  public int getExpiry() {
-    return expiry;
-  }
-
-  public void setExpiry(int expiry) {
-    this.expiry = expiry;
-  }
-  
-  public boolean containsField(String fieldname) {
-  	return decoded.containsKey(fieldname);
-  }
-  
-  public Object get(String fieldname) {
-  	return decoded.get(fieldname);
-  }
-  
-  private void parseJson() {
-  	ObjectMapper mapper = new ObjectMapper();
-  	try {
-  		if(!getRawValue().isEmpty()) {
-  			Map<String, Object> converted = mapper.readValue(getRawValue(), 
-  				new TypeReference<Map<String, Object>>() { });
-  			this.decoded = converted;
-  		}
-  	} catch(Exception e) {
-  		throw new MappingException("Error while decoding JSON object.", e);
-  	}
-  }
-
-}

src/main/java/org/springframework/data/couchbase/core/mapping/CouchbaseDocument.java

@@ -0,0 +1,278 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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.springframework.data.couchbase.core.mapping;
+
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A {@link CouchbaseDocument} is an abstract representation of a document stored
+ * inside Couchbase Server.
+ *
+ * <p>It acts like a {@link HashMap}, but only allows those types to be written
+ * that are supported by the underlying storage format, which is currently
+ * JSON. Note that JSON conversion is not happening here, but performed
+ * at a different stage based on the payload stored in the
+ * {@link CouchbaseDocument}.</p>
+ *
+ * <p>In addition to the actual content, meta data is also stored. This
+ * especially refers to the document ID and its expiration time. Note that
+ * this information is not mandatory, since documents can be nested and
+ * therefore only the topmost document most likely has an ID.</p>
+ *
+ * @author Michael Nitschinger
+ */
+public class CouchbaseDocument implements CouchbaseStorable {
+
+  /**
+   * Defnes the default expiration time for the document.
+   */
+  public static final int DEFAULT_EXPIRATION_TIME = 0;
+
+  /**
+   * Contains the actual data to be stored.
+   */
+  private HashMap<String, Object> payload;
+
+  /**
+   * Represents the document ID used to identify the document in the bucket.
+   */
+  private String id;
+
+  /**
+   * Contains the expiration time of the document.
+   */
+  private int expiration;
+
+  /**
+   * Holds types considered simple and allowed to be stored.
+   */
+  private SimpleTypeHolder simpleTypeHolder;
+
+  /**
+   * Creates a completely empty {@link CouchbaseDocument}.
+   */
+  public CouchbaseDocument() {
+    this(null);
+  }
+
+  /**
+   * Creates a empty {@link CouchbaseDocument}, and set the ID immediately.
+   *
+   * @param id the document ID.
+   */
+  public CouchbaseDocument(final String id) {
+    this(id, DEFAULT_EXPIRATION_TIME);
+  }
+
+  /**
+   * Creates a empty {@link CouchbaseDocument} with ID and expiration time.
+   *
+   * @param id the document ID.
+   * @param expiration the expiration time of the document.
+   */
+  public CouchbaseDocument(final String id, final int expiration) {
+    this.id = id;
+    this.expiration = expiration;
+    payload = new HashMap<String, Object>();
+
+    Set<Class<?>> additionalTypes = new HashSet<Class<?>>();
+    additionalTypes.add(CouchbaseDocument.class);
+    additionalTypes.add(CouchbaseList.class);
+    simpleTypeHolder = new SimpleTypeHolder(additionalTypes, true);
+  }
+
+  /**
+   * Store a value with the given key for later retreival.
+   *
+   * @param key the key of the attribute.
+   * @param value the actual content to be stored.
+   * @return the {@link CouchbaseDocument} for chaining.
+   */
+  public final CouchbaseDocument put(final String key, final Object value) {
+    verifyValueType(value.getClass());
+
+    payload.put(key, value);
+    return this;
+  }
+
+  /**
+   * Potentially get a value from the payload with the given key.
+   *
+   * @param key the key of the attribute.
+   * @return the value to which the specified key is mapped, or
+   *         null if does not contain a mapping for the key.
+   */
+  public final Object get(final String key) {
+    return payload.get(key);
+  }
+
+  /**
+   * Returns the current payload, including all recursive elements.
+   *
+   * It either returns the raw results or makes sure that the recusrive elements
+   * are also exported properly.
+   *
+   * @return
+   */
+  public final HashMap<String, Object> export() {
+    HashMap<String, Object> toExport = new HashMap<String, Object>(payload);
+    for (Map.Entry<String, Object> entry : payload.entrySet()) {
+      if (entry.getValue() instanceof CouchbaseDocument) {
+        toExport.put(entry.getKey(), ((CouchbaseDocument) entry.getValue()).export());
+      } else if (entry.getValue() instanceof CouchbaseList) {
+        toExport.put(entry.getKey(), ((CouchbaseList) entry.getValue()).export());
+      }
+    }
+    return toExport;
+  }
+
+  /**
+   * Returns true if it contains a payload for the specified key.
+   *
+   * @param key the key of the attribute.
+   * @return true if it contains a payload for the specified key.
+   */
+  public final boolean containsKey(final String key) {
+    return payload.containsKey(key);
+  }
+
+  /**
+   * Returns true if it contains the given value.
+   *
+   * @param value the value to check for.
+   * @return true if it contains the specified value.
+   */
+  public final boolean containsValue(final Object value) {
+    return payload.containsValue(value);
+  }
+
+  /**
+   * Returns the size of the attributes in this document (not nested).
+   *
+   * @return the size of the attributes in this document (not nested).
+   */
+  public final int size() {
+    return size(false);
+  }
+
+  /**
+   * Retruns the size of the attributes in this and recursive documents.
+   *
+   * @param recursive wheter nested attributes should be taken into account.
+   * @return the size of the attributes in this and recursive documents.
+   */
+  public final int size(final boolean recursive) {
+    int thisSize = payload.size();
+
+    if (!recursive || thisSize == 0) {
+      return thisSize;
+    }
+
+    int totalSize = thisSize;
+    for (Object value : payload.values()) {
+      if (value instanceof CouchbaseDocument) {
+        totalSize += ((CouchbaseDocument) value).size(true);
+      } else if (value instanceof CouchbaseList) {
+        totalSize += ((CouchbaseList) value).size(true);
+      }
+    }
+
+    return totalSize;
+  }
+
+  public HashMap<String, Object> getPayload() {
+    return payload;
+  }
+
+  /**
+   * Returns the expiration time of the document.
+   *
+   * If the expiration time is 0, then the document will be persisted until
+   * deleted manually ("forever").
+   *
+   * @return the expiration time of the document.
+   */
+  public int getExpiration() {
+    return expiration;
+  }
+
+  /**
+   * Set the expiration time of the document.
+   *
+   * If the expiration time is 0, then the document will be persisted until
+   * deleted manually ("forever").
+   *
+   * @param expiration
+   * @return the {@link CouchbaseDocument} for chaining.
+   */
+  public CouchbaseDocument setExpiration(int expiration) {
+    this.expiration = expiration;
+    return this;
+  }
+
+  /**
+   * Returns the ID of the document.
+   *
+   * @return the ID of the document.
+   */
+  public String getId() {
+    return id;
+  }
+
+  /**
+   * Sets the unique ID of the document per bucket.
+   *
+   * @param id the ID of the document.
+   * @return the {@link CouchbaseDocument} for chaining.
+   */
+  public CouchbaseDocument setId(String id) {
+    this.id = id;
+    return this;
+  }
+
+  /**
+   * Verifies that only values of a certain and supported type
+   * can be stored.
+   *
+   * <p>If this is not the case, a {@link IllegalArgumentException} is
+   * thrown.</p>
+   *
+   * @param clazz the class type to check and verify.
+   */
+  private void verifyValueType(final Class<?> clazz) {
+    if (simpleTypeHolder.isSimpleType(clazz)) {
+      return;
+    }
+
+    throw new IllegalArgumentException("Attribute of type "
+      + clazz.getCanonicalName() + " can not be stored and must be converted.");
+  }
+
+  @Override
+  public String toString() {
+    return "CouchbaseDocument{" +
+      "id=" + id +
+      ", exp=" + expiration +
+      ", payload=" + payload +
+      '}';
+  }
+}

src/main/java/org/springframework/data/couchbase/core/mapping/CouchbaseList.java

@@ -0,0 +1,148 @@
+package org.springframework.data.couchbase.core.mapping;
+
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+
+import java.util.*;
+
+/**
+ * A {@link CouchbaseList} is an abstract list that represents an array stored
+ * in a (most of the times JSON) document.
+ *
+ * This {@link CouchbaseList} is part of the potentially nested structure inside
+ * one or more {@link CouchbaseDocument}s. It can also contain them recursively,
+ * depending on how the document is modeled.
+ */
+public class CouchbaseList implements CouchbaseStorable {
+
+  /**
+   * Contains the actual data to be stored.
+   */
+  private List<Object> payload;
+
+  /**
+   * Holds types considered simple and allowed to be stored.
+   */
+  private SimpleTypeHolder simpleTypeHolder;
+
+  public CouchbaseList() {
+    this(new ArrayList<Object>());
+  }
+
+  public CouchbaseList(List<Object> initialPayload) {
+    payload = initialPayload;
+
+    Set<Class<?>> additionalTypes = new HashSet<Class<?>>();
+    additionalTypes.add(CouchbaseDocument.class);
+    additionalTypes.add(CouchbaseList.class);
+    simpleTypeHolder = new SimpleTypeHolder(additionalTypes, true);
+  }
+
+  public final CouchbaseList put(Object value) {
+    verifyValueType(value.getClass());
+
+    payload.add(value);
+    return this;
+  }
+
+  public final Object get(int index) {
+    return payload.get(index);
+  }
+
+  /**
+   * Returns the size of the attributes in this document (not nested).
+   *
+   * @return the size of the attributes in this document (not nested).
+   */
+  public final int size() {
+    return size(false);
+  }
+
+  /**
+   * Retruns the size of the attributes in this and recursive documents.
+   *
+   * @param recursive wheter nested attributes should be taken into account.
+   * @return the size of the attributes in this and recursive documents.
+   */
+  public final int size(final boolean recursive) {
+    int thisSize = payload.size();
+
+    if (!recursive || thisSize == 0) {
+      return thisSize;
+    }
+
+    int totalSize = thisSize;
+    for (Object value : payload) {
+      if (value instanceof CouchbaseDocument) {
+        totalSize += ((CouchbaseDocument) value).size(true);
+      } else if (value instanceof CouchbaseList) {
+        totalSize += ((CouchbaseList) value).size(true);
+      }
+    }
+
+    return totalSize;
+  }
+
+  /**
+   * Returns the current payload, including all recursive elements.
+   *
+   * It either returns the raw results or makes sure that the recusrive elements
+   * are also exported properly.
+   *
+   * @return
+   */
+  public final List<Object> export() {
+    List<Object> toExport = new ArrayList<Object>(payload);
+
+    int elem = 0;
+    for (Object entry : payload) {
+      if (entry instanceof CouchbaseDocument) {
+        toExport.remove(elem);
+        toExport.add(elem, ((CouchbaseDocument) entry).export());
+      } else if (entry instanceof CouchbaseList) {
+        toExport.remove(elem);
+        toExport.add(elem, ((CouchbaseList) entry).export());
+      }
+      elem++;
+    }
+    return toExport;
+  }
+
+  /**
+   * Returns true if it contains the given value.
+   *
+   * @param value the value to check for.
+   * @return true if it contains the specified value.
+   */
+  public final boolean containsValue(final Object value) {
+    return payload.contains(value);
+  }
+
+  public final boolean isEmpty() {
+    return payload.isEmpty();
+  }
+
+  /**
+   * Verifies that only values of a certain and supported type
+   * can be stored.
+   *
+   * <p>If this is not the case, a {@link IllegalArgumentException} is
+   * thrown.</p>
+   *
+   * @param clazz the class type to check and verify.
+   */
+  private void verifyValueType(final Class<?> clazz) {
+    if (simpleTypeHolder.isSimpleType(clazz)) {
+      return;
+    }
+
+    throw new IllegalArgumentException("Attribute of type "
+      + clazz.getCanonicalName() + "can not be stored and must be converted.");
+  }
+
+  @Override
+  public String toString() {
+    return "CouchbaseList{" +
+      "payload=" + payload +
+      '}';
+  }
+}

src/main/java/org/springframework/data/couchbase/core/mapping/CouchbaseStorable.java

@@ -0,0 +1,5 @@
+package org.springframework.data.couchbase.core.mapping;
+
+public interface CouchbaseStorable {
+
+}