// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DateFieldConverter.java
// public class DateFieldConverter extends SinkFieldConverter {
// 
//     public DateFieldConverter() {
//         super(Date.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DecimalFieldConverter.java
// public class DecimalFieldConverter extends SinkFieldConverter {
// 
//     public enum Format {
//         DECIMAL128,         //needs MongoDB v3.4+
//         LEGACYDOUBLE        //results in double approximation
//     }
// 
//     private Format format;
// 
//     public DecimalFieldConverter() {
//         super(Decimal.schema(0));
//         this.format = Format.DECIMAL128;
//     }
// 
//     public DecimalFieldConverter(Format format) {
//         super(Decimal.schema(0));
//         this.format = format;
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
// 
//         if(data instanceof BigDecimal) {
//             if(format.equals(Format.DECIMAL128))
//                 return new BsonDecimal128(new Decimal128((BigDecimal)data));
// 
//             if(format.equals(Format.LEGACYDOUBLE))
//                 return new BsonDouble(((BigDecimal)data).doubleValue());
//         }
// 
//         throw new DataException("error: decimal conversion not possible when data is"
//                         + " of type "+data.getClass().getName() + " and format is "+format);
// 
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimeFieldConverter.java
// public class TimeFieldConverter extends SinkFieldConverter {
// 
//     public TimeFieldConverter() {
//         super(Time.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimestampFieldConverter.java
// public class TimestampFieldConverter extends SinkFieldConverter {
// 
//     public TimestampFieldConverter() {
//         super(Timestamp.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }


import org.slf4j.LoggerFactory;
import java.util.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DateFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DecimalFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimeFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimestampFieldConverter;
import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.*;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.DataException;
import org.bson.BsonArray;
import org.bson.BsonDocument;
import org.bson.BsonNull;
import org.bson.BsonValue;
import org.slf4j.Logger;
/*
 * Copyright (c) 2017. Hans-Peter Grahsl (grahslhp@gmail.com)
 *
 * 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 at.grahsl.kafka.connect.mongodb.converter;



//looks like Avro and JSON + Schema is convertible by means of
//a unified conversion approach since they are using the
//same the Struct/Type information ...
public class AvroJsonSchemafulRecordConverter implements RecordConverter {

    public static final Set<String> LOGICAL_TYPE_NAMES = new HashSet<>(
            Arrays.asList(Date.LOGICAL_NAME, Decimal.LOGICAL_NAME,
                            Time.LOGICAL_NAME, Timestamp.LOGICAL_NAME)
        );

    private final Map<Schema.Type, SinkFieldConverter> converters = new HashMap<>();
    private final Map<String, SinkFieldConverter> logicalConverters = new HashMap<>();

    private static Logger logger = LoggerFactory.getLogger(AvroJsonSchemafulRecordConverter.class);

    public AvroJsonSchemafulRecordConverter() {

        //standard types
        registerSinkFieldConverter(new BooleanFieldConverter());
        registerSinkFieldConverter(new Int8FieldConverter());
        registerSinkFieldConverter(new Int16FieldConverter());
        registerSinkFieldConverter(new Int32FieldConverter());
        registerSinkFieldConverter(new Int64FieldConverter());
        registerSinkFieldConverter(new Float32FieldConverter());
        registerSinkFieldConverter(new Float64FieldConverter());
        registerSinkFieldConverter(new StringFieldConverter());
        registerSinkFieldConverter(new BytesFieldConverter());

        //logical types
        registerSinkFieldLogicalConverter(new DateFieldConverter());
        registerSinkFieldLogicalConverter(new TimeFieldConverter());
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DateFieldConverter.java
// public class DateFieldConverter extends SinkFieldConverter {
// 
//     public DateFieldConverter() {
//         super(Date.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DecimalFieldConverter.java
// public class DecimalFieldConverter extends SinkFieldConverter {
// 
//     public enum Format {
//         DECIMAL128,         //needs MongoDB v3.4+
//         LEGACYDOUBLE        //results in double approximation
//     }
// 
//     private Format format;
// 
//     public DecimalFieldConverter() {
//         super(Decimal.schema(0));
//         this.format = Format.DECIMAL128;
//     }
// 
//     public DecimalFieldConverter(Format format) {
//         super(Decimal.schema(0));
//         this.format = format;
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
// 
//         if(data instanceof BigDecimal) {
//             if(format.equals(Format.DECIMAL128))
//                 return new BsonDecimal128(new Decimal128((BigDecimal)data));
// 
//             if(format.equals(Format.LEGACYDOUBLE))
//                 return new BsonDouble(((BigDecimal)data).doubleValue());
//         }
// 
//         throw new DataException("error: decimal conversion not possible when data is"
//                         + " of type "+data.getClass().getName() + " and format is "+format);
// 
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimeFieldConverter.java
// public class TimeFieldConverter extends SinkFieldConverter {
// 
//     public TimeFieldConverter() {
//         super(Time.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }
// 
// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/TimestampFieldConverter.java
// public class TimestampFieldConverter extends SinkFieldConverter {
// 
//     public TimestampFieldConverter() {
//         super(Timestamp.SCHEMA);
//     }
// 
//     @Override
//     public BsonValue toBson(Object data) {
//         return new BsonDateTime(((java.util.Date)data).getTime());
//     }
// }


// Path: src/main/java/at/grahsl/kafka/connect/mongodb/converter/AvroJsonSchemafulRecordConverter.java
import org.slf4j.LoggerFactory;
import java.util.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.*;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DateFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.DecimalFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimeFieldConverter;
import at.grahsl.kafka.connect.mongodb.converter.types.sink.bson.logical.TimestampFieldConverter;
import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.*;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.DataException;
import org.bson.BsonArray;
import org.bson.BsonDocument;
import org.bson.BsonNull;
import org.bson.BsonValue;
import org.slf4j.Logger;

/*
 * Copyright (c) 2017. Hans-Peter Grahsl (grahslhp@gmail.com)
 *
 * 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 at.grahsl.kafka.connect.mongodb.converter;



//looks like Avro and JSON + Schema is convertible by means of
//a unified conversion approach since they are using the
//same the Struct/Type information ...
public class AvroJsonSchemafulRecordConverter implements RecordConverter {

    public static final Set<String> LOGICAL_TYPE_NAMES = new HashSet<>(
            Arrays.asList(Date.LOGICAL_NAME, Decimal.LOGICAL_NAME,
                            Time.LOGICAL_NAME, Timestamp.LOGICAL_NAME)
        );

    private final Map<Schema.Type, SinkFieldConverter> converters = new HashMap<>();
    private final Map<String, SinkFieldConverter> logicalConverters = new HashMap<>();

    private static Logger logger = LoggerFactory.getLogger(AvroJsonSchemafulRecordConverter.class);

    public AvroJsonSchemafulRecordConverter() {

        //standard types
        registerSinkFieldConverter(new BooleanFieldConverter());
        registerSinkFieldConverter(new Int8FieldConverter());
        registerSinkFieldConverter(new Int16FieldConverter());
        registerSinkFieldConverter(new Int32FieldConverter());
        registerSinkFieldConverter(new Int64FieldConverter());
        registerSinkFieldConverter(new Float32FieldConverter());
        registerSinkFieldConverter(new Float64FieldConverter());
        registerSinkFieldConverter(new StringFieldConverter());
        registerSinkFieldConverter(new BytesFieldConverter());

        //logical types
        registerSinkFieldLogicalConverter(new DateFieldConverter());
        registerSinkFieldLogicalConverter(new TimeFieldConverter());
        registerSinkFieldLogicalConverter(new TimestampFieldConverter());