content: /*
   * Copyright 2021-2024 Aklivity Inc
   *
   * Licensed under the Aklivity Community License (the "License"); you may not use
   * this file except in compliance with the License.  You may obtain a copy of the
   * License at
   *
   *   https://www.aklivity.io/aklivity-community-license/
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations under the License.
   */
  package io.aklivity.zilla.runtime.model.avro.internal;
  
  import java.io.IOException;
  
  import org.agrona.DirectBuffer;
  import org.apache.avro.AvroRuntimeException;
  import org.apache.avro.Schema;
  import org.apache.avro.generic.GenericDatumReader;
  import org.apache.avro.generic.GenericDatumWriter;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.avro.io.CanonicalJsonDecoder;
  
  import io.aklivity.zilla.runtime.engine.EngineContext;
  import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
  import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
  import io.aklivity.zilla.runtime.engine.model.function.ValueConsumer;
  import io.aklivity.zilla.runtime.model.avro.config.AvroModelConfig;
  
  public class AvroWriteConverterHandler extends AvroModelHandler implements ConverterHandler
  {
      public AvroWriteConverterHandler(
          AvroModelConfiguration config,
          AvroModelConfig options,
          EngineContext context)
      {
          super(config, options, context);
      }
  
      @Override
      public int padding(
          DirectBuffer data,
          int index,
          int length)
      {
          return handler.encodePadding(length);
      }
  
      @Override
      public int convert(
          long traceId,
          long bindingId,
          DirectBuffer data,
          int index,
          int length,
          ValueConsumer next)
      {
          int valLength = -1;
  
          int schemaId = catalog != null && catalog.id > 0
                  ? catalog.id
                  : handler.resolve(subject, catalog.version);
  
          if (VIEW_JSON.equals(view))
          {
              valLength = handler.encode(traceId, bindingId, schemaId, data, index, length, next, this::serializeJsonRecord);
          }
          else if (validate(traceId, bindingId, schemaId, data, index, length))
          {
              valLength = handler.encode(traceId, bindingId, schemaId, data, index, length, next, CatalogHandler.Encoder.IDENTITY);
          }
          return valLength;
      }
  
      private int serializeJsonRecord(
          long traceId,
          long bindingId,
          int schemaId,
          DirectBuffer buffer,
          int index,
          int length,
          ValueConsumer next)
      {
          int valLength = -1;
          try
          {
              Schema schema = supplySchema(schemaId);
  
              if (schema != null)
              {
                  switch (schema.getType())
                  {
                  case STRING:
                      next.accept(buffer, index, length);
                      valLength = length;
                      break;
                  case RECORD:
                      GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
                      GenericDatumWriter<GenericRecord> writer = supplyWriter(schemaId);
                      if (reader != null)
                      {
                          GenericRecord record = supplyRecord(schemaId);
                          in.wrap(buffer, index, length);
                          expandable.wrap(expandable.buffer());
                          CanonicalJsonDecoder decoder = new CanonicalJsonDecoder(schema, in);
                          record = reader.read(record, decoder);
                          encoderFactory.binaryEncoder(expandable, encoder);
                          writer.write(record, encoder);
                          encoder.flush();
                          int position = expandable.position();
                          if (position > 0)
                          {
                              next.accept(expandable.buffer(), 0, position);
                              valLength = position;
                          }
                      }
                      break;
                  default:
                      break;
                  }
              }
          }
          catch (IOException | AvroRuntimeException ex)
          {
              event.validationFailure(traceId, bindingId, ex.getMessage());
          }
          return valLength;
      }
  }
  