JavaModule.java
/*
* Copyright (C) 2017 Red Hat, Inc.
*
* 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 io.atlasmap.java.module;
import java.lang.reflect.Array;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.atlasmap.api.AtlasException;
import io.atlasmap.api.AtlasValidationException;
import io.atlasmap.core.AtlasPath;
import io.atlasmap.core.AtlasUtil;
import io.atlasmap.core.BaseAtlasModule;
import io.atlasmap.java.core.JavaFieldReader;
import io.atlasmap.java.core.JavaFieldWriter;
import io.atlasmap.java.core.JavaFieldWriterUtil;
import io.atlasmap.java.core.TargetValueConverter;
import io.atlasmap.java.v2.AtlasJavaModelFactory;
import io.atlasmap.java.v2.JavaEnumField;
import io.atlasmap.java.v2.JavaField;
import io.atlasmap.spi.AtlasInternalSession;
import io.atlasmap.spi.AtlasModuleDetail;
import io.atlasmap.v2.AtlasModelFactory;
import io.atlasmap.v2.AuditStatus;
import io.atlasmap.v2.CollectionType;
import io.atlasmap.v2.Field;
import io.atlasmap.v2.FieldGroup;
import io.atlasmap.v2.FieldType;
import io.atlasmap.v2.Validation;
@AtlasModuleDetail(name = "JavaModule", uri = "atlas:java", modes = { "SOURCE", "TARGET" }, dataFormats = {
"java" }, configPackages = { "io.atlasmap.java.v2" })
public class JavaModule extends BaseAtlasModule {
public static final String DEFAULT_LIST_CLASS = "java.util.ArrayList";
private static final Logger LOG = LoggerFactory.getLogger(JavaModule.class);
private TargetValueConverter targetValueConverter = null;
private JavaFieldWriterUtil writerUtil = null;
public JavaModule() {
this.setAutomaticallyProcessOutputFieldActions(false);
}
@Override
public void init() {
writerUtil = new JavaFieldWriterUtil(getClassLoader(), getConversionService());
targetValueConverter = new TargetValueConverter(getClassLoader(), getConversionService(), writerUtil);
}
@Override
public void destroy() {
targetValueConverter = null;
writerUtil = null;
}
@Override
public void processPreValidation(AtlasInternalSession atlasSession) throws AtlasException {
if (atlasSession == null || atlasSession.getMapping() == null) {
LOG.error("Invalid session: Session and AtlasMapping must be specified");
throw new AtlasValidationException("Invalid session");
}
JavaValidationService javaValidator = new JavaValidationService(getConversionService(), getFieldActionService());
javaValidator.setMode(getMode());
javaValidator.setDocId(getDocId());
List<Validation> javaValidations = javaValidator.validateMapping(atlasSession.getMapping());
atlasSession.getValidations().getValidation().addAll(javaValidations);
if (LOG.isDebugEnabled()) {
LOG.debug("Detected " + javaValidations.size() + " java validation notices");
}
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processPreValidation completed", getDocId());
}
}
@Override
public void processPreSourceExecution(AtlasInternalSession atlasSession) throws AtlasException {
if (atlasSession == null || atlasSession.getMapping() == null || atlasSession.getMapping().getMappings() == null
|| atlasSession.getMapping().getMappings().getMapping() == null) {
throw new AtlasException("AtlasSession not properly intialized with a mapping that contains field mappings");
}
Object sourceDocument = atlasSession.getSourceDocument(getDocId());
if (sourceDocument == null) {
AtlasUtil.addAudit(atlasSession, getDocId(), String.format(
"Null source document: docId='%s'", getDocId()),
AuditStatus.WARN, null);
} else {
JavaFieldReader reader = new JavaFieldReader();
reader.setConversionService(getConversionService());
reader.setDocument(sourceDocument);
atlasSession.setFieldReader(getDocId(), reader);
}
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processPreSourceExcution completed", getDocId());
}
}
@Override
public void processPreTargetExecution(AtlasInternalSession atlasSession) throws AtlasException {
if (atlasSession == null || atlasSession.getMapping() == null || atlasSession.getMapping().getMappings() == null
|| atlasSession.getMapping().getMappings().getMapping() == null) {
throw new AtlasException("AtlasSession not properly intialized with a mapping that contains field mappings");
}
Object rootObject;
String targetClassName = AtlasUtil.unescapeFromUri(AtlasUtil.getUriParameterValue(getUri(), "className"));
String collectionTypeStr = AtlasUtil.unescapeFromUri(AtlasUtil.getUriParameterValue(getUri(), "collectionType"));
CollectionType collectionType = collectionTypeStr != null ? CollectionType.fromValue(collectionTypeStr) : CollectionType.NONE;
String collectionClassName = AtlasUtil.unescapeFromUri(AtlasUtil.getUriParameterValue(getUri(), "collectionClassName"));
JavaFieldWriter writer = new JavaFieldWriter(this.writerUtil);
Class<?> clazz = writerUtil.loadClass(targetClassName);
if (collectionType == CollectionType.ARRAY) {
rootObject = Array.newInstance(clazz, 0);
} else if (collectionType != CollectionType.NONE) {
if (collectionClassName != null) {
rootObject = writerUtil.instantiateObject(writerUtil.loadClass(collectionClassName));
} else {
rootObject = writerUtil.instantiateObject(writerUtil.getDefaultCollectionImplClass(collectionType));
}
writer.setCollectionItemClass(clazz);
} else {
rootObject = writerUtil.instantiateObject(clazz);
}
writer.setRootObject(rootObject);
writer.setTargetValueConverter(targetValueConverter);
atlasSession.setFieldWriter(getDocId(), writer);
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processPreTargetExcution completed", getDocId());
}
}
@Override
public void readSourceValue(AtlasInternalSession session) throws AtlasException {
Field sourceField = session.head().getSourceField();
JavaFieldReader reader = session.getFieldReader(getDocId(), JavaFieldReader.class);
if (reader == null) {
AtlasUtil.addAudit(session, sourceField, String.format(
"Source document '%s' doesn't exist", getDocId()),
AuditStatus.ERROR, null);
return;
}
reader.read(session);
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processSourceFieldMapping completed: SourceField:[docId={}, path={}, type={}, value={}]",
getDocId(), sourceField.getDocId(), sourceField.getPath(), sourceField.getFieldType(), sourceField.getValue());
}
}
@Override
public void populateTargetField(AtlasInternalSession session) throws AtlasException {
Field sourceField = session.head().getSourceField();
Field targetField = session.head().getTargetField();
AtlasPath path = new AtlasPath(targetField.getPath());
FieldGroup targetFieldGroup = null;
if (path.hasCollection() && !path.isIndexedCollection()) {
targetFieldGroup = AtlasModelFactory.createFieldGroupFrom(targetField, true);
session.head().setTargetField(targetFieldGroup);
}
JavaFieldWriter writer = session.getFieldWriter(getDocId(), JavaFieldWriter.class);
if (targetFieldGroup == null) {
if (sourceField instanceof FieldGroup) {
List<Field> subFields = ((FieldGroup)sourceField).getField();
if (subFields == null || subFields.size() == 0) {
return;
}
Integer index = targetField.getIndex();
if (index != null) {
if (subFields.size() > index) {
sourceField = subFields.get(index);
} else {
AtlasUtil.addAudit(session, getDocId(), String.format(
"The number of source fields (%s) is smaller than target index (%s) - ignoring",
subFields.size(), index),
AuditStatus.WARN, null);
return;
}
} else {
// The last one wins for compatibility
sourceField = subFields.get(subFields.size() - 1);
}
session.head().setSourceField(sourceField);
}
/* Lazy parent instantiation, unless specific mapping defined for a complex type (Example json -> java)
Only instantiate the parent if there is a child value to avoid null src class -> empty dst class mapping
This will ensure null src class maps to null destination class
*/
Object parentObject = null;
if (null != sourceField.getValue() || (null == sourceField.getValue() && targetField.getFieldType() == FieldType.COMPLEX && !(targetField instanceof JavaEnumField))) {
parentObject = writer.prepareParentObject(session);
}
if (parentObject != null) {
writer.populateTargetFieldValue(session, parentObject);
writer.enqueueFieldAndParent(targetField, parentObject);
}
} else if (sourceField instanceof FieldGroup) {
if (((FieldGroup)sourceField).getField().size() == 0) {
Object parentObject = writer.prepareParentObject(session);
if (parentObject != null) {
writer.enqueueFieldAndParent(targetFieldGroup, parentObject);
}
}
Field previousTargetSubField = null;
for (int i=0; i<((FieldGroup)sourceField).getField().size(); i++) {
Field sourceSubField = ((FieldGroup)sourceField).getField().get(i);
Field targetSubField = targetField instanceof JavaEnumField ? new JavaEnumField() : new JavaField();
AtlasJavaModelFactory.copyField(targetField, targetSubField, false);
getCollectionHelper().copyCollectionIndexes(sourceField, sourceSubField, targetSubField, previousTargetSubField);
previousTargetSubField = targetSubField;
targetFieldGroup.getField().add(targetSubField);
session.head().setSourceField(sourceSubField);
session.head().setTargetField(targetSubField);
Object parentObject = writer.prepareParentObject(session);
if (parentObject != null) {
writer.populateTargetFieldValue(session, parentObject);
writer.enqueueFieldAndParent(targetSubField, parentObject);
}
}
session.head().setSourceField(sourceField);
session.head().setTargetField(targetFieldGroup);
} else {
Field targetSubField = targetField instanceof JavaEnumField ? new JavaEnumField() : new JavaField();
AtlasJavaModelFactory.copyField(targetField, targetSubField, false);
path.setVacantCollectionIndex(0);
targetSubField.setPath(path.toString());
targetFieldGroup.getField().add(targetSubField);
session.head().setTargetField(targetSubField);
Object parentObject = writer.prepareParentObject(session);
if (parentObject != null) {
writer.populateTargetFieldValue(session, parentObject);
writer.enqueueFieldAndParent(targetSubField, parentObject);
}
session.head().setTargetField(targetFieldGroup);
}
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processTargetFieldMapping completed: SourceField:[docId={}, path={}, type={}, value={}], TargetField:[docId={}, path={}, type={}, value={}]",
getDocId(), sourceField.getDocId(), sourceField.getPath(), sourceField.getFieldType(), sourceField.getValue(),
targetField.getDocId(), targetField.getPath(), targetField.getFieldType(), targetField.getValue());
}
}
@Override
public void writeTargetValue(AtlasInternalSession session) throws AtlasException {
JavaFieldWriter writer = session.getFieldWriter(getDocId(), JavaFieldWriter.class);
writer.commitWriting(session);
}
@Override
public void processPostSourceExecution(AtlasInternalSession session) throws AtlasException {
session.removeFieldReader(getDocId());
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processPostSourceExecution completed", getDocId());
}
}
@Override
public void processPostTargetExecution(AtlasInternalSession session) throws AtlasException {
JavaFieldWriter writer = session.getFieldWriter(getDocId(), JavaFieldWriter.class);
if (writer != null && writer.getRootObject() != null) {
session.setTargetDocument(getDocId(), writer.getRootObject());
} else {
AtlasUtil.addAudit(session, getDocId(),
String.format("No target document created for DataSource '%s'", getDocId()),
AuditStatus.WARN, null);
}
session.removeFieldWriter(getDocId());
if (LOG.isDebugEnabled()) {
LOG.debug("{}: processPostTargetExecution completed", getDocId());
}
}
@Override
public Boolean isSupportedField(Field field) {
if (super.isSupportedField(field)) {
return true;
}
return field instanceof JavaField || field instanceof JavaEnumField;
}
@Override
public Field cloneField(Field field) throws AtlasException {
return AtlasJavaModelFactory.cloneJavaField(field, true);
}
@Override
public JavaField createField() {
return AtlasJavaModelFactory.createJavaField();
}
}