Observer sets now extend Synchronization::Object

Author: jdantonio

lib/concurrent/atomic/copy_on_notify_observer_set.rb

@@ -1,15 +1,12 @@
+require 'concurrent/synchronization'
+
 module Concurrent
 
   # A thread safe observer set implemented using copy-on-read approach:
   # observers are added and removed from a thread safe collection; every time
   # a notification is required the internal data structure is copied to
   # prevent concurrency issues
-  class CopyOnNotifyObserverSet
-
-    def initialize
-      @mutex = Mutex.new
-      @observers = {}
-    end
+  class CopyOnNotifyObserverSet < Synchronization::Object
 
     # Adds an observer to this set. If a block is passed, the observer will be
     # created by this method and no other params should be passed
@@ -30,41 +27,33 @@ def add_observer(observer=nil, func=:update, &block)
         func = :call
       end
 
-      begin
-        @mutex.lock
+      synchronize do
         @observers[observer] = func
         observer
-      ensure
-        @mutex.unlock
       end
     end
 
     # @param [Object] observer the observer to remove
     # @return [Object] the deleted observer
     def delete_observer(observer)
-      @mutex.lock
-      @observers.delete(observer)
-      observer
-    ensure
-      @mutex.unlock
+      synchronize do
+        @observers.delete(observer)
+        observer
+      end
     end
 
     # Deletes all observers
     # @return [CopyOnWriteObserverSet] self
     def delete_observers
-      @mutex.lock
-      @observers.clear
-      self
-    ensure
-      @mutex.unlock
+      synchronize do
+        @observers.clear
+        self
+      end
     end
 
     # @return [Integer] the observers count
     def count_observers
-      @mutex.lock
-      @observers.count
-    ensure
-      @mutex.unlock
+      synchronize { @observers.count }
     end
 
     # Notifies all registered observers with optional args
@@ -86,23 +75,24 @@ def notify_and_delete_observers(*args, &block)
       self
     end
 
+    protected
+
+    def ns_initialize
+      @observers = {}
+    end
+
     private
 
     def duplicate_and_clear_observers
-      @mutex.lock
-      observers = @observers.dup
-      @observers.clear
-      observers
-    ensure
-      @mutex.unlock
+      synchronize do
+        observers = @observers.dup
+        @observers.clear
+        observers
+      end
     end
 
     def duplicate_observers
-      @mutex.lock
-      observers = @observers.dup
-      observers
-    ensure
-      @mutex.unlock
+      synchronize { observers = @observers.dup }
     end
 
     def notify_to(observers, *args)

lib/concurrent/atomic/copy_on_write_observer_set.rb

@@ -1,14 +1,11 @@
+require 'concurrent/synchronization'
+
 module Concurrent
 
   # A thread safe observer set implemented using copy-on-write approach:
   # every time an observer is added or removed the whole internal data structure is
   # duplicated and replaced with a new one.
-  class CopyOnWriteObserverSet
-
-    def initialize
-      @mutex = Mutex.new
-      @observers = {}
-    end
+  class CopyOnWriteObserverSet < Synchronization::Object
 
     # Adds an observer to this set
     # If a block is passed, the observer will be created by this method and no
@@ -29,27 +26,23 @@ def add_observer(observer=nil, func=:update, &block)
         func = :call
       end
 
-      begin
-        @mutex.lock
+      synchronize do
         new_observers = @observers.dup
         new_observers[observer] = func
         @observers = new_observers
         observer
-      ensure
-        @mutex.unlock
       end
     end
 
     # @param [Object] observer the observer to remove
     # @return [Object] the deleted observer
     def delete_observer(observer)
-      @mutex.lock
-      new_observers = @observers.dup
-      new_observers.delete(observer)
-      @observers = new_observers
-      observer
-    ensure
-      @mutex.unlock
+      synchronize do
+        new_observers = @observers.dup
+        new_observers.delete(observer)
+        @observers = new_observers
+        observer
+      end
     end
 
     # Deletes all observers
@@ -59,7 +52,6 @@ def delete_observers
       self
     end
 
-
     # @return [Integer] the observers count
     def count_observers
       observers.count
@@ -83,6 +75,12 @@ def notify_and_delete_observers(*args, &block)
       self
     end
 
+    protected
+
+    def ns_initialize
+      @observers = {}
+    end
+
     private
 
     def notify_to(observers, *args)
@@ -94,26 +92,19 @@ def notify_to(observers, *args)
     end
 
     def observers
-      @mutex.lock
-      @observers
-    ensure
-      @mutex.unlock
+      synchronize { @observers }
     end
 
     def observers=(new_set)
-      @mutex.lock
-      @observers = new_set
-    ensure
-      @mutex.unlock
+      synchronize { @observers = new_set }
     end
 
     def clear_observers_and_return_old
-      @mutex.lock
-      old_observers = @observers
-      @observers = {}
-      old_observers
-    ensure
-      @mutex.unlock
+      synchronize do
+        old_observers = @observers
+        @observers = {}
+        old_observers
+      end
     end
   end
 end