TT#108605 redis "location" store schema #1
Change-Id: I96150ac365658e1cf0c156e725e88b75757977e9
(cherry picked from commit e9e315fd8b)
mr7.5.5
Rene Krenn
5 years ago
parent
41634ee566
commit
1673e11728
@ -0,0 +1,84 @@
|
||||
package NGCP::BulkProcessor::NoSqlConnectors::RedisEntry;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use Tie::IxHash;
|
||||
|
||||
use NGCP::BulkProcessor::NoSqlConnectors::RedisProcessor qw(init_entry);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter);
|
||||
our @EXPORT_OK = qw(
|
||||
$HASH_TYPE
|
||||
$SET_TYPE
|
||||
$LIST_TYPE
|
||||
$ZSET_TYPE
|
||||
$STRING_TYPE
|
||||
);
|
||||
|
||||
#should correspond to the type names for redis SCAN:
|
||||
our $HASH_TYPE = 'hash';
|
||||
our $SET_TYPE = 'set';
|
||||
our $LIST_TYPE = 'list';
|
||||
our $ZSET_TYPE = 'zset';
|
||||
our $STRING_TYPE = 'string';
|
||||
|
||||
sub new {
|
||||
|
||||
my $base_class = shift;
|
||||
my $class = shift;
|
||||
my $type = shift;
|
||||
my $self = bless {}, $class;
|
||||
$type = 'default' unless $type;
|
||||
$type = lc($type);
|
||||
my $value;
|
||||
if ($type eq 'set') {
|
||||
$value = {};
|
||||
} elsif ($type eq 'list') {
|
||||
$value = [];
|
||||
} elsif ($type eq 'zset') {
|
||||
my %value = ();
|
||||
tie(%value, 'Tie::IxHash');
|
||||
$value = \%value;
|
||||
} elsif ($type eq 'hash') {
|
||||
$value = {};
|
||||
} else { #($type eq 'string') {
|
||||
$type = 'string';
|
||||
$value = undef;
|
||||
}
|
||||
$self->{type} = $type;
|
||||
$self->{value} = $value;
|
||||
return init_entry($self,@_);
|
||||
|
||||
}
|
||||
|
||||
sub getvalue {
|
||||
my $self = shift;
|
||||
#$self->{value} = shift if scalar @_;
|
||||
return $self->{value};
|
||||
}
|
||||
|
||||
sub gettype {
|
||||
my $self = shift;
|
||||
return $self->{type};
|
||||
}
|
||||
|
||||
sub gethash {
|
||||
my $self = shift;
|
||||
my $fieldvalues;
|
||||
if ($self->{type} eq 'set') {
|
||||
$fieldvalues = [ sort keys %{$self->{value}} ];
|
||||
} elsif ($self->{type} eq 'list') {
|
||||
$fieldvalues = $self->{value};
|
||||
} elsif ($self->{type} eq 'zset') {
|
||||
$fieldvalues = [ keys %{$self->{value}} ];
|
||||
} elsif ($self->{type} eq 'hash') {
|
||||
$fieldvalues = [ map { $self->{value}->{$_}; } sort keys %{$self->{value}} ];
|
||||
} else { #($type eq 'string') {
|
||||
$fieldvalues = [ $self->{value} ];
|
||||
}
|
||||
return get_rowhash($fieldvalues);
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,64 @@
|
||||
use strict;
|
||||
|
||||
use NGCP::BulkProcessor::Globals qw();
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(destroy_stores);
|
||||
use NGCP::BulkProcessor::NoSqlConnectors::Redis qw();
|
||||
use NGCP::BulkProcessor::Redis::Trunk::location::entry qw(
|
||||
get_entry
|
||||
get_entry_by_ruid
|
||||
process_keys
|
||||
);
|
||||
|
||||
goto SKIP;
|
||||
{
|
||||
my $host = '192.168.0.146';
|
||||
my $port = '6379';
|
||||
my $sock = undef;
|
||||
my $password = undef;
|
||||
my $databaseindex = '0';
|
||||
|
||||
my $store = NGCP::BulkProcessor::NoSqlConnectors::Redis->new(undef);
|
||||
$store->connect(20,undef,$host);
|
||||
|
||||
my @result = $store->keys_shared('*',sub {
|
||||
my ($reply, $error) = @_;
|
||||
die "Oops, got an error: $error\n" if defined $error;
|
||||
print "$_\n" for @$reply;
|
||||
});
|
||||
#print join("\n",@keys);
|
||||
}
|
||||
|
||||
SKIP:
|
||||
{
|
||||
$NGCP::BulkProcessor::Globals::location_host = '192.168.0.146';
|
||||
my $static_context = {
|
||||
|
||||
};
|
||||
my $result = process_keys(
|
||||
process_code => sub {
|
||||
my ($context,$records,$row_offset) = @_;
|
||||
return 1;
|
||||
},
|
||||
static_context => $static_context,
|
||||
blocksize => 10000,
|
||||
init_process_context_code => sub {
|
||||
my ($context)= @_;
|
||||
},
|
||||
uninit_process_context_code => sub {
|
||||
my ($context)= @_;
|
||||
destroy_stores();
|
||||
},
|
||||
multithreading => 1,
|
||||
numofthreads => 4,
|
||||
#load_recursive => ,
|
||||
);
|
||||
|
||||
}
|
||||
|
||||
SKIP:
|
||||
{
|
||||
my $location = get_entry_by_ruid();
|
||||
}
|
||||
|
||||
exit;
|
||||
|
||||
@ -0,0 +1,3 @@
|
||||
|
||||
|
||||
get_location_store
|
||||
@ -0,0 +1,155 @@
|
||||
package NGCP::BulkProcessor::Redis::Trunk::location::entry;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use NGCP::BulkProcessor::ConnectorPool qw(
|
||||
get_location_store
|
||||
destroy_stores
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::NoSqlConnectors::RedisProcessor qw(
|
||||
copy_value
|
||||
process_entries
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::NoSqlConnectors::RedisEntry qw();
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::NoSqlConnectors::RedisEntry);
|
||||
our @EXPORT_OK = qw(
|
||||
get_entry
|
||||
get_entry_by_ruid
|
||||
process_keys
|
||||
);
|
||||
|
||||
my $get_store = \&get_location_store;
|
||||
|
||||
my $table = 'location:entry';
|
||||
my $type = $NGCP::BulkProcessor::NoSqlConnectors::RedisEntry::HASH_TYPE;
|
||||
my $get_key_from_ruid = sub {
|
||||
my ($ruid) = @_;
|
||||
return $table . '::' . $ruid;
|
||||
};
|
||||
|
||||
my $fieldnames = [
|
||||
'instance',
|
||||
'domain',
|
||||
'cseq',
|
||||
'partition',
|
||||
'ruid',
|
||||
'connection_id',
|
||||
'username',
|
||||
'keepalive',
|
||||
'path',
|
||||
'reg_id',
|
||||
'contact',
|
||||
'flags',
|
||||
'received',
|
||||
'callid',
|
||||
'socket',
|
||||
'cflags',
|
||||
'expires',
|
||||
'methods',
|
||||
'user_agent',
|
||||
'q',
|
||||
'last_modified',
|
||||
'server_id',
|
||||
];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::NoSqlConnectors::RedisEntry->new($class,$type,$fieldnames);
|
||||
|
||||
copy_value($self,shift,$fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub get_entry {
|
||||
|
||||
my ($key,$load_recursive) = @_;
|
||||
my $store = &$get_store();
|
||||
|
||||
return builditems_fromrows({ $store->hgetall($key) },$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub get_entry_by_ruid {
|
||||
|
||||
my ($ruid,$load_recursive) = @_;
|
||||
my $store = &$get_store();
|
||||
|
||||
return builditems_fromrows({ $store->hgetall(&$get_key_from_ruid($ruid)) },$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub builditems_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my $item;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
my @items = ();
|
||||
foreach my $row (@$rows) {
|
||||
$item = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @items,$item;
|
||||
}
|
||||
return \@items;
|
||||
} elsif (defined $rows and ref $rows eq 'HASH') {
|
||||
$item = __PACKAGE__->new($rows);
|
||||
return $item;
|
||||
}
|
||||
return undef;
|
||||
|
||||
}
|
||||
|
||||
sub process_keys {
|
||||
|
||||
my %params = @_;
|
||||
my ($process_code,
|
||||
$static_context,
|
||||
$blocksize,
|
||||
$init_process_context_code,
|
||||
$uninit_process_context_code,
|
||||
$multithreading,
|
||||
$numofthreads,
|
||||
$load_recursive) = @params{qw/
|
||||
process_code
|
||||
static_context
|
||||
blocksize
|
||||
init_process_context_code
|
||||
uninit_process_context_code
|
||||
multithreading
|
||||
numofthreads
|
||||
load_recursive
|
||||
/};
|
||||
|
||||
return process_entries(
|
||||
get_store => $get_store,
|
||||
scan_pattern => &$get_key_from_ruid('*'),
|
||||
type => $type,
|
||||
process_code => sub {
|
||||
my ($context,$rowblock,$row_offset) = @_;
|
||||
return &$process_code($context,builditems_fromrows([
|
||||
map { { &$get_store()->hgetall($_) }; } @$rowblock
|
||||
],$load_recursive),$row_offset);
|
||||
},
|
||||
static_context => $static_context,
|
||||
blocksize => $blocksize,
|
||||
init_process_context_code => $init_process_context_code,
|
||||
uninit_process_context_code => $uninit_process_context_code,
|
||||
destroy_reader_stores_code => \&destroy_stores,
|
||||
multithreading => $multithreading,
|
||||
nosqlprocessing_threads => $numofthreads,
|
||||
);
|
||||
}
|
||||
|
||||
1;
|
||||
|
||||
Loading…
Reference in new issue