AvroSerializer

AvroSerializer.kt

open class AvroSerializer @JvmOverloads constructor(val codecFactory: CodecFactory = CodecFactory.snappyCodec()) {

  /**
   * Avro instance 를 직렬화하여 byte array 로 변환합니다.
   */
  fun <T : SpecificRecordBase> writeAvroObject(graph: T?): ByteArray {
    return graph?.let {
      val sdw = SpecificDatumWriter<T>(graph.schema)
      val dfw = DataFileWriter(sdw).setCodec(codecFactory)

      ByteArrayOutputStream() use { bos ->
        dfw.create(graph.schema, bos)
        dfw.append(graph)
        dfw.flush()
        bos.toByteArray()
      }
    } ?: ByteArray(0)
  }

  /**
   * byte array 를 읽어 avro instance 로 빌드합니다.
   */
  @Suppress("UNCHECKED_CAST")
  fun <T : SpecificRecordBase> readAvroObject(avroBytes: ByteArray?, clazz: Class<T>): T? {
    return avroBytes?.let {
      val sdr = SpecificDatumReader(clazz)
      SeekableByteArrayInput(avroBytes) use { sin ->
        val dfr = DataFileReader(sin, sdr)
        if (dfr.hasNext()) dfr.next() else null
      }
    }
  }

  /**
   * avro instance collection 을 byte array 로 변환합니다.
   */
  fun <T : SpecificRecordBase> writeAvroList(graphs: List<T>, clazz: Class<T>): ByteArray {
//    val gclazz = getGenericParameterTypes<List<T>>(graphs)
    val schema = SpecificData.get().getSchema(clazz)
    val sdw: SpecificDatumWriter<T> = SpecificDatumWriter(schema)
    val dfw: DataFileWriter<T> = DataFileWriter(sdw).setCodec(codecFactory)

    val bos = ByteArrayOutputStream()
    try {
      dfw.create(schema, bos)
      graphs.forEach { dfw.append(it) }
      dfw.flush()
      return bos.toByteArray()
    } finally {
      bos.close()
    }
  }

  /**
   * byte array 를 읽어 avro instance collection 을 빌드합니다.
   */
  fun <T : SpecificRecordBase> readAvroList(avroBytes: ByteArray?, clazz: Class<T>): List<T> {
    val results = mutableListOf<T>()
    avroBytes?.let {
      val sdr = SpecificDatumReader(clazz)
      SeekableByteArrayInput(avroBytes) use { sin ->
        val dfr = DataFileReader(sin, sdr)
        while (dfr.hasNext()) {
          results.add(dfr.next())
        }
      }
    }
    return results
  }
}