view release on metacpan or  search on metacpan

lib/Melian.pm  view on Meta::CPAN

}

sub fetch_by_int_from {
    my ( $self, $table_name, $column_name, $id ) = @_;

    # Get table ID
    my $table = $self->get_table_id($table_name);
    my $table_id = $table->{'id'};

    # Get column ID
    my $column_id = $self->get_column_id( $table, $column_name );

    return $self->fetch_by_int( $table_id, $column_id, $id );
}

sub fetch_by_int {
    my ($self, $table_id, $column_id, $id) = @_;
    return $self->fetch_by_string($table_id, $column_id, pack 'V', $id);
}

# $conn, $table_id, $column_id, $id
sub fetch_by_int_with {
    return fetch_by_string_with($_[0], $_[1], $_[2], pack 'V', $_[3]);
}

sub load_schema_from_describe {
    my $payload = _send_with( $_[0], ACTION_DESCRIBE(), 0, 0, '' );
    defined $payload && length $payload
        or croak('Could not get schema data');
    return decode_json($payload);
}


sub load_schema_from_file {
    my $path = shift;

    open my $fh, '<', $path
        or croak("Cannot open schema file $path: $!");
    local $/;
    my $content = <$fh>;
    close $fh
        or croak("Cannot close schema file: $path: $!");

    my $decoded;
    eval {
        $decoded = decode_json($content);
        1;
    } or do {
        my $error = $@ || 'Zombie error';
        croak("Failed to parse JSON schema in file '$path': $error");
    };

    return $decoded;
}

# table1#0|60|id#0:int,table2#1|45|id#0:int;hostname#1:string
sub load_schema_from_spec {
    my $spec = shift;
    my %data;

    for my $section_data ( split m{,}, $spec ) {
        my ( $table_data, $table_period, $columns ) = split m{\|}, $section_data;
        my ( $table_name, $table_id ) = split m{#}, $table_data;
        defined $table_name && defined $table_id
            or croak('Schema spec failure: Missing table name or table ID');

        my %table_entry = (
            'name'   => $table_name,
            'id'     => $table_id,
            'period' => $table_period,
        );

        my @columns;
        foreach my $column_data ( split m{;}, $columns ) {
            my ( $column_data, $column_type ) = split /:/, $column_data;
            my ( $column_name, $column_id )   = split m{#}, $column_data;

            push @{ $table_entry{'indexes'} }, {
                'id'     => $column_id,
                'column' => $column_name,
                'type'   => $column_type,
            }
        }

        push @{ $data{'tables'} }, \%table_entry;
    }

    return \%data;
}

sub _send {
    my ( $self, $action, $table_id, $column_id, $payload ) = @_;
    $payload //= '';
    defined $table_id && defined $column_id
        or croak("Invalid table ID or index ID");

    my $header = pack(
        'CCCCN',
        MELIAN_HEADER_VERSION(),
        $action,
        $table_id,
        $column_id,
        length $payload,
    );

    _write_all( $self->{'socket'}, $header . $payload );
    my $len_buf = _read_exactly( $self->{'socket'}, 4 );
    my $len = unpack 'N', $len_buf;
    return '' if $len == 0;
    return _read_exactly( $self->{'socket'}, $len );
}

# $conn, $action, $table_id, $column_id, $payload
sub _send_with {
    $_[4] //= '';
    defined $_[2] && defined $_[3]
        or croak("Invalid table ID or index ID");

    my $header = pack(
        'CCCCN',
        MELIAN_HEADER_VERSION(),
        $_[1],
        $_[2],
        $_[3],
        length $_[4],
    );

    _write_all( $_[0], $header . $_[4] );
    my $len_buf = _read_exactly( $_[0], 4 );
    my $len = unpack 'N', $len_buf;
    return '' if $len == 0;
    return _read_exactly( $_[0], $len );
}

# $socket, $buf
sub _write_all {