Kafka protocol 版本适应的通信协议

/**
 * Represents a type for an array of a particular type
 */
public class ArrayOf extends Type {

  private final Type type;
  private final boolean nullable;

  public ArrayOf(Type type) {
      this(type, false);
  }

  public static ArrayOf nullable(Type type) {
      return new ArrayOf(type, true);
  }

  private ArrayOf(Type type, boolean nullable) {
      this.type = type;
      this.nullable = nullable;
  }

  @Override
  public boolean isNullable() {
      return nullable;
  }

  @Override
  public void write(ByteBuffer buffer, Object o) {
      if (o == null) {
          buffer.putInt(-1);
          return;
      }

      Object[] objs = (Object[]) o;
      int size = objs.length;
      buffer.putInt(size);

      for (Object obj : objs)
          type.write(buffer, obj);
  }

  @Override
  public Object read(ByteBuffer buffer) {
      int size = buffer.getInt();
      if (size < 0 && isNullable())
          return null;
      else if (size < 0)
          throw new SchemaException("Array size " + size + " cannot be negative");

      if (size > buffer.remaining())
          throw new SchemaException("Error reading array of size " + size + ", only " + buffer.remaining() + " bytes available");
      Object[] objs = new Object[size];
      for (int i = 0; i < size; i++)
          objs[i] = type.read(buffer);
      return objs;
  }

/**
 * The schema for a compound record definition
 */
public class Schema extends Type {

    private final BoundField[] fields;
    private final Map<String, BoundField> fieldsByName;

    /**
     * Construct the schema with a given list of its field values
     *
     * @throws SchemaException If the given list have duplicate fields
     */
    public Schema(Field... fs) {
        this.fields = new BoundField[fs.length];
        this.fieldsByName = new HashMap<>();
        for (int i = 0; i < this.fields.length; i++) {
            Field def = fs[i];
            if (fieldsByName.containsKey(def.name))
                throw new SchemaException("Schema contains a duplicate field: " + def.name);
            this.fields[i] = new BoundField(def, this, i);
            this.fieldsByName.put(def.name, this.fields[i]);
        }
    }

    ...

    /**
     * Read a struct from the buffer
     */
    @Override
    public Struct read(ByteBuffer buffer) {
        Object[] objects = new Object[fields.length];
        for (int i = 0; i < fields.length; i++) {
            try {
                objects[i] = fields[i].def.type.read(buffer);
            } catch (Exception e) {
                throw new SchemaException("Error reading field '" + fields[i].def.name + "': " +
                                          (e.getMessage() == null ? e.getClass().getName() : e.getMessage()));
            }
        }
        return new Struct(this, objects);
    }

/**
 * A field definition bound to a particular schema.
 */
public class BoundField {
    public final Field def;
    final int index;
    final Schema schema;

    public BoundField(Field def, Schema schema, int index) {
        this.def = def;
        this.schema = schema;
        this.index = index;
    }

    @Override
    public String toString() {
        return def.name + ":" + def.type;
    }
}

/**
 * A record that can be serialized and deserialized according to a pre-defined schema
 */
public class Struct {
    private final Schema schema;
    private final Object[] values;

/**
     * Get the record value for the field with the given name by doing a hash table lookup (slower!)
     *
     * @param name The name of the field
     * @return The value in the record
     * @throws SchemaException If no such field exists
     */
    public Object get(String name) {
        BoundField field = schema.get(name);
        if (field == null)
            throw new SchemaException("No such field: " + name);
        return getFieldOrDefault(field);
    }

    public Object[] getArray(String name) {
        return (Object[]) get(name);
    }

   /**
     * Return the value of the given pre-validated field, or if the value is missing return the default value.
     *
     * @param field The field for which to get the default value
     * @throws SchemaException if the field has no value and has no default.
     */
    private Object getFieldOrDefault(BoundField field) {
        Object value = this.values[field.index];
        if (value != null)
            return value;
        else if (field.def.hasDefaultValue)
            return field.def.defaultValue;
        else if (field.def.type.isNullable())
            return null;
        else
            throw new SchemaException("Missing value for field '" + field.def.name + "' which has no default value.");
    }

ApiKeys(int id, String name, Schema[] requestSchemas, Schema[] responseSchemas) {
    this(id, name, false, requestSchemas, responseSchemas);
}

ApiKeys(int id, String name, boolean clusterAction, Schema[] requestSchemas, Schema[] responseSchemas) {
    this(id, name, clusterAction, RecordBatch.MAGIC_VALUE_V0, requestSchemas, responseSchemas);
}

ApiKeys(int id, String name, boolean clusterAction, byte minRequiredInterBrokerMagic,
        Schema[] requestSchemas, Schema[] responseSchemas) {
    if (id < 0)
        throw new IllegalArgumentException("id must not be negative, id: " + id);
    this.id = (short) id;
    this.name = name;
    

    ...
    this.requestSchemas = requestSchemas;
    this.responseSchemas = responseSchemas;
}

public Schema requestSchema(short version) {
    return schemaFor(requestSchemas, version);
}

public Schema responseSchema(short version) {
    return schemaFor(responseSchemas, version);
}

public Struct parseRequest(short version, ByteBuffer buffer) {
    return requestSchema(version).read(buffer);
}

public Struct parseResponse(short version, ByteBuffer buffer) {
    return responseSchema(version).read(buffer);
}

protected Struct parseResponse(short version, ByteBuffer buffer, short fallbackVersion) {
    int bufferPosition = buffer.position();
    try {
        return responseSchema(version).read(buffer);
    } catch (SchemaException e) {
        if (version != fallbackVersion) {
            buffer.position(bufferPosition);
            return responseSchema(fallbackVersion).read(buffer);
        } else
            throw e;
    }
}

private Schema schemaFor(Schema[] versions, short version) {
    if (!isVersionSupported(version))
        throw new IllegalArgumentException("Invalid version for API key " + this + ": " + version);
    return versions[version];
}

// ApiKeys.java
public Struct parseResponse(short version, ByteBuffer buffer) {
  return responseSchema(version).read(buffer);
}

// Schema.java
/**
 * Read a struct from the buffer
 */
@Override
public Struct read(ByteBuffer buffer) {
    Object[] objects = new Object[fields.length];
    for (int i = 0; i < fields.length; i++) {
        try {
            objects[i] = fields[i].def.type.read(buffer);
        } catch (Exception e) {
            throw new SchemaException("Error reading field '" + fields[i].def.name + "': " +
                                      (e.getMessage() == null ? e.getClass().getName() : e.getMessage()));
        }
    }
    return new Struct(this, objects);
}