lib/DBIx/Data/Collection/Set.pm

   1 package DBIx::Data::Collection::Set;
   2
   3 use Moose;
   4 use Method::Signatures::Simple;
   5 use Data::Perl::Stream::Array;
   6 use Scalar::Util qw(weaken refaddr);
   7
   8 has _store => (is => 'ro', required => 1, init_arg => 'store');
   9
  10 has _class => (is => 'ro', predicate => '_has_class', init_arg => 'class');
  11
  12 has _set_over => (is => 'ro', required => 1, init_arg => 'set_over');
  13
  14 ## member cache (all members)
  15
  16 has _member_cache => (
  17   is => 'ro', lazy_build => 1,
  18   predicate => '_member_cache_built',
  19   writer => '_set_member_cache',
  20 );
  21
  22 method _build__member_cache {
  23   my $stream = $self->_new_raw_stream;
  24   my @cache;
  25   while (my ($raw) = $stream->next) {
  26     my $obj = do {
  27       if (my ($obj) = $self->_key_cache_get_raw($raw)) {
  28         # can't just $self->_merge($obj, $raw) since $obj might have changed
  29         $self->_refresh($obj, $raw)
  30       } else {
  31         $self->_add_to_key_cache($self->_inflate($raw))
  32       }
  33     };
  34     push @cache, $obj;
  35   }
  36   $self->_notify_observers(all_members => \@cache);
  37   \@cache
  38 }
  39
  40 method _add_to_member_cache ($to_add) {
  41   return $to_add unless $self->_member_cache_built;
  42   push @{$self->_member_cache}, $to_add;
  43   $to_add
  44 }
  45
  46 method _remove_from_member_cache ($to_remove) {
  47   return $to_remove unless $self->_member_cache_built;
  48   @{$self->_member_cache} = grep $_ ne $to_remove, @{$self->_member_cache};
  49   $to_remove
  50 }
  51
  52 ## key cache - by primary/unique key
  53
  54 has _key_cache => (is => 'ro', default => sub { {} });
  55
  56 method _add_to_key_cache ($to_add) {
  57   $self->_key_cache->{$self->_object_to_id($to_add)} = $to_add;
  58   $to_add
  59 }
  60
  61 method _remove_from_key_cache ($to_remove) {
  62   # should return $to_remove
  63   delete $self->_key_cache->{$self->_object_to_id($to_remove)}
  64 }
  65
  66 method _key_cache_has_raw ($raw) {
  67   exists $self->_key_cache->{$self->_raw_to_id($raw)}
  68 }
  69
  70 method _key_cache_has_object ($obj) {
  71   exists $self->_key_cache->{$self->_object_to_id($obj)}
  72 }
  73
  74 method _key_cache_get_raw ($raw) {
  75   $self->_key_cache_get_id($self->_raw_to_id($raw))
  76 }
  77
  78 method _key_cache_get_object ($obj) {
  79   $self->_key_cache_get_id($self->_object_to_id($obj))
  80 }
  81
  82 method _key_cache_get_object_spec ($spec) {
  83   # see _object_spec_to_id for doc of what the difference is
  84   $self->_key_cache_get_id($self->_object_spec_to_id($spec))
  85 }
  86
  87 method _key_cache_get_id ($id) {
  88   exists $self->_key_cache->{$id}
  89     ? ($self->_key_cache->{$id})
  90     : ()
  91 }
  92
  93 method _all_key_cache_members {
  94   values %{$self->_key_cache}
  95 }
  96
  97 method _set_key_cache_members ($members) {
  98   %{$self->_key_cache} = (map +($self->_object_to_id($_) => $_), @$members);
  99   return
 100 }
 101
 102 ## observers
 103
 104 has _observer_callbacks => (
 105   is => 'ro', default => sub { {} },
 106 );
 107
 108 method _notify_observers ($event, $payload) {
 109   my $oc = $self->_observer_callbacks;
 110   foreach my $refaddr (keys %$oc) {
 111     my ($obj, $cb) = @{$oc->{$refaddr}};
 112     unless (defined $obj) { # weak ref was garbage collected
 113       delete $oc->{$refaddr};
 114       next;
 115     }
 116     $obj->$cb($self, $event, $payload);
 117   }
 118   $payload
 119 }
 120
 121 method _register_observer ($obj, $cb) {
 122   my $entry = [ $obj, $cb ];
 123   weaken($entry->[0]);
 124   $self->_observer_callbacks->{refaddr($obj)} = $entry;
 125   return
 126 }
 127
 128 method _setup_observation_of ($other) {
 129   $other->_register_observer($self, method ($from, $event, $payload) {
 130     if ($event eq 'add' or $event eq 'get') {
 131       $self->_add_to_caches($payload);
 132     } elsif ($event eq 'remove') {
 133       $self->_remove_from_caches($payload);
 134     } elsif ($event eq 'all_members') {
 135       # separate arrayref since future add will trigger push()
 136       $self->_set_caches([ @$payload ]);
 137     }
 138   });
 139   return
 140 }
 141
 142 ## thunking between the store representation and the set representation
 143 #
 144 # _inflate is raw data -> final repr
 145 # _deflate is final repr -> raw data
 146 # _merge takes final repr + raw data and updates the repr
 147 #    (this is used for pk-generated values and later lazy loading)
 148 #
 149 # _deflate_spec is attributes of final repr -> raw data
 150 # _merge_spec is final repr + extra attributes and update repr
 151
 152 method _inflate ($raw) {
 153   bless($raw, $self->_class) if $self->_has_class;
 154   $raw
 155 }
 156
 157 method _deflate ($obj) {
 158   +{ %$obj }
 159 }
 160
 161 method _merge ($obj, $raw) {
 162   @{$obj}{keys %$raw} = values %$raw;
 163   $obj
 164 }
 165
 166 method _refresh ($obj, $raw) {
 167   # if $obj has been changed but not flushed we'd destroy data doing
 168   # a blind merge - but if $obj has change tracking of some sort then
 169   # we -could- do something safely, so this method exists to be mangled
 170   # by subclasses
 171   $obj
 172 }
 173
 174 method _deflate_spec ($spec) {
 175   $spec
 176 }
 177
 178 method _merge_spec ($obj, $spec) {
 179   @{$obj}{keys %$spec} = values %$spec;
 180   $obj
 181 }
 182
 183 ## methods to get ids
 184
 185 method _raw_to_id ($raw) {
 186   # XXX must escape this. or do something else.
 187   join ';', map $raw->{$_}, @{$self->_set_over}
 188 }
 189
 190 method _object_to_id ($obj) {
 191   $self->_raw_to_id($self->_deflate($obj))
 192 }
 193
 194 method _object_spec_to_id ($spec) {
 195   # intentionally C&P from _raw_to - this is not the same thing. If a column
 196   # were mapped to an attribute of a different name, the raw would have the
 197   # column name as a key but an object spec would have the attribute name
 198   join ';', map $spec->{$_}, @{$self->_set_over}
 199 }
 200
 201 ## array-ish operations - i.e. get all members
 202
 203 method _new_raw_stream {
 204   $self->_store->new_select_command({})->execute
 205 }
 206
 207 method flatten {
 208   @{$self->_member_cache};
 209 }
 210
 211 method as_stream {
 212   Data::Perl::Stream::Array->new(array => $self->_member_cache);
 213 }
 214
 215 method _set_caches ($members) {
 216   $self->_set_member_cache($members);
 217   $self->_set_key_cache_members($members);
 218   return
 219 }
 220
 221 ## load single row
 222
 223 method get ($spec) {
 224   if (my ($got) = $self->_key_cache_get_object_spec($spec)) {
 225     return $got
 226   }
 227   if (my ($raw) = $self->_get_from_store($self->_deflate_spec($spec))) {
 228     return $self->_notify_observers(
 229       get => $self->_add_to_key_cache($self->_inflate($raw))
 230     );
 231   }
 232   return undef # we aren't handling cache misses here yet
 233 }
 234
 235 method _get_from_store ($raw) {
 236   $self->_store->new_select_single_command($raw)->execute
 237 }
 238
 239 ## add member
 240
 241 method add ($new) {
 242   $self->_add_to_store($new);
 243   $self->_add_to_caches($new);
 244   $self->_notify_observers(add => $new);
 245   $new
 246 }
 247
 248 method _add_to_store ($new) {
 249   my $new_raw = $self->_deflate($new);
 250   $self->_merge($new, $self->_store->new_insert_command($new_raw)->execute);
 251   $new
 252 }
 253
 254 method _add_to_caches ($new) {
 255   $self->_add_to_member_cache($new);
 256   $self->_add_to_key_cache($new);
 257   $new
 258 }
 259
 260 ## remove member
 261
 262 method remove ($old) {
 263   $self->_remove_from_store($old);
 264   $self->_remove_from_caches($old);
 265   $self->_notify_observers(remove => $old);
 266   $old
 267 }
 268
 269 method _remove_from_store ($old) {
 270   $self->_store->new_delete_single_command($self->_deflate($old))->execute
 271 }
 272
 273 method _remove_from_caches ($old) {
 274   $self->_remove_from_member_cache($old);
 275   $self->_remove_from_key_cache($old);
 276   $old
 277 }
 278
 279 ## update member
 280
 281 method _update_in_store ($obj) {
 282   # this is currently a call command but we should think about it
 283   # being a row command so that we can have RETURNING or other
 284   # mechanisms handle things like set-on-update datetime values
 285   $self->_store->new_update_single_command($self->_deflate($obj))->execute
 286 }
 287
 288 # I do wonder if we needed _merge_spec or if we'd be better off with
 289 # just using the raw merge routine ...
 290
 291 method _update_set_in_store ($spec) {
 292   $self->_store->new_update_command($self->_deflate_spec($spec))->execute;
 293   if ($self->_member_cache_built) {
 294     my $cache = $self->_member_cache;
 295     foreach my $obj (@{$cache}) {
 296       $self->_merge_spec($obj, $spec);
 297     }
 298     $self->_notify_observers(all_members => $cache);
 299   }
 300   return
 301 }
 302
 303 method _remove_set_from_store {
 304   $self->_store->new_delete_command->execute;
 305   $self->_set_caches([]);
 306   $self->_notify_observers(all_members => []);
 307   return
 308 }
 309
 310 1;