content: /*
   * 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 org.apache.kafka.connect.data;
  
  import org.apache.kafka.connect.errors.DataException;
  
  import java.nio.ByteBuffer;
  import java.util.Arrays;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  
  /**
   * <p>
   *     A structured record containing a set of named fields with values, each field using an independent {@link Schema}.
   *     Struct objects must specify a complete {@link Schema} up front, and only fields specified in the Schema may be set.
   * </p>
   * <p>
   *     The Struct's {@link #put(String, Object)} method returns the Struct itself to provide a fluent API for constructing
   *     complete objects:
   *     <pre>
   *         Schema schema = SchemaBuilder.struct().name("com.example.Person")
   *             .field("name", Schema.STRING_SCHEMA).field("age", Schema.INT32_SCHEMA).build()
   *         Struct struct = new Struct(schema).put("name", "Bobby McGee").put("age", 21)
   *     </pre>
   * </p>
   */
  public class Struct {
  
      private final Schema schema;
      private final Object[] values;
  
      /**
       * Create a new Struct for this {@link Schema}
       * @param schema the {@link Schema} for the Struct
       */
      public Struct(Schema schema) {
          if (schema.type() != Schema.Type.STRUCT)
              throw new DataException("Not a struct schema: " + schema);
          this.schema = schema;
          this.values = new Object[schema.fields().size()];
      }
  
      /**
       * Get the schema for this Struct.
       * @return the Struct's schema
       */
      public Schema schema() {
          return schema;
      }
  
      /**
       * Get the value of a field, returning the default value if no value has been set yet and a default value is specified
       * in the field's schema. Because this handles fields of all types, the value is returned as an {@link Object} and
       * must be cast to a more specific type.
       * @param fieldName the field name to lookup
       * @return the value for the field
       */
      public Object get(String fieldName) {
          Field field = lookupField(fieldName);
          return get(field);
      }
  
      /**
       * Get the value of a field, returning the default value if no value has been set yet and a default value is specified
       * in the field's schema. Because this handles fields of all types, the value is returned as an {@link Object} and
       * must be cast to a more specific type.
       * @param field the field to lookup
       * @return the value for the field
       */
      public Object get(Field field) {
          Object val = values[field.index()];
          if (val == null && field.schema().defaultValue() != null) {
              val = field.schema().defaultValue();
          }
          return val;
      }
  
      /**
       * Get the underlying raw value for the field without accounting for default values.
       * @param fieldName the field to get the value of
       * @return the raw value
       */
      public Object getWithoutDefault(String fieldName) {
          Field field = lookupField(fieldName);
          return values[field.index()];
      }
  
      // Note that all getters have to have boxed return types since the fields might be optional
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Byte.
       */
      public Byte getInt8(String fieldName) {
          return (Byte) getCheckType(fieldName, Schema.Type.INT8);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Short.
       */
      public Short getInt16(String fieldName) {
          return (Short) getCheckType(fieldName, Schema.Type.INT16);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to an Integer.
       */
      public Integer getInt32(String fieldName) {
          return (Integer) getCheckType(fieldName, Schema.Type.INT32);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Long.
       */
      public Long getInt64(String fieldName) {
          return (Long) getCheckType(fieldName, Schema.Type.INT64);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Float.
       */
      public Float getFloat32(String fieldName) {
          return (Float) getCheckType(fieldName, Schema.Type.FLOAT32);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Double.
       */
      public Double getFloat64(String fieldName) {
          return (Double) getCheckType(fieldName, Schema.Type.FLOAT64);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Boolean.
       */
      public Boolean getBoolean(String fieldName) {
          return (Boolean) getCheckType(fieldName, Schema.Type.BOOLEAN);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a String.
       */
      public String getString(String fieldName) {
          return (String) getCheckType(fieldName, Schema.Type.STRING);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a byte[].
       */
      public byte[] getBytes(String fieldName) {
          Object bytes = getCheckType(fieldName, Schema.Type.BYTES);
          if (bytes instanceof ByteBuffer)
              return ((ByteBuffer) bytes).array();
          return (byte[]) bytes;
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a List.
       */
      @SuppressWarnings("unchecked")
      public <T> List<T> getArray(String fieldName) {
          return (List<T>) getCheckType(fieldName, Schema.Type.ARRAY);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Map.
       */
      @SuppressWarnings("unchecked")
      public <K, V> Map<K, V> getMap(String fieldName) {
          return (Map<K, V>) getCheckType(fieldName, Schema.Type.MAP);
      }
  
      /**
       * Equivalent to calling {@link #get(String)} and casting the result to a Struct.
       */
      public Struct getStruct(String fieldName) {
          return (Struct) getCheckType(fieldName, Schema.Type.STRUCT);
      }
  
      /**
       * Set the value of a field. Validates the value, throwing a {@link DataException} if it does not match the field's
       * {@link Schema}.
       * @param fieldName the name of the field to set
       * @param value the value of the field
       * @return the Struct, to allow chaining of {@link #put(String, Object)} calls
       */
      public Struct put(String fieldName, Object value) {
          Field field = lookupField(fieldName);
          return put(field, value);
      }
  
      /**
       * Set the value of a field. Validates the value, throwing a {@link DataException} if it does not match the field's
       * {@link Schema}.
       * @param field the field to set
       * @param value the value of the field
       * @return the Struct, to allow chaining of {@link #put(String, Object)} calls
       */
      public Struct put(Field field, Object value) {
          if (null == field)
              throw new DataException("field cannot be null.");
          ConnectSchema.validateValue(field.name(), field.schema(), value);
          values[field.index()] = value;
          return this;
      }
  
  
      /**
       * Validates that this struct has filled in all the necessary data with valid values. For required fields
       * without defaults, this validates that a value has been set and has matching types/schemas. If any validation
       * fails, throws a {@link DataException}.
       */
      public void validate() {
          for (Field field : schema.fields()) {
              Schema fieldSchema = field.schema();
              Object value = values[field.index()];
              if (value == null && (fieldSchema.isOptional() || fieldSchema.defaultValue() != null))
                  continue;
              ConnectSchema.validateValue(field.name(), fieldSchema, value);
          }
      }
  
      @Override
      public boolean equals(Object o) {
          if (this == o) return true;
          if (o == null || getClass() != o.getClass()) return false;
          Struct struct = (Struct) o;
          return Objects.equals(schema, struct.schema) &&
                  Arrays.deepEquals(values, struct.values);
      }
  
      @Override
      public int hashCode() {
          return Objects.hash(schema, Arrays.deepHashCode(values));
      }
  
      private Field lookupField(String fieldName) {
          Field field = schema.field(fieldName);
          if (field == null)
              throw new DataException(fieldName + " is not a valid field name");
          return field;
      }
  
      // Get the field's value, but also check that the field matches the specified type, throwing an exception if it doesn't.
      // Used to implement the get*() methods that return typed data instead of Object
      private Object getCheckType(String fieldName, Schema.Type type) {
          Field field = lookupField(fieldName);
          if (field.schema().type() != type)
              throw new DataException("Field '" + fieldName + "' is not of type " + type);
          return values[field.index()];
      }
  
      @Override
      public String toString() {
          final StringBuilder sb = new StringBuilder("Struct{");
          boolean first = true;
          for (int i = 0; i < values.length; i++) {
              final Object value = values[i];
              if (value != null) {
                  final Field field = schema.fields().get(i);
                  if (first) {
                      first = false;
                  } else {
                      sb.append(",");
                  }
                  sb.append(field.name()).append("=").append(value);
              }
          }
          return sb.append("}").toString();
      }
  
  }
  