001/*
002 * (C) Copyright 2018 Nuxeo (http://nuxeo.com/) and others.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 *
016 * Contributors:
017 *     pierre
018 */
019package org.nuxeo.ecm.core.io.avro;
020
021import org.apache.avro.LogicalType;
022import org.apache.avro.LogicalTypes;
023import org.apache.avro.Schema;
024import org.apache.commons.lang3.NotImplementedException;
025import org.nuxeo.ecm.core.schema.types.ComplexType;
026import org.nuxeo.ecm.core.schema.types.ListType;
027import org.nuxeo.ecm.core.schema.types.Type;
028import org.nuxeo.ecm.core.schema.types.primitives.BinaryType;
029import org.nuxeo.ecm.core.schema.types.primitives.BooleanType;
030import org.nuxeo.ecm.core.schema.types.primitives.DateType;
031import org.nuxeo.ecm.core.schema.types.primitives.DoubleType;
032import org.nuxeo.ecm.core.schema.types.primitives.IntegerType;
033import org.nuxeo.ecm.core.schema.types.primitives.LongType;
034import org.nuxeo.ecm.core.schema.types.primitives.StringType;
035import org.nuxeo.runtime.avro.AvroSchemaFactory;
036import org.nuxeo.runtime.avro.AvroSchemaFactoryContext;
037
038/**
039 * @since 10.2
040 */
041public class TypeSchemaFactory extends AvroSchemaFactory<Type> {
042
043    protected ComplexTypeSchemaFactory complexTypeFactory;
044
045    public TypeSchemaFactory(AvroSchemaFactoryContext context) {
046        super(context);
047        complexTypeFactory = new ComplexTypeSchemaFactory(context);
048    }
049
050    @Override
051    public Schema createSchema(Type input) {
052        if (input.isSimpleType()) {
053            if (input == IntegerType.INSTANCE) {
054                return Schema.create(org.apache.avro.Schema.Type.INT);
055            } else if (input == LongType.INSTANCE) {
056                return Schema.create(org.apache.avro.Schema.Type.LONG);
057            } else if (input == DoubleType.INSTANCE) {
058                return Schema.create(org.apache.avro.Schema.Type.DOUBLE);
059            } else if (input == StringType.INSTANCE) {
060                return Schema.create(org.apache.avro.Schema.Type.STRING);
061            } else if (input == BooleanType.INSTANCE) {
062                return Schema.create(org.apache.avro.Schema.Type.BOOLEAN);
063            } else if (input == BinaryType.INSTANCE) {
064                return Schema.create(org.apache.avro.Schema.Type.BYTES);
065            } else if (input == DateType.INSTANCE) {
066                return LogicalTypes.timestampMillis().addToSchema(Schema.create(org.apache.avro.Schema.Type.LONG));
067            } else if (input.getSuperType() != null && input.getSuperType().isSimpleType()) {
068                return new LogicalType(getName(input)).addToSchema(createSchema(input.getSuperType()));
069            }
070        } else if (input.isListType()) {
071            Schema array = Schema.createArray(context.createSchema(((ListType) input).getFieldType()));
072            String logicalType = ((ListType) input).isArray() ? "array" : "list";
073            new LogicalType(logicalType).addToSchema(array);
074            return array;
075        } else if (input.isComplexType()) {
076            return complexTypeFactory.createSchema((ComplexType) input);
077        } else if (input.isCompositeType()) {
078            throw new NotImplementedException("Composite types are not supported yet");
079        }
080        throw new NotImplementedException("Cannot create Avro type schema for nuxeo type: " + input);
081    }
082
083    @Override
084    public String getName(Type input) {
085        return context.getService().encodeName(input.getName());
086    }
087
088}