1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
package IPC::PubSub::Cache::JiftyDBI;
use strict;
use warnings;
use base 'IPC::PubSub::Cache';
use IPC::PubSub::Cache::JiftyDBI::Stash;
use vars qw/$STASH/;
sub new {
my $class = shift;
my $self = {};
bless $self => $class;
$STASH ||= IPC::PubSub::Cache::JiftyDBI::Stash->new(@_);
return $self;
}
sub disconnect {
$STASH->handle->disconnect;
}
sub fetch {
my $self = shift;
my @keys_in_order = (@_);
my $items = IPC::PubSub::Cache::JiftyDBI::Stash::ItemCollection->new( handle => $STASH->handle );
foreach my $val (@keys_in_order) {
$items->limit(
column => 'data_key',
entry_aggregator => 'or',
value => $val
);
}
my %items = map { $_->data_key, [$_->expiry, $_->val] } @{ $items->items_array_ref };
return @items{@keys_in_order};
}
sub store {
my ($self, $key, $val, $time, $expiry) = @_;
$expiry ||= 0;
my $item = IPC::PubSub::Cache::JiftyDBI::Stash::Item->new(handle => $STASH->handle);
$item->load_by_cols( data_key => $key );
if ( $item->id ) {
$item->set_val($val);
$item->set_expiry($time+$expiry);
}
else {
$item->create( data_key => $key, expiry => ($time+$expiry), val => $val );
}
}
sub publisher_indices {
my ($self, $chan) = @_;
my $publishers = IPC::PubSub::Cache::JiftyDBI::Stash::PublisherCollection->new(handle => $STASH->handle);
$publishers->limit(column => 'channel', value => $chan);
my %indices;
map {$indices{$_->name} = $_->idx} @{$publishers->items_array_ref};
return \%indices;
}
sub add_publisher {
my ($self, $chan, $pub) = @_;
my $publisher = IPC::PubSub::Cache::JiftyDBI::Stash::Publisher->new(handle => $STASH->handle);
$publisher->create( channel => $chan, name => $pub, idx => 0);
}
sub remove_publisher {
my ($self, $chan, $pub) = @_;
my $publisher = _get_publisher($chan => $pub);
$publisher->delete();
}
sub get_index {
my ($self, $chan, $pub) = @_;
my $publisher = _get_publisher($chan => $pub);
if ($publisher->id) {
return $publisher->idx
}
}
sub set_index {
my ($self, $chan, $pub, $idx) = @_;
return _get_publisher($chan => $pub)->set_idx($idx);
}
sub _get_publisher {
my $chan = shift;
my $pub = shift;
my $publisher = IPC::PubSub::Cache::JiftyDBI::Stash::Publisher->new(handle => $STASH->handle);
$publisher->load_by_cols( channel => $chan, name => $pub);
return $publisher;
}
1;
|
