diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoBeanWrapper.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoBeanWrapper.java
new file mode 100644
index 000000000..5305db819
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoBeanWrapper.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2011 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.document.mongodb;
+
+import static org.springframework.beans.PropertyAccessorFactory.*;
+
+import org.springframework.beans.BeanWrapper;
+import org.springframework.beans.ConfigurablePropertyAccessor;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.document.mongodb.MongoPropertyDescriptors.MongoPropertyDescriptor;
+import org.springframework.util.Assert;
+
+/**
+ * Custom Mongo specific {@link BeanWrapper} to allow access to bean properties via {@link MongoPropertyDescriptor}s.
+ * 
+ * @author Oliver Gierke
+ */
+class MongoBeanWrapper {
+
+	private final ConfigurablePropertyAccessor accessor;
+	private final MongoPropertyDescriptors descriptors;
+
+	/**
+	 * Creates a new {@link MongoBeanWrapper} for the given target object and {@link ConversionService}.
+	 * 
+	 * @param target
+	 * @param conversionService
+	 * @param fieldAccess
+	 */
+	public MongoBeanWrapper(Object target, ConversionService conversionService, boolean fieldAccess) {
+
+		Assert.notNull(target);
+		Assert.notNull(conversionService);
+
+		this.accessor = fieldAccess ? forDirectFieldAccess(target) : forBeanPropertyAccess(target);
+		this.accessor.setConversionService(conversionService);
+		this.descriptors = new MongoPropertyDescriptors(target.getClass());
+	}
+
+	/**
+	 * Returns all {@link MongoPropertyDescriptors.MongoPropertyDescriptor}s for the underlying target object.
+	 * 
+	 * @return
+	 */
+	public MongoPropertyDescriptors getDescriptors() {
+		return this.descriptors;
+	}
+
+	/**
+	 * Returns the value of the underlying object for the given property.
+	 * 
+	 * @param descriptor
+	 * @return
+	 */
+	public Object getValue(MongoPropertyDescriptors.MongoPropertyDescriptor descriptor) {
+		Assert.notNull(descriptor);
+		return accessor.getPropertyValue(descriptor.getName());
+	}
+
+	/**
+	 * Sets the property of the underlying object to the given value.
+	 * 
+	 * @param descriptor
+	 * @param value
+	 */
+	public void setValue(MongoPropertyDescriptors.MongoPropertyDescriptor descriptor, Object value) {
+		Assert.notNull(descriptor);
+		accessor.setPropertyValue(descriptor.getName(), value);
+	}
+}
\ No newline at end of file
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoConverter.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoConverter.java
index 48853c534..45c70c287 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoConverter.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoConverter.java
@@ -15,7 +15,27 @@
  */
 package org.springframework.data.document.mongodb;
 
+import org.bson.types.ObjectId;
+
 
 public interface MongoConverter extends MongoWriter<Object>, MongoReader<Object> {
 
+	/**
+	 * Converts the given {@link ObjectId} to the given target type.
+	 * 
+	 * @param <T> the actual type to create
+	 * @param id the source {@link ObjectId}
+	 * @param targetType the target type to convert the {@link ObjectId} to
+	 * @return
+	 */
+	public <T> T convertObjectId(ObjectId id, Class<T> targetType);
+	
+	
+	/**
+	 * Returns the {@link ObjectId} instance for the given id.
+	 * 
+	 * @param id
+	 * @return
+	 */
+	public ObjectId convertObjectId(Object id);
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoOperations.java
index d50012e3a..7407cc7b1 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoOperations.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoOperations.java
@@ -507,4 +507,13 @@ public interface MongoOperations {
 	 */
 	<T> List<T> query(String collectionName, DBObject query, Class<T> targetClass, MongoReader<T> reader);
 
+
+	/**
+	 * Returns the object with the given id for the given target class.
+	 *
+	 * @param targetClass the type of the object to be retrieved
+	 * @param id the id of the object to be retrieved
+	 * @return the converted object
+	 */
+	<T> T find(Class<T> targetClass, Object id);
 }
\ No newline at end of file
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoPropertyDescriptors.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoPropertyDescriptors.java
new file mode 100644
index 000000000..6e31be051
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoPropertyDescriptors.java
@@ -0,0 +1,242 @@
+/*
+ * Copyright 2011 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.document.mongodb;
+
+import java.beans.PropertyDescriptor;
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.math.BigInteger;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+import org.bson.types.ObjectId;
+import org.springframework.beans.BeanUtils;
+import org.springframework.util.Assert;
+
+/**
+ * An iterable of {@link MongoPropertyDescriptor}s that allows dedicated access to the {@link MongoPropertyDescriptor}
+ * that captures the id-property.
+ * 
+ * @author Oliver Gierke
+ */
+public class MongoPropertyDescriptors implements Iterable<MongoPropertyDescriptors.MongoPropertyDescriptor> {
+
+	private final Collection<MongoPropertyDescriptors.MongoPropertyDescriptor> descriptors;
+	private final MongoPropertyDescriptors.MongoPropertyDescriptor idDescriptor;
+
+	/**
+	 * Creates the {@link MongoPropertyDescriptors} for the given type.
+	 * 
+	 * @param type
+	 */
+	public MongoPropertyDescriptors(Class<?> type) {
+
+		Assert.notNull(type);
+		Set<MongoPropertyDescriptors.MongoPropertyDescriptor> descriptors = new HashSet<MongoPropertyDescriptors.MongoPropertyDescriptor>();
+		MongoPropertyDescriptors.MongoPropertyDescriptor idDesciptor = null;
+
+		for (PropertyDescriptor candidates : BeanUtils.getPropertyDescriptors(type)) {
+			MongoPropertyDescriptor descriptor = new MongoPropertyDescriptors.MongoPropertyDescriptor(candidates);
+			descriptors.add(descriptor);
+			if (descriptor.isIdProperty()) {
+				idDesciptor = descriptor;
+			}
+		}
+
+		this.descriptors = Collections.unmodifiableSet(descriptors);
+		this.idDescriptor = idDesciptor;
+	}
+
+	/**
+	 * Returns the {@link MongoPropertyDescriptor} for the id property.
+	 * 
+	 * @return the idDescriptor
+	 */
+	public MongoPropertyDescriptors.MongoPropertyDescriptor getIdDescriptor() {
+		return idDescriptor;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see java.lang.Iterable#iterator()
+	 */
+	public Iterator<MongoPropertyDescriptors.MongoPropertyDescriptor> iterator() {
+		return descriptors.iterator();
+	}
+
+	/**
+	 * Simple value object to have a more suitable abstraction for MongoDB specific property handling.
+	 * 
+	 * @author Oliver Gierke
+	 */
+	public static class MongoPropertyDescriptor {
+		
+		public static Collection<Class<?>> SUPPORTED_ID_CLASSES;
+		
+		static {
+			Set<Class<?>> classes = new HashSet<Class<?>>();
+			classes.add(ObjectId.class);
+			classes.add(String.class);
+			classes.add(BigInteger.class);
+			SUPPORTED_ID_CLASSES = Collections.unmodifiableCollection(classes);
+		}
+
+		private static final String ID_PROPERTY = "id";
+		static final String ID_KEY = "_id";
+
+		private final PropertyDescriptor delegate;
+
+		/**
+		 * Creates a new {@link MongoPropertyDescriptor} for the given {@link PropertyDescriptor}.
+		 * 
+		 * @param descriptor
+		 */
+		public MongoPropertyDescriptor(PropertyDescriptor descriptor) {
+			Assert.notNull(descriptor);
+			this.delegate = descriptor;
+		}
+
+		/**
+		 * Returns whether the property is the id-property. Will be identified by name for now ({@value #ID_PROPERTY}).
+		 * 
+		 * @return
+		 */
+		boolean isIdProperty() {
+			return ID_PROPERTY.equals(delegate.getName());
+		}
+
+		/**
+		 * Returns whether the property is of one of the supported id types. Currently we support {@link String},
+		 * {@link ObjectId} and {@link BigInteger}.
+		 * 
+		 * @return
+		 */
+		public boolean isOfIdType() {
+			boolean isString = String.class.isAssignableFrom(getPropertyType());
+			boolean isObjectId = ObjectId.class.isAssignableFrom(getPropertyType());
+			boolean isBigInteger = BigInteger.class.isAssignableFrom(getPropertyType());
+			return isString || isObjectId || isBigInteger;
+		}
+
+		/**
+		 * Returns the key that shall be used for mapping. Will return {@value #ID_KEY} for the id property and the
+		 * plain name for all other ones.
+		 * 
+		 * @return
+		 */
+		public String getKeyToMap() {
+			return isIdProperty() ? ID_KEY : delegate.getName();
+		}
+
+		/**
+		 * Returns the name of the property.
+		 * 
+		 * @return
+		 */
+		public String getName() {
+			return delegate.getName();
+		}
+
+		/**
+		 * Returns whether the underlying property is actually mappable. By default this will exclude the
+		 * {@literal class} property and only include properties with a getter.
+		 * 
+		 * @return
+		 */
+		public boolean isMappable() {
+			return !delegate.getName().equals("class") && delegate.getReadMethod() != null;
+		}
+
+		/**
+		 * Returns the plain property type.
+		 * 
+		 * @return
+		 */
+		public Class<?> getPropertyType() {
+			return delegate.getPropertyType();
+		}
+
+		/**
+		 * Returns the type type to be set. Will return the setter method's type and fall back to the getter method's
+		 * return type in case no setter is available. Useful for further (generics) inspection.
+		 * 
+		 * @return
+		 */
+		public Type getTypeToSet() {
+
+			Method method = delegate.getWriteMethod();
+			return method == null ? delegate.getReadMethod().getGenericReturnType()
+					: method.getGenericParameterTypes()[0];
+		}
+
+		/**
+		 * Returns whther we describe a {@link Map}.
+		 * @return
+		 */
+		public boolean isMap() {
+			return Map.class.isAssignableFrom(getPropertyType());
+		}
+
+		/**
+		 * Returns whether the descriptor is for a collection.
+		 * @return
+		 */
+		public boolean isCollection() {
+			return Collection.class.isAssignableFrom(getPropertyType());
+		}
+
+		/**
+		 * Returns whether the descriptor is for an {@link Enum}.
+		 * 
+		 * @return
+		 */
+		public boolean isEnum() {
+			return Enum.class.isAssignableFrom(getPropertyType());
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * 
+		 * @see java.lang.Object#equals(java.lang.Object)
+		 */
+		@Override
+		public boolean equals(Object obj) {
+			if (obj == this) {
+				return true;
+			}
+			if (obj == null || !getClass().equals(obj.getClass())) {
+				return false;
+			}
+			MongoPropertyDescriptor that = (MongoPropertyDescriptor) obj;
+			return that.delegate.equals(this.delegate);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * 
+		 * @see java.lang.Object#hashCode()
+		 */
+		@Override
+		public int hashCode() {
+			return delegate.hashCode();
+		}
+	}
+}
\ No newline at end of file
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoReader.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoReader.java
index 59f7bc8d4..3d1341af8 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoReader.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoReader.java
@@ -36,5 +36,5 @@ public interface MongoReader<T> {
 	 * @param dbo theDBObject
 	 * @return the converted object
 	 */
-	T read(Class<? extends T> clazz, DBObject dbo);
+	<S extends T> S read(Class<S> clazz, DBObject dbo);
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoTemplate.java
index 4a256ca10..c125c920a 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoTemplate.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/MongoTemplate.java
@@ -16,19 +16,18 @@
 
 package org.springframework.data.document.mongodb;
 
-import java.beans.PropertyDescriptor;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
 import org.bson.types.ObjectId;
-import org.springframework.beans.BeanUtils;
-import org.springframework.beans.BeanWrapper;
+import org.springframework.beans.ConfigurablePropertyAccessor;
 import org.springframework.beans.PropertyAccessorFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.document.mongodb.MongoPropertyDescriptors.MongoPropertyDescriptor;
 import org.springframework.jca.cci.core.ConnectionCallback;
 import org.springframework.util.Assert;
 
@@ -40,6 +39,7 @@ import com.mongodb.DBCursor;
 import com.mongodb.DBObject;
 import com.mongodb.Mongo;
 import com.mongodb.MongoException;
+import com.mongodb.QueryBuilder;
 import com.mongodb.util.JSON;
 
 /**
@@ -350,7 +350,7 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 	public <T> void insert(String collectionName, T objectToSave, MongoWriter<T> writer) {
 		BasicDBObject dbDoc = new BasicDBObject();
 		writer.write(objectToSave, dbDoc);
-		Object id = insertDBObject(collectionName, dbDoc);
+		ObjectId id = insertDBObject(collectionName, dbDoc);
 		populateIdIfNecessary(objectToSave, id);
 	}
 
@@ -381,7 +381,7 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 			writer.write(o, dbDoc);
 			dbObjectList.add(dbDoc);
 		}
-		List<Object> ids = insertDBObjectList(collectionName, dbObjectList);
+		List<ObjectId> ids = insertDBObjectList(collectionName, dbObjectList);
 		for (int i = 0; i < listToSave.size(); i++) {
 			if (i < ids.size()) {
 				populateIdIfNecessary(listToSave.get(i), ids.get(i));
@@ -409,21 +409,21 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 	public <T> void save(String collectionName, T objectToSave, MongoWriter<T> writer) {
 		BasicDBObject dbDoc = new BasicDBObject();
 		writer.write(objectToSave, dbDoc);
-		Object id = saveDBObject(collectionName, dbDoc);
+		ObjectId id = saveDBObject(collectionName, dbDoc);
 		populateIdIfNecessary(objectToSave, id);
 	}
 
 
-	protected Object insertDBObject(String collectionName, final DBObject dbDoc) {
+	protected ObjectId insertDBObject(String collectionName, final DBObject dbDoc) {
 
 		if (dbDoc.keySet().isEmpty()) {
 			return null;
 		}
-		
-		return execute(new CollectionCallback<Object>() {
-			public Object doInCollection(DBCollection collection) throws MongoException, DataAccessException {
+
+		return execute(new CollectionCallback<ObjectId>() {
+			public ObjectId doInCollection(DBCollection collection) throws MongoException, DataAccessException {
 				collection.insert(dbDoc);
-				return dbDoc.get(ID);
+				return (ObjectId) dbDoc.get(ID);
 			}
 		}, collectionName);
 	}
@@ -431,8 +431,8 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 	
 	
 
-	protected List<Object> insertDBObjectList(String collectionName, final List<DBObject> dbDocList) {
-		
+	protected List<ObjectId> insertDBObjectList(String collectionName, final List<DBObject> dbDocList) {
+
 		if (dbDocList.isEmpty()) {
 			return Collections.emptyList();
 		}
@@ -444,24 +444,23 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 			}
 		}, collectionName);
 
-		List<Object> ids = new ArrayList<Object>();
+		List<ObjectId> ids = new ArrayList<ObjectId>();
 		for (DBObject dbo : dbDocList) {
-			ids.add(dbo.get(ID));
+			ids.add((ObjectId) dbo.get(ID));
 		}
 		return ids;
 	}
 
-	protected Object saveDBObject(String collectionName, final DBObject dbDoc) {
-		
+	protected ObjectId saveDBObject(String collectionName, final DBObject dbDoc) {
+
 		if (dbDoc.keySet().isEmpty()) {
 			return null;
 		}
-		
-		return execute(new CollectionCallback<Object>() {
 
-			public Object doInCollection(DBCollection collection) throws MongoException, DataAccessException {
+		return execute(new CollectionCallback<ObjectId>() {
+			public ObjectId doInCollection(DBCollection collection) throws MongoException, DataAccessException {
 				collection.save(dbDoc);
-				return dbDoc.get(ID);
+				return (ObjectId) dbDoc.get(ID);
 			}
 		}, collectionName);
 	}
@@ -635,6 +634,18 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 		return executeEach(new FindCallback(query), null, new ReadDbObjectCallback<T>(reader, targetClass),
 				collectionName);
 	}
+	
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.data.document.mongodb.MongoOperations#find(java.lang.Class, java.lang.Object)
+	 */
+	public <T> T find(Class<T> targetClass, Object id) {
+		
+		ObjectId objectId = mongoConverter.convertObjectId(id);
+		List<T> result = query(QueryBuilder.start(MongoPropertyDescriptor.ID_KEY).is(objectId).get(), targetClass);
+		
+		return result.isEmpty() ? null : result.get(0);
+	}
 
 
 	public DB getDb() {
@@ -658,23 +669,24 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 		return dbo;
 	}
 
-	protected void populateIdIfNecessary(Object savedObject, Object id) {
-		//TODO Needs proper conversion support and should be integrated with reader implementation somehow
-		BeanWrapper bw = PropertyAccessorFactory.forBeanPropertyAccess(savedObject);
-		PropertyDescriptor idPd = BeanUtils.getPropertyDescriptor(savedObject.getClass(), "id");
-		if (idPd == null) {
-			idPd = BeanUtils.getPropertyDescriptor(savedObject.getClass(), ID);
+	/**
+	 * Populates the id property of the saved object, if it's not set already.
+	 * 
+	 * @param savedObject
+	 * @param id
+	 */
+	protected void populateIdIfNecessary(Object savedObject, ObjectId id) {
+
+		ConfigurablePropertyAccessor bw = PropertyAccessorFactory.forDirectFieldAccess(savedObject);
+		MongoPropertyDescriptor idDescriptor = new MongoPropertyDescriptors(savedObject.getClass()).getIdDescriptor();
+
+		if (idDescriptor == null) {
+			return;
 		}
-		if (idPd != null) {
-			Object v = bw.getPropertyValue(idPd.getName());
-			if (v == null) {
-				if (id instanceof ObjectId) {
-					bw.setPropertyValue(idPd.getName(), id.toString());
-				}
-				else if (id.getClass().isAssignableFrom(idPd.getPropertyType())) {
-					bw.setPropertyValue(idPd.getName(), id);
-				}
-			}
+
+		if (bw.getPropertyValue(idDescriptor.getName()) == null) {
+			Object target = this.mongoConverter.convertObjectId(id, idDescriptor.getPropertyType());
+			bw.setPropertyValue(idDescriptor.getName(), target);
 		}
 	}
 	
@@ -750,9 +762,8 @@ public class MongoTemplate implements InitializingBean, MongoOperations {
 			this.type = type;
 		}
 		
-		@SuppressWarnings("unchecked")
 		public T doWith(DBObject object) {
-			return (T) reader.read(type, object);
+			return reader.read(type, object);
 		}
 	}
 }
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/SimpleMongoConverter.java b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/SimpleMongoConverter.java
index 9eb736d90..516888f7e 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/SimpleMongoConverter.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/document/mongodb/SimpleMongoConverter.java
@@ -15,13 +15,12 @@
  */
 package org.springframework.data.document.mongodb;
 
-import java.beans.PropertyDescriptor;
 import java.lang.reflect.Array;
 import java.lang.reflect.GenericArrayType;
-import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.lang.reflect.TypeVariable;
+import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -39,23 +38,30 @@ import org.apache.commons.logging.LogFactory;
 import org.bson.types.CodeWScope;
 import org.bson.types.ObjectId;
 import org.springframework.beans.BeanUtils;
-import org.springframework.beans.BeanWrapper;
-import org.springframework.beans.PropertyAccessorFactory;
+import org.springframework.core.convert.ConversionFailedException;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.support.ConversionServiceFactory;
 import org.springframework.core.convert.support.GenericConversionService;
+import org.springframework.data.document.mongodb.MongoPropertyDescriptors.MongoPropertyDescriptor;
 import org.springframework.util.Assert;
+import org.springframework.util.comparator.CompoundComparator;
 
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
 import com.mongodb.DBRef;
 
+/**
+ * Basic {@link MongoConverter} implementation to convert between domain classes and {@link DBObject}s.
+ * 
+ * @author Mark Pollack
+ * @author Thomas Risberg
+ * @author Oliver Gierke
+ */
 public class SimpleMongoConverter implements MongoConverter {
 
-
-	/** Logger available to subclasses */
-	protected final Log logger = LogFactory.getLog(getClass());
-
-	public static final Set<String> SIMPLE_TYPES;
+	private static final Log LOG = LogFactory.getLog(SimpleMongoConverter.class);
+	private static final Set<String> SIMPLE_TYPES;
 
 	static {
 		Set<String> basics = new HashSet<String>();
@@ -102,142 +108,178 @@ public class SimpleMongoConverter implements MongoConverter {
 		basics.add(Pattern.class.getName());
 		basics.add(CodeWScope.class.getName());
 		basics.add(ObjectId.class.getName());
-		// TODO check on enums.. 
+		// TODO check on enums..
 		basics.add(Enum.class.getName());
 		SIMPLE_TYPES = Collections.unmodifiableSet(basics);
 	}
 
-	protected GenericConversionService conversionService = new GenericConversionService();
+	private final GenericConversionService conversionService;
 
+	/**
+	 * Creates a {@link SimpleMongoConverter}.
+	 */
 	public SimpleMongoConverter() {
+		this.conversionService = ConversionServiceFactory.createDefaultConversionService();
 		initializeConverters();
 	}
 
+	/**
+	 * Creates a new {@link SimpleMongoConverter} for the given {@link ConversionService}.
+	 * 
+	 * @param conversionService
+	 */
 	public SimpleMongoConverter(GenericConversionService conversionService) {
-		super();
+		Assert.notNull(conversionService);
 		this.conversionService = conversionService;
+		initializeConverters();
 	}
 
+	/**
+	 * Initializes additional converters that handle {@link ObjectId} conversion. Will register converters for supported
+	 * id types if none are registered for those conversion already. {@link GenericConversionService} is configured.
+	 */
 	protected void initializeConverters() {
-		conversionService.addConverter(ObjectIdToStringConverter.INSTANCE);
+		
+		if (!conversionService.canConvert(ObjectId.class, String.class)) {
+			conversionService.addConverter(ObjectIdToStringConverter.INSTANCE);
+			conversionService.addConverter(StringToObjectIdConverter.INSTANCE);
+		}
+		
+		if (!conversionService.canConvert(ObjectId.class, BigInteger.class)) {
+			conversionService.addConverter(ObjectIdToBigIntegerConverter.INSTANCE);
+			conversionService.addConverter(BigIntegerToIdConverter.INSTANCE);
+		}
 	}
 
 	/*
-	public ConversionContext getConversionContext() {
-		return conversionContext;
-	}
-
-	public void setConversionContext(ConversionContext conversionContext) {
-		this.conversionContext = conversionContext;
-	}
-
-	public void writeNew(Object obj, DBObject dbo) {
-		conversionContext.convertToDBObject(dbo, null, obj);
-	}*/
-
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.data.document.mongodb.MongoWriter#write(java.lang.Object, com.mongodb.DBObject)
+	 */
 	@SuppressWarnings("rawtypes")
 	public void write(Object obj, DBObject dbo) {
 
-		BeanWrapper bw = PropertyAccessorFactory.forBeanPropertyAccess(obj);
-		// This will leverage the conversion service.
-		initBeanWrapper(bw);
-
-		PropertyDescriptor[] propertyDescriptors = BeanUtils
-				.getPropertyDescriptors(obj.getClass());
-		for (PropertyDescriptor pd : propertyDescriptors) {
-			// if (isSimpleType(pd.getPropertyType())) {
-			Object value = bw.getPropertyValue(pd.getName());
-			String keyToUse = ("id".equals(pd.getName()) ? "_id" : pd.getName());
-			if (isValidProperty(pd)) {
+		MongoBeanWrapper beanWrapper = createWraper(obj, false);
+		for (MongoPropertyDescriptor descriptor : beanWrapper.getDescriptors()) {
+			if (descriptor.isMappable()) {
+				Object value = beanWrapper.getValue(descriptor);
+				
+				if(value == null) {
+					continue;
+				}
+				
+				String keyToUse = descriptor.getKeyToMap();
 				// TODO validate Enums...
-				if (value != null && Enum.class.isAssignableFrom(pd.getPropertyType())) {
-					writeValue(dbo, keyToUse, ((Enum)value).name());
-				}
-				else if (value != null && "_id".equals(keyToUse) && String.class.isAssignableFrom(pd.getPropertyType())) {
-					try {
-						ObjectId id =  new ObjectId((String)value);
-						writeValue(dbo, keyToUse, id);
-					}
-					catch (IllegalArgumentException iae) {
-						logger.debug("Unable to convert the String " + value
-								+ " to an ObjectId");
-						writeValue(dbo, keyToUse, value);
-					}
-				}
-				else {
+				if (descriptor.isEnum()) {
+					writeValue(dbo, keyToUse, ((Enum) value).name());
+				} else if (descriptor.isIdProperty() && descriptor.isOfIdType()) {
+
+						try {
+							writeValue(dbo, keyToUse, conversionService.convert(value, ObjectId.class));
+						} catch (ConversionFailedException iae) {
+							LOG.debug("Unable to convert the String " + value + " to an ObjectId");
+							writeValue(dbo, keyToUse, value);
+						}
+				} else {
 					writeValue(dbo, keyToUse, value);
 				}
-				// dbo.put(keyToUse, value);
 			} else {
-				if (!"class".equals(pd.getName())) {
-					logger.warn("Unable to map property " + pd.getName()
-						+ ".  Skipping.");
+				if (!"class".equals(descriptor.getName())) {
+					LOG.warn("Unable to map property " + descriptor.getName() + ".  Skipping.");
 				}
 			}
-			// }
-
 		}
-
 	}
 
+	/**
+	 * Writes the given value to the given {@link DBObject}. Will skip {@literal null} values.
+	 * 
+	 * @param dbo
+	 * @param keyToUse
+	 * @param value
+	 */
 	private void writeValue(DBObject dbo, String keyToUse, Object value) {
 
-		// is not asimple type.
-		if (value != null) {
-			if (!isSimpleType(value.getClass())) {
-				writeCompoundValue(dbo, keyToUse, value);
-			} else {
-				dbo.put(keyToUse, value);
-			}
+		if (!isSimpleType(value.getClass())) {
+			writeCompoundValue(dbo, keyToUse, value);
+		} else {
+			dbo.put(keyToUse, value);
 		}
-
 	}
 
+	/**
+	 * Writes the given {@link CompoundComparator} value to the given {@link DBObject}.
+	 * 
+	 * @param dbo
+	 * @param keyToUse
+	 * @param value
+	 */
 	@SuppressWarnings("unchecked")
 	private void writeCompoundValue(DBObject dbo, String keyToUse, Object value) {
 		if (value instanceof Map) {
-			writeMap(dbo, keyToUse, (Map<String, Object>)value);
+			writeMap(dbo, keyToUse, (Map<String, Object>) value);
 			return;
 		}
 		if (value instanceof Collection) {
 			// Should write a collection!
-			writeArray(dbo, keyToUse, ((Collection<Object>)value).toArray());
+			writeArray(dbo, keyToUse, ((Collection<Object>) value).toArray());
 			return;
 		}
 		if (value instanceof Object[]) {
-			// Should write a collection!
-			writeArray(dbo, keyToUse, (Object[])value);
+			// Should write an array!
+			writeArray(dbo, keyToUse, (Object[]) value);
 			return;
 		}
 		DBObject nestedDbo = new BasicDBObject();
 		write(value, nestedDbo);
 		dbo.put(keyToUse, nestedDbo);
-		
+
 	}
 
-	protected void writeMap(DBObject dbo, String keyToUse, Map<String, Object> map) {
-		//TODO support non-string based keys as long as there is a Spring Converter obj->string and (optionally) string->obj
+	/**
+	 * Writes the given {@link Map} to the given {@link DBObject}.
+	 * 
+	 * @param dbo
+	 * @param mapKey
+	 * @param map
+	 */
+	protected void writeMap(DBObject dbo, String mapKey, Map<String, Object> map) {
+		// TODO support non-string based keys as long as there is a Spring Converter obj->string and (optionally)
+		// string->obj
 		DBObject dboToPopulate = null;
-		if (keyToUse != null) {
+
+		// TODO - Does that make sense? If we create a new object here it's content will never make it out of this
+		// method
+		if (mapKey != null) {
 			dboToPopulate = new BasicDBObject();
 		} else {
 			dboToPopulate = dbo;
 		}
 		if (map != null) {
-			for (Map.Entry<String, Object> entry : map.entrySet()) {
+			for (Entry<String, Object> entry : map.entrySet()) {
+
 				Object entryValue = entry.getValue();
-				if (!isSimpleType(entryValue.getClass())) {			
-					writeCompoundValue(dboToPopulate, entry.getKey(), entryValue);
+				String entryKey = entry.getKey();
+
+				if (!isSimpleType(entryValue.getClass())) {
+					writeCompoundValue(dboToPopulate, entryKey, entryValue);
 				} else {
-					dboToPopulate.put(entry.getKey(), entryValue);
+					dboToPopulate.put(entryKey, entryValue);
 				}
-			}					
-			dbo.put(keyToUse, dboToPopulate);			
+			}
+			dbo.put(mapKey, dboToPopulate);
 		}
 	}
 
+	/**
+	 * Writes the given array to the given {@link DBObject}.
+	 * 
+	 * @param dbo
+	 * @param keyToUse
+	 * @param array
+	 */
 	protected void writeArray(DBObject dbo, String keyToUse, Object[] array) {
-		//TODO 
+		// TODO
 		Object[] dboValues;
 		if (array != null) {
 			dboValues = new Object[array.length];
@@ -251,142 +293,119 @@ public class SimpleMongoConverter implements MongoConverter {
 					dboValues[i] = o;
 				}
 				i++;
-			}					
-			dbo.put(keyToUse, dboValues);			
+			}
+			dbo.put(keyToUse, dboValues);
 		}
 	}
 
 	/*
-	public Object readNew(Class<? extends Object> clazz, DBObject dbo) {
-		return conversionContext.convertToObject(clazz, dbo);
-	}*/
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.data.document.mongodb.MongoReader#read(java.lang.Class, com.mongodb.DBObject)
+	 */
+	public <S> S read(Class<S> clazz, DBObject source) {
 
-	public Object read(Class<? extends Object> clazz, DBObject dbo) {
-		Assert.state(clazz != null, "Mapped class was not specified");
-		Object mappedObject = BeanUtils.instantiate(clazz);
-		BeanWrapper bw = PropertyAccessorFactory
-				.forBeanPropertyAccess(mappedObject);
-		initBeanWrapper(bw);
+		Assert.notNull(clazz, "Mapped class was not specified");
+		S target = BeanUtils.instantiateClass(clazz);
+		MongoBeanWrapper bw = new MongoBeanWrapper(target, conversionService, true);
 
-		// Iterate over properties of the object.b
-		// TODO iterate over the properties of DBObject and support nested property names with SpEL
-		// e.g. { "parameters.p1" : "1" , "count" : 5.0}
-		PropertyDescriptor[] propertyDescriptors = BeanUtils
-				.getPropertyDescriptors(clazz);
-		for (PropertyDescriptor pd : propertyDescriptors) {
-			
-			String keyToUse = ("id".equals(pd.getName()) ? "_id" : pd.getName());
-			if (dbo.containsField(keyToUse)) {
-				Object value = dbo.get(keyToUse);
-				if (value instanceof ObjectId) {
-					setObjectIdOnObject(bw, pd, (ObjectId) value);
-				} else {
-					if (isValidProperty(pd)) {
-						// This will leverage the conversion service.
-						if (!isSimpleType(value.getClass())) {
-							if (value instanceof DBObject) {
-								bw.setPropertyValue(pd.getName(), readCompoundValue(pd, (DBObject) value));
-							}
-							else if (value instanceof Object[]) {
-								Object[] values = new Object[((Object[])value).length];
-								int i = 0;
-								for (Object o : (Object[])value) {
-									if (o instanceof DBObject) {
-										Class<?> type;
-										if (pd.getPropertyType().isArray()) {
-											type = pd.getPropertyType().getComponentType();
-										}
-										else {
-											type = getGenericParameterClass(pd.getWriteMethod()).get(0);
-										}
-										values[i] = read(type, (DBObject)o);
+		for (MongoPropertyDescriptor descriptor : bw.getDescriptors()) {
+			String keyToUse = descriptor.getKeyToMap();
+			if (source.containsField(keyToUse)) {
+				if (descriptor.isMappable()) {
+					Object value = source.get(keyToUse);
+					if (!isSimpleType(value.getClass())) {
+						if (value instanceof DBObject) {
+							bw.setValue(descriptor, readCompoundValue(descriptor, (DBObject) value));
+						} else if (value instanceof Object[]) {
+							Object[] values = new Object[((Object[]) value).length];
+							int i = 0;
+							for (Object o : (Object[]) value) {
+								if (o instanceof DBObject) {
+									Class<?> type;
+									if (descriptor.getPropertyType().isArray()) {
+										type = descriptor.getPropertyType().getComponentType();
+									} else {
+										type = getGenericParameters(descriptor.getTypeToSet()).get(0);
 									}
-									else {
-										values[i] = o;
-									}
-									i++;
+									values[i] = read(type, (DBObject) o);
+								} else {
+									values[i] = o;
 								}
-								bw.setPropertyValue(pd.getName(), values);
+								i++;
 							}
-							else {
-								logger.warn("Unable to map compound DBObject field "
-										+ keyToUse + " to property " + pd.getName()
-										+ ".  The field value should have been a 'DBObject.class' but was " 
-										+ value.getClass().getName());
-							}
-						}
-						else {
-							bw.setPropertyValue(pd.getName(), value);
+							bw.setValue(descriptor, values);
+						} else {
+							LOG.warn("Unable to map compound DBObject field " + keyToUse + " to property "
+									+ descriptor.getName()
+									+ ".  The field value should have been a 'DBObject.class' but was "
+									+ value.getClass().getName());
 						}
 					} else {
-						logger.warn("Unable to map DBObject field "
-								+ keyToUse + " to property " + pd.getName()
-								+ ".  Skipping.");
+						bw.setValue(descriptor, value);
 					}
+				} else {
+					LOG.warn("Unable to map DBObject field " + keyToUse + " to property " + descriptor.getName()
+							+ ".  Skipping.");
 				}
 			}
 		}
 
-		return mappedObject;
+		return target;
 	}
-	
-	private Object readCompoundValue(PropertyDescriptor pd, DBObject dbo) {
-		Class<?> propertyClazz = pd.getPropertyType();
-		if (Map.class.isAssignableFrom(propertyClazz)) {
-			//TODO assure is assignable to BasicDBObject
-			return readMap(pd, (BasicDBObject)dbo, getGenericParameterClass(pd.getWriteMethod()).get(1) );			
+
+	/**
+	 * Reads a compund value from the given {@link DBObject} for the given property.
+	 * 
+	 * @param pd
+	 * @param dbo
+	 * @return
+	 */
+	private Object readCompoundValue(MongoPropertyDescriptors.MongoPropertyDescriptor pd, DBObject dbo) {
+
+		if (pd.isMap()) {
+			return readMap(pd, (BasicDBObject) dbo, getGenericParameters(pd.getTypeToSet()).get(1));
+		} else if (pd.isCollection()) {
+			throw new UnsupportedOperationException("Reading nested collections not supported yet!");
+		} else {
+			return read(pd.getPropertyType(), dbo);
 		}
-		if (Collection.class.isAssignableFrom(propertyClazz)) {
-			// Should read a collection!
-			return null;
-		}
-		return read(propertyClazz, dbo);
 	}
-	
+
+	/**
+	 * Create a {@link Map} instance. Will return a {@link HashMap} by default. Subclasses might want to override this
+	 * method to use a custom {@link Map} implementation.
+	 * 
+	 * @return
+	 */
 	protected Map<String, Object> createMap() {
 		return new HashMap<String, Object>();
 	}
-	
-	protected Map<?, ?> readMap(PropertyDescriptor pd, BasicDBObject dbo, Class<?> valueClazz) {
+
+	/**
+	 * Reads every key/value pair from the {@link DBObject} into a {@link Map} instance.
+	 * 
+	 * @param pd
+	 * @param dbo
+	 * @param targetType
+	 * @return
+	 */
+	protected Map<?, ?> readMap(MongoPropertyDescriptors.MongoPropertyDescriptor pd, BasicDBObject dbo, Class<?> targetType) {
 		Map<String, Object> map = createMap();
 		for (Entry<String, Object> entry : dbo.entrySet()) {
 			Object entryValue = entry.getValue();
-			BeanWrapper bw = PropertyAccessorFactory.forBeanPropertyAccess(entryValue);
-			initBeanWrapper(bw);
-			
 			if (!isSimpleType(entryValue.getClass())) {
-				map.put(entry.getKey(), read(valueClazz, (DBObject) entryValue));
-				//Can do some reflection tricks here -
-				//throw new RuntimeException("User types not supported yet as values for Maps");
+				map.put(entry.getKey(), read(targetType, (DBObject) entryValue));
+				// Can do some reflection tricks here -
+				// throw new RuntimeException("User types not supported yet as values for Maps");
 			} else {
-				map.put(entry.getKey(), entryValue );
+				map.put(entry.getKey(), conversionService.convert(entryValue, targetType));
 			}
 		}
 		return map;
 	}
 
-
-
-	protected void setObjectIdOnObject(BeanWrapper bw, PropertyDescriptor pd, ObjectId value) {
-		// TODO strategy for setting the id field. suggest looking for public
-		// property 'Id' or private field id or _id;
-		if (String.class.isAssignableFrom(pd.getPropertyType())) {
-			bw.setPropertyValue(pd.getName(), value);
-		}
-		else {
-			logger.warn("Unable to map _id field "
-					+ " to property " + pd.getName()
-					+ ".  Should have been a 'String' property but was " 
-					+ pd.getPropertyType().getName());
-		}
-	}
-
-	protected boolean isValidProperty(PropertyDescriptor descriptor) {
-		return (descriptor.getReadMethod() != null && 
-				descriptor.getWriteMethod() != null);
-	}
-
-	protected boolean isSimpleType(Class<?> propertyType) {
+	protected static boolean isSimpleType(Class<?> propertyType) {
 		if (propertyType == null) {
 			return false;
 		}
@@ -396,53 +415,63 @@ public class SimpleMongoConverter implements MongoConverter {
 		return SIMPLE_TYPES.contains(propertyType.getName());
 	}
 
-	protected void initBeanWrapper(BeanWrapper bw) {
-		bw.setConversionService(conversionService);
-	}
-	
-	
 	/**
-	 * TODO - should that be factored out into a utility class?
+	 * Callback to allow customizing creation of a {@link MongoBeanWrapper}.
 	 * 
-	 * @param setMethod
+	 * @param target the target object to wrap
+	 * @param fieldAccess whether to use field access or property access
 	 * @return
 	 */
-	public List<Class<?>> getGenericParameterClass(Method setMethod) {
-		List<Class<?>> actualGenericParameterTypes  = new ArrayList<Class<?>>();
-		Type[] genericParameterTypes = setMethod.getGenericParameterTypes();
+	protected MongoBeanWrapper createWraper(Object target, boolean fieldAccess) {
 
-		for(Type genericParameterType  : genericParameterTypes){		
-		    if(genericParameterType  instanceof ParameterizedType){
-		        ParameterizedType aType = (ParameterizedType) genericParameterType;
-		        Type[] parameterArgTypes = aType.getActualTypeArguments();		        
-		        for(Type parameterArgType : parameterArgTypes){
-		        	if (parameterArgType instanceof GenericArrayType)
-		            {
-		                Class<?> arrayType = (Class<?>) ((GenericArrayType) parameterArgType).getGenericComponentType();
-		                actualGenericParameterTypes.add(Array.newInstance(arrayType, 0).getClass());
-		            }
-		        	else {
-		        		if (parameterArgType instanceof ParameterizedType) {
-			        		ParameterizedType paramTypeArgs = (ParameterizedType) parameterArgType;
-			        		actualGenericParameterTypes.add((Class<?>)paramTypeArgs.getRawType());
-			        	} else {
-			        		 if (parameterArgType instanceof TypeVariable) {
-			        			 throw new RuntimeException("Can not map " + ((TypeVariable<?>) parameterArgType).getName());
-			        		 } else {
-			        			 if (parameterArgType instanceof Class) {
-			        				 actualGenericParameterTypes.add((Class<?>) parameterArgType);
-			        			 } else  {
-			        				 throw new RuntimeException("Can not map " + parameterArgType); 
-			        			 }
-			        		 }
-			        	}
-		        	}
-		        	
-		        }
-		    }
+		return new MongoBeanWrapper(target, conversionService, fieldAccess);
+	}
+
+	List<Class<?>> getGenericParameters(Type genericParameterType) {
+
+		List<Class<?>> actualGenericParameterTypes = new ArrayList<Class<?>>();
+
+		if (genericParameterType instanceof ParameterizedType) {
+			ParameterizedType aType = (ParameterizedType) genericParameterType;
+			Type[] parameterArgTypes = aType.getActualTypeArguments();
+			for (Type parameterArgType : parameterArgTypes) {
+				if (parameterArgType instanceof GenericArrayType) {
+					Class<?> arrayType = (Class<?>) ((GenericArrayType) parameterArgType).getGenericComponentType();
+					actualGenericParameterTypes.add(Array.newInstance(arrayType, 0).getClass());
+				} else {
+					if (parameterArgType instanceof ParameterizedType) {
+						ParameterizedType paramTypeArgs = (ParameterizedType) parameterArgType;
+						actualGenericParameterTypes.add((Class<?>) paramTypeArgs.getRawType());
+					} else {
+						if (parameterArgType instanceof TypeVariable) {
+							throw new RuntimeException("Can not map " + ((TypeVariable<?>) parameterArgType).getName());
+						} else {
+							if (parameterArgType instanceof Class) {
+								actualGenericParameterTypes.add((Class<?>) parameterArgType);
+							} else {
+								throw new RuntimeException("Can not map " + parameterArgType);
+							}
+						}
+					}
+				}
+			}
 		}
-		return actualGenericParameterTypes;
 
+		return actualGenericParameterTypes;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.document.mongodb.MongoConverter#convertObjectId(org.bson.types.ObjectId, java.lang.Class)
+	 */
+	public <T> T convertObjectId(ObjectId id, Class<T> targetType) {
+		return conversionService.convert(id, targetType);
+	}
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.data.document.mongodb.MongoConverter#convertObjectId(java.lang.Object)
+	 */
+	public ObjectId convertObjectId(Object id) {
+		return conversionService.convert(id, ObjectId.class);
 	}
 
 	/**
@@ -450,17 +479,46 @@ public class SimpleMongoConverter implements MongoConverter {
 	 * 
 	 * @author Oliver Gierke
 	 */
-	private static enum ObjectIdToStringConverter implements Converter<ObjectId, String> {
-
+	public static enum ObjectIdToStringConverter implements Converter<ObjectId, String> {
 		INSTANCE;
-
-		/*
-		 * (non-Javadoc)
-		 * 
-		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
-		 */
 		public String convert(ObjectId id) {
 			return id.toString();
 		}
 	}
+
+	/**
+	 * Simple singleton to convert {@link String}s to their {@link ObjectId} representation.
+	 * 
+	 * @author Oliver Gierke
+	 */
+	public static enum StringToObjectIdConverter implements Converter<String, ObjectId> {
+		INSTANCE;
+		public ObjectId convert(String source) {
+			return new ObjectId(source);
+		}
+	}
+
+	/**
+	 * Simple singleton to convert {@link ObjectId}s to their {@link BigInteger} representation.
+	 * 
+	 * @author Oliver Gierke
+	 */
+	public static enum ObjectIdToBigIntegerConverter implements Converter<ObjectId, BigInteger> {
+		INSTANCE;
+		public BigInteger convert(ObjectId source) {
+			return new BigInteger(source.toString(), 16);
+		}
+	}
+
+	/**
+	 * Simple singleton to convert {@link BigInteger}s to their {@link ObjectId} representation.
+	 * 
+	 * @author Oliver Gierke
+	 */
+	public static enum BigIntegerToIdConverter implements Converter<BigInteger, ObjectId> {
+		INSTANCE;
+		public ObjectId convert(BigInteger source) {
+			return new ObjectId(source.toString(16));
+		}
+	}
 }
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/MongoOperationsUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/MongoOperationsUnitTests.java
index c75f2e90a..3759cd31b 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/MongoOperationsUnitTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/MongoOperationsUnitTests.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.*;
 import java.util.Arrays;
 import java.util.List;
 
+import org.bson.types.ObjectId;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -57,13 +58,17 @@ public abstract class MongoOperationsUnitTests {
 		
 		converter = new MongoConverter() {
 			
-			public Object read(Class<? extends Object> clazz, DBObject dbo) {
-				return person;
-			}
-			
 			public void write(Object t, DBObject dbo) {
 				dbo.put("firstName", person.getFirstName());
 			}
+
+			public <S extends Object> S read(Class<S> clazz, DBObject dbo) {
+				return (S) person;
+			}
+			
+			public <T> T convertObjectId(ObjectId id, Class<T> targetType) {
+				return null;
+			}
 		};
 	}
 	
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/Person.java b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/Person.java
index 4be2deb24..387918e4c 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/Person.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/Person.java
@@ -15,20 +15,33 @@
  */
 package org.springframework.data.document.mongodb;
 
+import org.bson.types.ObjectId;
+
 public class Person {
 
+	private final ObjectId id;
+	
 	private String firstName;
 	
 	private Person friend;
 	
 	public Person() {
-		
+		this.id = new ObjectId();
+	}
+	
+	public Person(ObjectId id, String firstname) {
+		this.id = id;
+		this.firstName = firstname;
 	}
 	
 	public Person(String firstname) {
+		this();
 		this.firstName = firstname;
 	}
 
+	public ObjectId getId() {
+		return id;
+	}
 	
 	public String getFirstName() {
 		return firstName;
@@ -46,5 +59,36 @@ public class Person {
 		this.friend = friend;
 	}
 	
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see java.lang.Object#equals(java.lang.Object)
+	 */
+	@Override
+	public boolean equals(Object obj) {
+		if (obj == this) {
+			return true;
+		}
+
+		if (obj == null) {
+			return false;
+		}
+
+		if (!(getClass().equals(obj.getClass()))) {
+			return false;
+		}
+
+		Person that = (Person) obj;
+
+		return this.id == null ? false : this.id.equals(that.id);
+	}
 	
+	
+	/* (non-Javadoc)
+	 * @see java.lang.Object#hashCode()
+	 */
+	@Override
+	public int hashCode() {
+		return id.hashCode();
+	}
 }
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/SimpleMongoConverterTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/SimpleMongoConverterTests.java
index d7c040720..9ad81683e 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/SimpleMongoConverterTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/SimpleMongoConverterTests.java
@@ -15,10 +15,11 @@
  */
 package org.springframework.data.document.mongodb;
 
-import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.Field;
+import java.math.BigInteger;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -32,14 +33,20 @@ import org.springframework.util.ReflectionUtils;
 
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
+import com.mongodb.util.JSON;
 
 public class SimpleMongoConverterTests {
-	
+
+	static final String SIMPLE_JSON = "{ \"map\" : { \"foo\" : 3 , \"bar\" : 4}, \"number\" : 15 }";
+	static final String COMPLEX_JSON = "{ \"map\" : { \"trade\" : { \"orderType\" : \"BUY\" , \"price\" : 90.5 , \"quantity\" : 0 , \"ticker\" : \"VMW\"}}}";
+
 	SimpleMongoConverter converter;
-	
+	DBObject object;
+
 	@Before
 	public void setUp() {
 		converter = new SimpleMongoConverter();
+		object = new BasicDBObject();
 	}
 
 	@Test
@@ -47,12 +54,11 @@ public class SimpleMongoConverterTests {
 		User user = new User();
 		user.setAccountName("My Account");
 		user.setUserName("Mark");
-		DBObject dbo = new BasicDBObject();
-		converter.write(user, dbo);
-		assertEquals("My Account", dbo.get("accountName"));
-		assertEquals("Mark", dbo.get("userName"));
+		converter.write(user, object);
+		assertEquals("My Account", object.get("accountName"));
+		assertEquals("Mark", object.get("userName"));
 
-		User u = (User) converter.read(User.class, dbo);
+		User u = converter.read(User.class, object);
 
 		assertEquals("My Account", u.getAccountName());
 		assertEquals("Mark", u.getUserName());
@@ -61,14 +67,13 @@ public class SimpleMongoConverterTests {
 	@Test
 	public void nestedObject() {
 		Portfolio p = createPortfolioWithNoTrades();
-		DBObject dbo = new BasicDBObject();
-		converter.write(p, dbo);
-		
-		assertEquals("High Risk Trading Account", dbo.get("portfolioName"));
-		assertTrue(dbo.containsField("user"));
+		converter.write(p, object);
+
+		assertEquals("High Risk Trading Account", object.get("portfolioName"));
+		assertTrue(object.containsField("user"));
+
+		Portfolio cp = converter.read(Portfolio.class, object);
 
-		Portfolio cp = (Portfolio) converter.read(Portfolio.class, dbo);
-		
 		assertEquals("High Risk Trading Account", cp.getPortfolioName());
 		assertEquals("Joe Trader", cp.getUser().getUserName());
 		assertEquals("ACCT-123", cp.getUser().getAccountName());
@@ -78,20 +83,18 @@ public class SimpleMongoConverterTests {
 	@Test
 	public void objectWithMap() {
 		Portfolio p = createPortfolioWithPositions();
-		DBObject dbo = new BasicDBObject();
-		converter.write(p, dbo);
+		converter.write(p, object);
 
-		Portfolio cp = (Portfolio) converter.read(Portfolio.class, dbo);
+		Portfolio cp = converter.read(Portfolio.class, object);
 		assertEquals("High Risk Trading Account", cp.getPortfolioName());
 	}
 
 	@Test
 	public void objectWithMapContainingNonPrimitiveTypeAsValue() {
 		Portfolio p = createPortfolioWithManagers();
-		DBObject dbo = new BasicDBObject();
-		converter.write(p, dbo);
+		converter.write(p, object);
 
-		Portfolio cp = (Portfolio) converter.read(Portfolio.class, dbo);
+		Portfolio cp = converter.read(Portfolio.class, object);
 		assertEquals("High Risk Trading Account", cp.getPortfolioName());
 	}
 
@@ -130,10 +133,9 @@ public class SimpleMongoConverterTests {
 	@Test
 	public void objectWithArrayContainingNonPrimitiveType() {
 		TradeBatch b = createTradeBatch();
-		DBObject dbo = new BasicDBObject();
-		converter.write(b, dbo);
+		converter.write(b, object);
 
-		TradeBatch b2 = (TradeBatch) converter.read(TradeBatch.class, dbo);
+		TradeBatch b2 = converter.read(TradeBatch.class, object);
 		assertEquals(b.getBatchId(), b2.getBatchId());
 		assertNotNull(b2.getTradeList());
 		assertEquals(b.getTradeList().size(), b2.getTradeList().size());
@@ -155,8 +157,8 @@ public class SimpleMongoConverterTests {
 		t2.setTicker("MSFT");
 		t2.setQuantity(100);
 		t2.setPrice(27.92D);
-		tb.setTrades(new Trade[] {t2, t1});
-		tb.setTradeList(Arrays.asList(new Trade[] {t1, t2}));
+		tb.setTrades(new Trade[] { t2, t1 });
+		tb.setTradeList(Arrays.asList(new Trade[] { t1, t2 }));
 		return tb;
 	}
 
@@ -169,24 +171,180 @@ public class SimpleMongoConverterTests {
 		test.setNumberEnum(NumberEnum.FIVE);
 		DBObject dbo = new BasicDBObject();
 		converter.write(test, dbo);
-		
-		SomeEnumTest results = (SomeEnumTest) converter.read(SomeEnumTest.class, dbo);
+
+		SomeEnumTest results = converter.read(SomeEnumTest.class, dbo);
 		assertNotNull(results);
 		assertEquals(test.getId(), results.getId());
 		assertEquals(test.getName(), results.getName());
 		assertEquals(test.getStringEnum(), results.getStringEnum());
 		assertEquals(test.getNumberEnum(), results.getNumberEnum());
-}
+	}
 
 	@Test
-	public void testReflection() {
-		Method method = ReflectionUtils.findMethod(Portfolio.class, "setPortfolioManagers", Map.class);
-		assertNotNull(method);
-		List<Class<?>> paramClass = converter.getGenericParameterClass(method);
+	public void serializesClassWithFinalObjectIdCorrectly() throws Exception {
+
+		BasicDBObject object = new BasicDBObject();
+		Person person = new Person("Oliver");
+		converter.write(person, object);
+
+		assertThat(object.get("class"), is(nullValue()));
+		assertThat(object.get("_id"), is((Object) person.getId()));
+	}
+
+	@Test
+	public void discoversGenericsForType() throws Exception {
+
+		Field field = ReflectionUtils.findField(Sample.class, "map");
+		assertListOfStringAndLong(converter.getGenericParameters(field.getGenericType()));
+	}
+
+	@Test
+	public void writesSimpleMapCorrectly() throws Exception {
+
+		Map<String, Long> map = new HashMap<String, Long>();
+		map.put("foo", 1L);
+		map.put("bar", 2L);
+
+		Sample sample = new Sample();
+		sample.setMap(map);
+		sample.setNumber(15L);
+
+		converter.write(sample, object);
+
+		assertThat(object.get("number"), is((Object) 15L));
+
+		Object result = object.get("map");
+		assertTrue(result instanceof Map);
+
+		@SuppressWarnings("unchecked")
+		Map<String, Long> mapResult = (Map<String, Long>) result;
+		assertThat(mapResult.size(), is(2));
+		assertThat(mapResult.get("foo"), is(1L));
+		assertThat(mapResult.get("bar"), is(2L));
+	}
+
+	@Test
+	public void writesComplexMapCorrectly() throws Exception {
+
+		Trade trade = new Trade();
+		trade.setOrderType("BUY");
+		trade.setTicker("VMW");
+		trade.setPrice(90.50d);
+
+		Map<String, Trade> map = new HashMap<String, Trade>();
+		map.put("trade", trade);
+
+		converter.write(new Sample2(map), object);
+		DBObject tradeDbObject = new BasicDBObject();
+		converter.write(trade, tradeDbObject);
+
+		Object result = object.get("map");
+		assertTrue(result instanceof Map);
+
+		@SuppressWarnings("unchecked")
+		Map<String, DBObject> mapResult = (Map<String, DBObject>) result;
+		assertThat(mapResult.size(), is(1));
+		assertThat(mapResult.get("trade"), is(tradeDbObject));
+	}
+
+	@Test
+	public void readsMapWithSetterCorrectly() throws Exception {
+
+		DBObject input = (DBObject) JSON.parse(SIMPLE_JSON);
+		Sample result = converter.read(Sample.class, input);
+		assertThat(result.getNumber(), is(15L));
+
+		Map<String, Long> map = result.getMap();
+		assertThat(map, is(notNullValue()));
+		assertThat(map.size(), is(2));
+		assertThat(map.get("foo"), is(3L));
+		assertThat(map.get("bar"), is(4L));
+	}
+
+	@Test
+	public void readsMapWithFieldOnlyCorrectly() throws Exception {
+
+		DBObject input = (DBObject) JSON.parse(COMPLEX_JSON);
+		Sample2 result = converter.read(Sample2.class, input);
+
+		Map<String, Trade> map = result.getMap();
+
+		Trade trade = new Trade();
+		trade.setOrderType("BUY");
+		trade.setTicker("VMW");
+		trade.setPrice(90.50d);
+
+		assertThat(map.size(), is(1));
+		assertThat(map.get("trade").getTicker(), is("VMW"));
+		assertThat(map.get("trade").getOrderType(), is("BUY"));
+		assertThat(map.get("trade").getPrice(), is(90.50d));
+	}
+
+	@Test
+	public void supportsBigIntegerAsIdProperty() throws Exception {
 		
-		assertThat(paramClass.isEmpty(), is(false));
-		assertThat(paramClass.size(), is(2));
-		assertEquals(String.class, paramClass.get(0));
-		assertEquals(Person.class, paramClass.get(1));
+		Sample3 sample3 = new Sample3();
+		sample3.id = new BigInteger("4d24809660413b687f5d323e", 16);
+		converter.write(sample3, object);
+		assertThat(object.get("_id"), is(notNullValue()));
+
+		Sample3 result = converter.read(Sample3.class,
+				(DBObject) JSON.parse("{\"_id\" : {\"$oid\" : \"4d24809660413b687f5d323e\" }}"));
+		assertThat(result.getId().toString(16), is("4d24809660413b687f5d323e"));
+	}
+
+	private void assertListOfStringAndLong(List<Class<?>> types) {
+
+		assertThat(types.size(), is(2));
+		assertEquals(String.class, types.get(0));
+		assertEquals(Long.class, types.get(1));
+	}
+
+	public static class Sample {
+
+		private Map<String, Long> map;
+		private Long number;
+
+		public void setMap(Map<String, Long> map) {
+			this.map = map;
+		}
+
+		public Map<String, Long> getMap() {
+			return map;
+		}
+
+		public void setNumber(Long number) {
+			this.number = number;
+		}
+
+		public Long getNumber() {
+			return number;
+		}
+	}
+
+	public static class Sample2 {
+
+		private final Map<String, Trade> map;
+
+		protected Sample2() {
+			this.map = null;
+		}
+
+		public Sample2(Map<String, Trade> map) {
+			this.map = map;
+		}
+
+		public Map<String, Trade> getMap() {
+			return map;
+		}
+	}
+
+	private static class Sample3 {
+
+		private BigInteger id;
+
+		public BigInteger getId() {
+			return id;
+		}
 	}
 }
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/analytics/MvcAnalyticsTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/analytics/MvcAnalyticsTests.java
index 2aaae19a3..774fe8587 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/analytics/MvcAnalyticsTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/document/mongodb/analytics/MvcAnalyticsTests.java
@@ -58,7 +58,7 @@ public class MvcAnalyticsTests {
 		
 		mongoTemplate.getCollection("mvc", MvcEvent.class,
 				new MongoReader<MvcEvent>() {
-					public MvcEvent read(Class<? extends MvcEvent> clazz, DBObject dbo) {
+					public <S extends MvcEvent> S read(Class<S> clazz, DBObject dbo) {
 						return null;
 					}
 				});