Avro schema projection test 02

projection_02.c

/*
* Build using:
* gcc -Wall -std=c99 projection_02.c -o projection_02_test -I/usr/include -I$AVRO_INCLUDE_DIR -L$AVRO_LIB_DIR -lavro
*/

#include <stdlib.h>
#include <stdio.h>
#include <avro.h>

/* Simple schema for this test */

const char  SIMPLE_SCHEMA[] =
"{\"type\":\"record\",\
  \"name\":\"SimpleScehma\",\
  \"fields\":[\
     {\"name\": \"Field_1\", \"type\": \"int\"},\
     {\"name\": \"Field_2\", \"type\": {\
       \"name\": \"SubRecord\",\
       \"type\": \"record\",\
       \"fields\": [\
        {\"name\": \"SubField_1\", \"type\": \"long\", \"default\": [\"0\"]},\
        {\"name\": \"SubField_2\", \"type\": \"int\", \"default\": [\"0\"]}]}},\
     {\"name\": \"Field_3\", \"type\": \"int\"}]}";


const char  PROJECTED_SCHEMA[] =
"{\"type\":\"record\",\
  \"name\":\"SimpleScehma\",\
  \"fields\":[\
     {\"name\": \"Field_2\", \"type\": {\
       \"name\": \"SubRecord\",\
       \"type\": \"record\",\
       \"fields\": [\
        {\"name\": \"SubField_1\", \"type\": \"long\", \"default\": [\"0\"]},\
        {\"name\": \"SubField_2\", \"type\": \"int\", \"default\": [\"0\"]}]}}]}";


const char *archive_file = "archive_file.avro";

int field_1_val = 0;
int field_2_val = 0;

avro_schema_t schema;



void add_item(avro_file_writer_t writer) {

    avro_value_t value;
    avro_value_iface_t *iface = avro_generic_class_from_schema(schema);
    avro_generic_value_new(iface, &value);

    avro_value_t field_1;
    avro_value_t field_2;
    avro_value_t field_3;
    avro_value_t sub_field_1;
    avro_value_t sub_field_2;
    size_t index = 0;

    if (avro_value_get_by_name(&value, "Field_1", &field_1, &index) == 0) {
        avro_value_set_int(&field_1, ++field_1_val);
    }

    if (avro_value_get_by_name(&value, "Field_2", &field_2, &index) == 0) {

        if (avro_value_get_by_name(&field_2, "SubField_1", &sub_field_1, &index) == 0) {
            avro_value_set_long(&sub_field_1, 42);
        }
        if (avro_value_get_by_name(&field_2, "SubField_2", &sub_field_2, &index) == 0) {
            avro_value_set_int(&sub_field_2, 24);
        }
    }

    if (avro_value_get_by_name(&value, "Field_3", &field_3, &index) == 0) {
        avro_value_set_int(&field_3, 3);
    }

    if (avro_file_writer_append_value(writer, &value)) {
        printf("Error appending item to archive: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    avro_value_decref(&field_1);
    //avro_value_decref(&field_2);   // causes segfault!?
    avro_value_iface_decref(iface);
    avro_value_decref(&value);
}


void create_archive_test() {

    remove(archive_file);

    avro_file_writer_t writer;
    if (avro_file_writer_create(archive_file, schema, &writer)) {
        printf("Error creating file writer: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    for (int i=0; i<5; i++) {
        add_item(writer);
    }

    avro_file_writer_flush(writer);
    avro_file_writer_close(writer);

}


void read_archive_test() {

    avro_file_reader_t reader;
    if (avro_file_reader(archive_file, &reader)) {
        printf("Error creating reader for test: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    avro_schema_t writer_schema;
    writer_schema = avro_file_reader_get_writer_schema(reader);

    avro_value_t value;
    avro_value_iface_t *iface = avro_generic_class_from_schema(writer_schema);
    avro_generic_value_new(iface, &value);

    while(avro_file_reader_read_value(reader, &value)) {
        char *json;
        if (avro_value_to_json(&value, 1, &json)) {
            printf("Problem converting value to JSON: %s\n", avro_strerror());
        } else {
            printf("%s\n", json);
        }
        free(json);
        avro_value_reset(&value);
    }

    avro_value_decref(&value);
    avro_value_iface_decref(iface);
    avro_file_reader_close(reader);
    avro_schema_decref(writer_schema);
}


void projection_test() {

    avro_schema_t projection_schema;
    if (avro_schema_from_json_literal(PROJECTED_SCHEMA, &projection_schema)) {
        printf("Error loading projection schema from file: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    avro_value_t value;
    avro_value_iface_t *iface = avro_generic_class_from_schema(projection_schema);
    avro_generic_value_new(iface, &value);

    avro_file_reader_t reader;
    if (avro_file_reader(archive_file, &reader)) {
        printf("Error creating reader for projection test: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    while(avro_file_reader_read_value(reader, &value)==0) {
        char *json;
        if (avro_value_to_json(&value, 1, &json)) {
            printf("Problem converting projected value to JSON: %s\n", avro_strerror());
        } else {
            printf("%s\n", json);
        }
        free(json);
        avro_value_reset(&value);
    }

    avro_value_decref(&value);
    avro_value_iface_decref(iface);
    avro_file_reader_close(reader);
    avro_schema_decref(projection_schema);
}



int main(int argc, char *argv[]) {

    if (avro_schema_from_json_literal(SIMPLE_SCHEMA, &schema)) {
        printf("Error loading schema from file: %s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }

    printf("Creating archive...\n");
    create_archive_test();
    printf("Reading archive...\n");
    read_archive_test();
    printf("Reading archive using a projection schema...\n");
    projection_test();

    avro_schema_decref(schema);
    return 0;
}