hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add initial support for deflate compression in maxavro

Browse Source 
Maxavro now supports reading records with the zlib deflate
algorithm.

With this change, each data block is read into memory in one IO
operation. This allows the library to decompress the data block if
necessary.

The avrorouter does not yet use compression when writing the records.
This commit is contained in:
Markus Mäkelä
2017-02-21 13:03:38 +02:00
parent 1d8d526a01
commit c47ef968f7
10 changed files with 295 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
avro/maxavro_record.c
View File

@ -12,7 +12,7 @@
*/
#include <maxscale/cdefs.h>
#include "maxavro.h"
#include "maxavro_internal.h"
#include <string.h>
#include <maxscale/debug.h>
#include <maxscale/log_manager.h>
@ -36,12 +36,11 @@ static json_t* read_and_pack_value(MAXAVRO_FILE *file, MAXAVRO_SCHEMA_FIELD *fie
switch (field->type)
{
case MAXAVRO_TYPE_BOOL:
if (file->buffer_ptr < file->buffer_end)
{
int i = 0;
if (fread(&i, 1, 1, file->file) == 1)
{
value = json_pack("b", i);
}
memcpy(&i, file->buffer_ptr++, 1);
value = json_pack("b", i);
}
break;
@ -103,7 +102,7 @@ static json_t* read_and_pack_value(MAXAVRO_FILE *file, MAXAVRO_SCHEMA_FIELD *fie
if (str)
{
value = json_string(str);
free(str);
MXS_FREE(str);
}
}
break;
@ -220,19 +219,7 @@ bool maxavro_next_block(MAXAVRO_FILE *file)
{
if (file->last_error == MAXAVRO_ERR_NONE)
{
if (file->records_read_from_block < file->records_in_block)
{
file->records_read += file->records_in_block - file->records_read_from_block;
long curr_pos = ftell(file->file);
long offset = (long) file->block_size - (curr_pos - file->data_start_pos);
if (offset > 0)
{
fseek(file->file, offset, SEEK_CUR);
}
}
return maxavro_verify_block(file) && maxavro_read_datablock_start(file);
return maxavro_read_datablock_start(file);
}
return false;
}
@ -268,7 +255,7 @@ bool maxavro_record_seek(MAXAVRO_FILE *file, uint64_t offset)
{
/** Skip full blocks that don't have the position we want */
offset -= file->records_in_block;
fseek(file->file, file->block_size, SEEK_CUR);
fseek(file->file, file->buffer_size, SEEK_CUR);
maxavro_next_block(file);
}
@ -321,7 +308,7 @@ GWBUF* maxavro_record_read_binary(MAXAVRO_FILE *file)
return NULL;
}
long data_size = (file->data_start_pos - file->block_start_pos) + file->block_size;
long data_size = (file->data_start_pos - file->block_start_pos) + file->buffer_size;
ss_dassert(data_size > 0);
rval = gwbuf_alloc(data_size + SYNC_MARKER_SIZE);