text: /*
   * 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.
   */
  
  #include "avro_private.h"
  #include "avro/allocation.h"
  #include "avro/consumer.h"
  #include "avro/errors.h"
  #include "avro/resolver.h"
  #include "avro/value.h"
  #include <stdlib.h>
  #include <errno.h>
  #include <string.h>
  #include "encoding.h"
  #include "schema.h"
  #include "datum.h"
  
  
  static int
  read_enum(avro_reader_t reader, const avro_encoding_t * enc,
  	  avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	int64_t index;
  
  	check_prefix(rval, enc->read_long(reader, &index),
  		     "Cannot read enum value: ");
  	return avro_consumer_call(consumer, enum_value, index, ud);
  }
  
  static int
  read_array(avro_reader_t reader, const avro_encoding_t * enc,
  	   avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	int64_t i;          /* index within the current block */
  	int64_t index = 0;  /* index within the entire array */
  	int64_t block_count;
  	int64_t block_size;
  
  	check_prefix(rval, enc->read_long(reader, &block_count),
  		     "Cannot read array block count: ");
  	check(rval, avro_consumer_call(consumer, array_start_block,
  				       1, block_count, ud));
  
  	while (block_count != 0) {
  		if (block_count < 0) {
  			block_count = block_count * -1;
  			check_prefix(rval, enc->read_long(reader, &block_size),
  				     "Cannot read array block size: ");
  		}
  
  		for (i = 0; i < block_count; i++, index++) {
  			avro_consumer_t  *element_consumer = NULL;
  			void  *element_ud = NULL;
  
  			check(rval,
  			      avro_consumer_call(consumer, array_element,
  					         index, &element_consumer, &element_ud,
  						 ud));
  
  			check(rval, avro_consume_binary(reader, element_consumer, element_ud));
  		}
  
  		check_prefix(rval, enc->read_long(reader, &block_count),
  			     "Cannot read array block count: ");
  		check(rval, avro_consumer_call(consumer, array_start_block,
  					       0, block_count, ud));
  	}
  
  	return 0;
  }
  
  static int
  read_map(avro_reader_t reader, const avro_encoding_t * enc,
  	 avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	int64_t i;          /* index within the current block */
  	int64_t index = 0;  /* index within the entire array */
  	int64_t block_count;
  	int64_t block_size;
  
  	check_prefix(rval, enc->read_long(reader, &block_count),
  		     "Cannot read map block count: ");
  	check(rval, avro_consumer_call(consumer, map_start_block,
  				       1, block_count, ud));
  
  	while (block_count != 0) {
  		if (block_count < 0) {
  			block_count = block_count * -1;
  			check_prefix(rval, enc->read_long(reader, &block_size),
  				     "Cannot read map block size: ");
  		}
  
  		for (i = 0; i < block_count; i++, index++) {
  			char *key;
  			int64_t key_size;
  			avro_consumer_t  *element_consumer = NULL;
  			void  *element_ud = NULL;
  
  			check_prefix(rval, enc->read_string(reader, &key, &key_size),
  				     "Cannot read map key: ");
  
  			rval = avro_consumer_call(consumer, map_element,
  						  index, key,
  						  &element_consumer, &element_ud,
  						  ud);
  			if (rval) {
  				avro_free(key, key_size);
  				return rval;
  			}
  
  			rval = avro_consume_binary(reader, element_consumer, element_ud);
  			if (rval) {
  				avro_free(key, key_size);
  				return rval;
  			}
  
  			avro_free(key, key_size);
  		}
  
  		check_prefix(rval, enc->read_long(reader, &block_count),
  			     "Cannot read map block count: ");
  		check(rval, avro_consumer_call(consumer, map_start_block,
  					       0, block_count, ud));
  	}
  
  	return 0;
  }
  
  static int
  read_union(avro_reader_t reader, const avro_encoding_t * enc,
  	   avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	int64_t discriminant;
  	avro_consumer_t  *branch_consumer = NULL;
  	void  *branch_ud = NULL;
  
  	check_prefix(rval, enc->read_long(reader, &discriminant),
  		     "Cannot read union discriminant: ");
  	check(rval, avro_consumer_call(consumer, union_branch,
  				       discriminant,
  				       &branch_consumer, &branch_ud, ud));
  	return avro_consume_binary(reader, branch_consumer, branch_ud);
  }
  
  static int
  read_record(avro_reader_t reader, const avro_encoding_t * enc,
  	    avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	size_t  num_fields;
  	unsigned int  i;
  
  	AVRO_UNUSED(enc);
  
  	check(rval, avro_consumer_call(consumer, record_start, ud));
  
  	num_fields = avro_schema_record_size(consumer->schema);
  	for (i = 0; i < num_fields; i++) {
  		avro_consumer_t  *field_consumer = NULL;
  		void  *field_ud = NULL;
  
  		check(rval, avro_consumer_call(consumer, record_field,
  					       i, &field_consumer, &field_ud,
  					       ud));
  
  		if (field_consumer) {
  			check(rval, avro_consume_binary(reader, field_consumer, field_ud));
  		} else {
  			avro_schema_t  field_schema =
  			    avro_schema_record_field_get_by_index(consumer->schema, i);
  			check(rval, avro_skip_data(reader, field_schema));
  		}
  	}
  
  	return 0;
  }
  
  int
  avro_consume_binary(avro_reader_t reader, avro_consumer_t *consumer, void *ud)
  {
  	int rval;
  	const avro_encoding_t *enc = &avro_binary_encoding;
  
  	check_param(EINVAL, reader, "reader");
  	check_param(EINVAL, consumer, "consumer");
  
  	switch (avro_typeof(consumer->schema)) {
  	case AVRO_NULL:
  		check_prefix(rval, enc->read_null(reader),
  			     "Cannot read null value: ");
  		check(rval, avro_consumer_call(consumer, null_value, ud));
  		break;
  
  	case AVRO_BOOLEAN:
  		{
  			int8_t b;
  			check_prefix(rval, enc->read_boolean(reader, &b),
  				     "Cannot read boolean value: ");
  			check(rval, avro_consumer_call(consumer, boolean_value, b, ud));
  		}
  		break;
  
  	case AVRO_STRING:
  		{
  			int64_t len;
  			char *s;
  			check_prefix(rval, enc->read_string(reader, &s, &len),
  				     "Cannot read string value: ");
  			check(rval, avro_consumer_call(consumer, string_value, s, len, ud));
  		}
  		break;
  
  	case AVRO_INT32:
  		{
  			int32_t i;
  			check_prefix(rval, enc->read_int(reader, &i),
  				    "Cannot read int value: ");
  			check(rval, avro_consumer_call(consumer, int_value, i, ud));
  		}
  		break;
  
  	case AVRO_INT64:
  		{
  			int64_t l;
  			check_prefix(rval, enc->read_long(reader, &l),
  				     "Cannot read long value: ");
  			check(rval, avro_consumer_call(consumer, long_value, l, ud));
  		}
  		break;
  
  	case AVRO_FLOAT:
  		{
  			float f;
  			check_prefix(rval, enc->read_float(reader, &f),
  				     "Cannot read float value: ");
  			check(rval, avro_consumer_call(consumer, float_value, f, ud));
  		}
  		break;
  
  	case AVRO_DOUBLE:
  		{
  			double d;
  			check_prefix(rval, enc->read_double(reader, &d),
  				     "Cannot read double value: ");
  			check(rval, avro_consumer_call(consumer, double_value, d, ud));
  		}
  		break;
  
  	case AVRO_BYTES:
  		{
  			char *bytes;
  			int64_t len;
  			check_prefix(rval, enc->read_bytes(reader, &bytes, &len),
  				     "Cannot read bytes value: ");
  			check(rval, avro_consumer_call(consumer, bytes_value, bytes, len, ud));
  		}
  		break;
  
  	case AVRO_FIXED:
  		{
  			char *bytes;
  			int64_t size =
  			    avro_schema_to_fixed(consumer->schema)->size;
  
  			bytes = (char *) avro_malloc(size);
  			if (!bytes) {
  				avro_prefix_error("Cannot allocate new fixed value");
  				return ENOMEM;
  			}
  			rval = avro_read(reader, bytes, size);
  			if (rval) {
  				avro_prefix_error("Cannot read fixed value: ");
  				avro_free(bytes, size);
  				return rval;
  			}
  
  			rval = avro_consumer_call(consumer, fixed_value, bytes, size, ud);
  			if (rval) {
  				avro_free(bytes, size);
  				return rval;
  			}
  		}
  		break;
  
  	case AVRO_ENUM:
  		check(rval, read_enum(reader, enc, consumer, ud));
  		break;
  
  	case AVRO_ARRAY:
  		check(rval, read_array(reader, enc, consumer, ud));
  		break;
  
  	case AVRO_MAP:
  		check(rval, read_map(reader, enc, consumer, ud));
  		break;
  
  	case AVRO_UNION:
  		check(rval, read_union(reader, enc, consumer, ud));
  		break;
  
  	case AVRO_RECORD:
  		check(rval, read_record(reader, enc, consumer, ud));
  		break;
  
  	case AVRO_LINK:
  		avro_set_error("Consumer can't consume a link schema directly");
  		return EINVAL;
  	}
  
  	return 0;
  }
  