public void doMarshal(final Object source, final HierarchicalStreamWriter writer, final MarshallingContext context) {
        writer.addAttribute(ATTRIBUTE_SERIALIZATION, ATTRIBUTE_VALUE_CUSTOM);

        // this is an array as it's a non final value that's accessed from an anonymous inner class.
        final Class[] currentType = new Class[1];
        final boolean[] writtenClassWrapper = {false};

        CustomObjectOutputStream.StreamCallback callback = new CustomObjectOutputStream.StreamCallback() {

            public void writeToStream(Object object) {
                if (object == null) {
                    writer.startNode(ELEMENT_NULL);
                    writer.endNode();
                } else {
                    ExtendedHierarchicalStreamWriterHelper.startNode(writer, mapper.serializedClass(object.getClass()), object.getClass());
                    context.convertAnother(object);
                    writer.endNode();
                }
            }

            public void writeFieldsToStream(Map fields) {
                ObjectStreamClass objectStreamClass = ObjectStreamClass.lookup(currentType[0]);

                writer.startNode(ELEMENT_DEFAULT);
                for (Iterator iterator = fields.keySet().iterator(); iterator.hasNext();) {
                    String name = (String) iterator.next();
                    if ( mapper.shouldSerializeMember( currentType[ 0 ], name ) ) {
	                    ObjectStreamField field = objectStreamClass.getField(name);
	                    Object value = fields.get(name);
	                    if (field == null) {
	                        throw new ObjectAccessException("Class " + value.getClass().getName()
	                                + " may not write a field named '" + name + "'");
	                    }
	                    if (value != null) {
	                        writer.startNode(mapper.serializedMember(currentType[0], name));
	                        if (field.getType() != value.getClass() && !field.getType().isPrimitive()) {
	                            writer.addAttribute(ATTRIBUTE_CLASS, mapper.serializedClass(value.getClass()));
	                        }
	                        context.convertAnother(value);
	                        writer.endNode();
	                    }
                    }
                }
                writer.endNode();
            }

            public void defaultWriteObject() {
                boolean writtenDefaultFields = false;

                ObjectStreamClass objectStreamClass = ObjectStreamClass.lookup(currentType[0]);

                if (objectStreamClass == null) {
                    return;
                }

                ObjectStreamField[] fields = objectStreamClass.getFields();
                for (int i = 0; i < fields.length; i++) {
                    ObjectStreamField field = fields[i];
                    final String fieldName = field.getName();
                    if ( mapper.shouldSerializeMember( currentType[ 0 ], fieldName ) ) {
	                    Object value = readField(field, currentType[0], source);
	                    if (value != null) {
	                        if (!writtenClassWrapper[0]) {
	                            writer.startNode(mapper.serializedClass(currentType[0]));
	                            writtenClassWrapper[0] = true;
	                        }
	                        if (!writtenDefaultFields) {
	                            writer.startNode(ELEMENT_DEFAULT);
	                            writtenDefaultFields = true;
	                        }
	
	                        writer.startNode(mapper.serializedMember(currentType[0], fieldName));
	
	                        Class actualType = value.getClass();
	                        Class defaultType = mapper.defaultImplementationOf(field.getType());
	                        if (!actualType.equals(defaultType)) {
	                            writer.addAttribute(ATTRIBUTE_CLASS, mapper.serializedClass(actualType));
	                        }
	
	                        context.convertAnother(value);
	
	                        writer.endNode();
	                    }
	                }
                }
                if (writtenClassWrapper[0] && !writtenDefaultFields) {
                    writer.startNode(ELEMENT_DEFAULT);
                    writer.endNode();
                } else if (writtenDefaultFields) {
                    writer.endNode();
                }
            }

            public void flush() {
                writer.flush();
            }

            public void close() {
                throw new UnsupportedOperationException("Objects are not allowed to call ObjectOutputStream.close() from writeObject()");
            }
        };

        try {
            boolean mustHandleUnserializableParent = false;
            Iterator classHieararchy = hierarchyFor(source.getClass()).iterator();
            while (classHieararchy.hasNext()) {
                currentType[0] = (Class) classHieararchy.next();
                if (!Serializable.class.isAssignableFrom(currentType[0])) {
                    mustHandleUnserializableParent = true;
                    continue;
                } else {
                    if (mustHandleUnserializableParent) {
                        marshalUnserializableParent(writer, context, source);
                        mustHandleUnserializableParent = false;
                    }
                    if (serializationMethodInvoker.supportsWriteObject(currentType[0], false)) {
                        writtenClassWrapper[0] = true;
                        writer.startNode(mapper.serializedClass(currentType[0]));
                        CustomObjectOutputStream objectOutputStream = CustomObjectOutputStream.getInstance(context, callback);
                        serializationMethodInvoker.callWriteObject(currentType[0], source, objectOutputStream);
                        objectOutputStream.popCallback();
                        writer.endNode();
                    } else if (serializationMethodInvoker.supportsReadObject(currentType[0], false)) {
                        // Special case for objects that have readObject(), but not writeObject().
                        // The class wrapper is always written, whether or not this class in the hierarchy has
                        // serializable fields. This guarantees that readObject() will be called upon deserialization.
                        writtenClassWrapper[0] = true;
                        writer.startNode(mapper.serializedClass(currentType[0]));
                        callback.defaultWriteObject();
                        writer.endNode();
                    } else {
                        writtenClassWrapper[0] = false;
                        callback.defaultWriteObject();
                        if (writtenClassWrapper[0]) {
                            writer.endNode();
                        }
                    }
                }
            }
        } catch (IOException e) {
            throw new ObjectAccessException("Could not call defaultWriteObject()", e);
        }
    }