path: root/sandbox/kgoodson/mappingFramework/sdo-snapshot/src/main/java/com/agfa/hap/sdo/implementation/SnapshotImplementation.java

/**
 *
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you 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 com.agfa.hap.sdo.implementation;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.StreamCorruptedException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.agfa.hap.sdo.DataMapper;
import com.agfa.hap.sdo.ObjectPropertyVisitor;
import com.agfa.hap.sdo.Property;
import com.agfa.hap.sdo.Snapshot;
import com.agfa.hap.sdo.SnapshotDefinition;
import com.agfa.hap.sdo.Type;
import com.agfa.hap.sdo.helper.TypeHelper;
import com.agfa.hap.sdo.impl.TypeConverter;

public class SnapshotImplementation implements Snapshot {

	private static Logger logger = Logger.getLogger(SnapshotImplementation.class);
	private static final long serialVersionUID = 5403046473375751795L;

	private transient byte[] serializedContent;
	
	//workaround for http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6554519
	private void writeObject(java.io.ObjectOutputStream out) throws IOException {
		out.writeInt(serializedContent.length);
		out.write(serializedContent);
	}
	
//	workaround for http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6554519
	private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
		int size = in.readInt();
		serializedContent = new byte[size];
		int index = 0;
		int remaining = size;
		int read = 0;
		while (index < size && remaining > 0){
			read = in.read(serializedContent, index, remaining);
			index += read;
			remaining -= read;
		}
		if (remaining != 0){
			throw new StreamCorruptedException("read=" + read + " index=" + index + 
					" remaining=" + remaining + " size=" + size + " serializedcontent=" + serializedContent);
		}
	}
	
 	public <T> SnapshotImplementation(DataMapper<T> mapper, SnapshotDefinition definition, Collection<? extends T> roots) {
		buildSerializedContent(mapper, definition, roots);
	}
	
	public <T> List<T> extract(DataMapper<T> mapper) {
		return extractRoots(mapper);		
	}	

	private <T> List<T> extractRoots(DataMapper<T> mapper) {
		ByteArrayInputStream bis = new ByteArrayInputStream(serializedContent);
		DataInputStream dis = new DataInputStream(bis);
		SnapshotSerializer<T> serializer = new SnapshotSerializer<T>(mapper);
		try {
			return serializer.read(dis);
		} catch (IOException e) {
			throw new SdoImplementationException(e);
		}		
	}
	
	private <T> void buildSerializedContent(DataMapper<T> mapper, SnapshotDefinition definition, Collection<? extends T> targets) {
		ByteArrayOutputStream bos = new ByteArrayOutputStream();
		DataOutputStream out = new DataOutputStream(bos);
		SnapshotSerializer<T> serializer = new SnapshotSerializer<T>(mapper);
		try {
			serializer.write(definition, targets, out);
			out.close();
		} catch (Exception e) {
			throw new SdoImplementationException("definition=" + definition + ":targets=" + targets, e);
		}
		serializedContent = bos.toByteArray();
	}

	static class SnapshotSerializer<T> implements ObjectPropertyVisitor {
		 //protected visibility for testing purposes
		public static final int MAX_UTF_LENGTH = 65535;
		private DataOutput out;
		private DataMapper<T> mapper;
		private List<Object> serializedObjects;
		private Map<Property, BulkProperty<T>> bulkProperties;

		public SnapshotSerializer(DataMapper<T> mapper) {
			this.mapper = mapper;
		}
		
		protected T create(Type type) {
			T result = mapper.create(type);    	
			serializedObjects.add(result);
			return result;
		}

		public List<T> read(DataInput in) throws IOException {
			this.serializedObjects = new ArrayList<Object>();
			return readFromStream(in);
		}
		
		private List<T> readFromStream(DataInput in) throws IOException {
			int nrOfObjects = in.readInt();
			if (nrOfObjects < 1){
				return new ArrayList<T>();
			}
			List<T> result = basicReadFromStream(in, nrOfObjects);
			readBulkProperties(in);
			return result;
		}

		private List<T> basicReadFromStream(DataInput in, int nrOfObjects) throws IOException {
			List<T> result = new ArrayList<T>(nrOfObjects);
			for (int i = 0; i < nrOfObjects; i++){
				result.add(readObject(in));
			}
			return result;
		}

		private T readObject(DataInput in) throws IOException {
			Type type = TypeHelper.INSTANCE.getType(in.readUTF(), in.readUTF());			
			T instance = create(type);
			int propIndex = in.readInt();
			while (propIndex != Integer.MIN_VALUE) {
				if (propIndex < 0) {
					Property property = type.getProperty(-propIndex-1);
					if (property.isMany()) {
						mapper.setUnavailable(instance, property);
					} else {
						Object identity = read(in, null, property.getType().getIdentityProperty());
						mapper.setProperty(instance, property, mapper.newProxy(property.getType(), identity));
					}
				} else {
					Property prop = type.getProperty(propIndex);
					Object value = read(in, instance, prop);
					mapper.setProperty(instance, prop, value);
				}
				propIndex = in.readInt();
			}
			return instance;
		}

		private Object read(DataInput in, Object parent, Property prop) throws IOException {
			if (prop.getType().isDataType()) {
				String value = this.readPotentiallyLongUTF(in);
				try {
					return TypeConverter.get(prop.getType()).parse(value);
				} catch (RuntimeException e){
					logger.error("unable to parse " + value + "for property " + prop + " parent=" + parent, e);
					throw e;
				}
			} else {
				return readChildObject(in, prop, parent);
			}

		}
		
		String readPotentiallyLongUTF(DataInput in) throws IOException {
			//package visibility for testing purposes
			StringBuilder builder = new StringBuilder();
			while (true){
				String buf = in.readUTF();
				builder.append(buf);
				if (buf.length() < MAX_UTF_LENGTH){
					break;
				}
				String finalOrContinue = in.readUTF();
				if ("F".equals(finalOrContinue)){
					break;
				}
			}									
			return builder.toString();
		}

		protected Object readChildObject(DataInput in, Property prop, Object parent) throws IOException {
			int index = in.readInt();
			if (index < 0) {
				return readObject(in);
			} else {
				return serializedObjects.get(index);
			}
		}

		public void write(SnapshotDefinition def, Collection<? extends T> roots, DataOutput out) throws Exception {
			this.out = out;
			this.serializedObjects = new ArrayList<Object>();
			out.writeInt(roots.size());
			for (T root : roots){
				def.visit(mapper, this, root);
			}
			writeBulkProperties(out);
		}

		protected void writeBulkProperties(DataOutput out) throws Exception {
			if (bulkProperties != null) {
				while (!bulkProperties.isEmpty()){ 
					//we can't just simply iterate over the bulkproperties.values, because while writing a bulkproperty we might visit an additional bulk property in the subselect
					BulkProperty bp = bulkProperties.values().iterator().next();
					BulkProperty emptyBulkProperty = new BulkProperty(bp.property,  bp.subselect);
					bulkProperties.put(bp.property, emptyBulkProperty); //replace the bulk property that we are going to write with a copy of itself, but withouth instances
					writeBulkProperty(bp);
					if (emptyBulkProperty.instances.isEmpty()){
						bulkProperties.remove(bp.property); //if no new instances were added (through some nested bulkproperties) than we can remove the bulkproperty from the map of bulk properties to be processed
					}
				}
			}			
			out.writeInt(-1);
		}
		
		protected void writeBulkProperty(BulkProperty<T> bp) throws Exception {
			Collection<T> values = this.mapper.getProperties(bp.instances, bp.property, bp.subselect);
			Iterator<T> valuesIt = values.iterator();
			Iterator<?> parents = bp.instances.iterator();
			while (parents.hasNext()) {
				Object parent = parents.next();
				T value = valuesIt.next();				
				if (bp.property.isMany()) {
					Collection<T> children = (Collection<T>) value;
					for (T child : children) {
						writeBulkPropertyValue(bp, parent, child);
					}
				} else {
					writeBulkPropertyValue(bp, parent, value);
				}
			}
		}

		protected void writeBulkPropertyValue(BulkProperty bp, Object parent, T value) throws IOException, Exception {
			out.writeInt(serializedObjects.indexOf(parent));
			if (this.visitProperty(parent, bp.property, value)) {
				bp.subselect.visit(mapper, this, value);					
			}
		}

		protected void readBulkProperties(DataInput in) throws IOException {
			while (readBulkProperty(in)) {
			}
		}
		
		protected boolean readBulkProperty(DataInput in) throws IOException {
			int index = in.readInt();
			if (index < 0) {
				return false;
			}
			T parent = (T) serializedObjects.get(index);
			index = in.readInt();
			if (index < 0){ //null value
				Property property = mapper.getType(parent).getProperty(-index-1);
				mapper.setProperty(parent, property, null);
				return true;
			}
			Property property = mapper.getType(parent).getProperty(index);
			Object value = read(in, parent, property);
			mapper.setProperty(parent, property, value);
			return true;
		}

		public void startDataObject(Object dataObject, Type type) throws IOException {
			serializedObjects.add(dataObject);
			out.writeUTF(type.getURI());
			out.writeUTF(type.getName());
		}

		public boolean visitProperty(Object dataObject, Property property, Object value) throws IOException {
			if (value == null) {
				out.writeInt(-property.getIndex()-1);
				return false;
			}
			out.writeInt(property.getIndex());
			if (property.getType().isDataType()) {
				writePotentiallyLongStringUTF(TypeConverter.get(property.getType()).toString(value), out);
				return false;
			} else {
				int index = serializedObjects.indexOf(value);
				out.writeInt(index);
				return index < 0;
			}
		}	
		
		void writePotentiallyLongStringUTF(String stringValue, DataOutput out) throws IOException {
			 //package visibilty for testing purposes
			int totalLength = stringValue.length();
			if (totalLength < MAX_UTF_LENGTH){
				out.writeUTF(stringValue);
				return;
			}
			int lastPossibleMaxSubstringBeginIndex = totalLength - MAX_UTF_LENGTH;
			int substringBeginIndex = 0;
			for (; substringBeginIndex <= lastPossibleMaxSubstringBeginIndex; substringBeginIndex += MAX_UTF_LENGTH ){
				out.writeUTF(stringValue.substring(substringBeginIndex, substringBeginIndex + MAX_UTF_LENGTH));
				if (substringBeginIndex == lastPossibleMaxSubstringBeginIndex){
					out.writeUTF("F"); //final 
				} else {
					out.writeUTF("C"); //continue
				}
			}
			if (substringBeginIndex != totalLength -1){
				out.writeUTF(stringValue.substring(substringBeginIndex, totalLength))	;
			}								
		}	

		public void visitProxyProperty(Object instance, Property property, Object identity) throws Exception {
			out.writeInt(-property.getIndex()-1);
			if (identity != null) {
				out.writeUTF(TypeConverter.get(property.getType().getIdentityProperty().getType()).toString(identity));
			}
		}

		public void endDataObject(Object dataObject, Type type) throws IOException {
			out.writeInt(Integer.MIN_VALUE);
		}

		public void visitBulkProperty(Object instance, Property property, SnapshotDefinition subselect) {
			getInstancesForBulkProperty(property, subselect).add((T)instance);
		}
		
		protected List<T> getInstancesForBulkProperty(Property property, SnapshotDefinition subselect) {
			if (bulkProperties == null) {
				bulkProperties = new HashMap<Property, BulkProperty<T>>();
			}
			BulkProperty<T> bp = bulkProperties.get(property);
			if (bp == null) {
				bp = new BulkProperty<T>(property, subselect);
				bulkProperties.put(property, bp);
			} else {
				if (!bp.subselect.equals(subselect)) {
					throw new IllegalArgumentException("Different subselects not yet supported!");
				}
			}
			return bp.instances;
		}
		
	}

	static class BulkProperty<T> {
		BulkProperty(Property property, SnapshotDefinition subselect) {
			this.property = property;
			this.subselect = subselect;
			this.instances = new ArrayList<T>();
		}
		Property property;
		List<T> instances;
		SnapshotDefinition subselect;		
	}
}