indexed by

[dbsrgits/DBIx-Data-Store-old.git] / lib / DBIx / Data / Collection / Set.pm

diff --git a/lib/DBIx/Data/Collection/Set.pm b/lib/DBIx/Data/Collection/Set.pm
index ceb1e3f..2672dd4 100644 (file)
--- a/lib/DBIx/Data/Collection/Set.pm
+++ b/lib/DBIx/Data/Collection/Set.pm
@@ -3,18 +3,20 @@ package DBIx::Data::Collection::Set;
 use Moose;
 use Method::Signatures::Simple;
 use Data::Perl::Stream::Array;
+use Scalar::Util qw(weaken refaddr);
 
 has _store => (is => 'ro', required => 1, init_arg => 'store');
 
-has _class => (is => 'ro', predicate => '_has_class');
+has _class => (is => 'ro', predicate => '_has_class', init_arg => 'class');
 
 has _set_over => (is => 'ro', required => 1, init_arg => 'set_over');
 
 ## member cache (all members)
 
 has _member_cache => (
-  is => 'rw', lazy_build => 1,
+  is => 'ro', lazy_build => 1,
   predicate => '_member_cache_built',
+  writer => '_set_member_cache',
 );
 
 method _build__member_cache {
@@ -23,13 +25,15 @@ method _build__member_cache {
   while (my ($raw) = $stream->next) {
     my $obj = do {
       if (my ($obj) = $self->_key_cache_get_raw($raw)) {
-        $self->_merge($obj, $raw)
+        # can't just $self->_merge($obj, $raw) since $obj might have changed
+        $self->_refresh($obj, $raw)
       } else {
         $self->_add_to_key_cache($self->_inflate($raw))
       }
     };
     push @cache, $obj;
   }
+  $self->_notify_observers(all_members => \@cache);
   \@cache
 }
 
@@ -86,6 +90,50 @@ method _key_cache_get_id ($id) {
     : ()
 }
 
+method _all_key_cache_members {
+  values %{$self->_key_cache}
+}
+
+## observers
+
+has _observer_callbacks => (
+  is => 'ro', default => sub { {} },
+);
+
+method _notify_observers ($event, $payload) {
+  my $oc = $self->_observer_callbacks;
+  foreach my $refaddr (keys %$oc) {
+    my ($obj, $cb) = @{$oc->{$refaddr}};
+    unless (defined $obj) { # weak ref was garbage collected
+      delete $oc->{$refaddr};
+      next;
+    }
+    $obj->$cb($self, $event, $payload);
+  }
+  $payload
+}
+
+method _register_observer ($obj, $cb) {
+  my $entry = [ $obj, $cb ];
+  weaken($entry->[0]);
+  $self->_observer_callbacks->{refaddr($obj)} = $entry;
+  return
+}
+
+method _setup_observation_of ($other) {
+  $other->_register_observer($self, method ($from, $event, $payload) {
+    if ($event eq 'add' or $event eq 'get') {
+      $self->_add_to_caches($payload);
+    } elsif ($event eq 'remove') {
+      $self->_remove_from_caches($payload);
+    } elsif ($event eq 'all_members') {
+      # separate arrayref since future add will trigger push()
+      $self->_set_member_cache([ @$payload ]);
+    }
+  });
+  return
+}
+
 ## thunking between the store representation and the set representation
 #
 # _inflate is raw data -> final repr
@@ -109,6 +157,14 @@ method _merge ($obj, $raw) {
   $obj
 }
 
+method _refresh ($obj, $raw) {
+  # if $obj has been changed but not flushed we'd destroy data doing
+  # a blind merge - but if $obj has change tracking of some sort then
+  # we -could- do something safely, so this method exists to be mangled
+  # by subclasses
+  $obj
+}
+
 method _deflate_spec ($spec) {
   $spec
 }
@@ -152,7 +208,9 @@ method get ($spec) {
     return $got
   }
   if (my ($raw) = $self->_get_from_store($self->_deflate_spec($spec))) {
-    return $self->_add_to_key_cache($self->_inflate($raw))
+    return $self->_notify_observers(
+      get => $self->_add_to_key_cache($self->_inflate($raw))
+    );
   }
   return undef # we aren't handling cache misses here yet
 }
@@ -166,6 +224,7 @@ method _get_from_store ($raw) {
 method add ($new) {
   $self->_add_to_store($new);
   $self->_add_to_caches($new);
+  $self->_notify_observers(add => $new);
   $new
 }
 
@@ -186,6 +245,7 @@ method _add_to_caches ($new) {
 method remove ($old) {
   $self->_remove_from_store($old);
   $self->_remove_from_caches($old);
+  $self->_notify_observers(remove => $old);
   $old
 }