RDBC-730 Improve Changes API

Author: poissoncorp

README.md

@@ -894,7 +894,7 @@ all_documents_changes = []
 # Subscribe for all documents, or for specific collection (or other database items)
 all_observer = self.store.changes().for_all_documents()
 
-close_method = all_observer.subscribe(ActionObserver(on_next=all_documents_changes.append))
+close_method = all_observer.subscribe_with_observer(ActionObserver(on_next=all_documents_changes.append))
 all_observer.ensure_subscribe_now()
 
 session = store.open_session()

ravendb/changes/observers.py

@@ -1,9 +1,9 @@
-from abc import ABCMeta, abstractmethod
+from __future__ import annotations
+from concurrent.futures import Future
+from threading import Lock
 from typing import Callable, Generic, TypeVar
 
 from ravendb.tools.concurrentset import ConcurrentSet
-from concurrent.futures import Future
-from threading import Lock
 
 _T_Change = TypeVar("_T_Change")
 
@@ -32,21 +32,36 @@ def on_document_change_notification(self, value: _T_Change):
         except Exception as e:
             self.error(e)
 
-    def subscribe(self, observer) -> Callable[[], None]:
+    def subscribe(self, on_next_callback: Callable[[], None]) -> Callable[[], None]:
         """
-        @param Observer or func observer: The observer that will do action when changes happens
+        @param func on_next_callback: The action that observer will do when changes happens
         :return: method that close the subscriber
         """
         self.inc()
-        if not isinstance(observer, Observer):
-            observer = ActionObserver(on_next=observer)
+        observer = ActionObserver(on_next=on_next_callback)
         self._subscribers.add(observer)
 
-        def close_action():
+        def close_action() -> None:
             self.dec()
             self._subscribers.remove(observer)
-            if "on_complete" in observer.__dict__ and "__call__" in observer.on_complete.__dict__:
-                observer.on_complete()
+            if observer.on_completed_callback is not None:
+                observer.on_completed()
+
+        return close_action
+
+    def subscribe_with_observer(self, observer: ActionObserver) -> Callable[[], None]:
+        """
+        @param Observer observer: The observer that will do action when changes happens
+        :return: method that close the subscriber
+        """
+        self.inc()
+        self._subscribers.add(observer)
+
+        def close_action() -> None:
+            self.dec()
+            self._subscribers.remove(observer)
+            if observer.on_completed_callback is not None:
+                observer.on_completed()
 
         return close_action
 
@@ -106,38 +121,28 @@ def send(self, msg: _T_Change):
             return
 
 
-class Observer(metaclass=ABCMeta):
-    @abstractmethod
-    def on_completed(self) -> None:
-        pass
-
-    @abstractmethod
-    def on_error(self, exception: Exception) -> None:
-        pass
-
-    @abstractmethod
-    def on_next(self, value: _T_Change) -> None:
-        pass
-
-
-class ActionObserver(Observer):
+class ActionObserver:
     def __init__(
         self,
         on_next: Callable[[_T_Change], None],
         on_error: Callable[[Exception], None] = None,
         on_completed: Callable[[], None] = None,
     ):
-        self._on_next = on_next
-        self._on_error = on_error
-        self._on_completed = on_completed
+        self._on_next_callback = on_next
+        self._on_error_callback = on_error
+        self._on_completed_callback = on_completed
 
     def on_next(self, value: _T_Change) -> None:
-        self._on_next(value)
+        self._on_next_callback(value)
 
     def on_error(self, exception: Exception) -> None:
-        if self._on_error:
-            self._on_error(exception)
+        if self._on_error_callback:
+            self._on_error_callback(exception)
 
     def on_completed(self) -> None:
-        if self._on_completed:
-            self._on_completed()
+        if self._on_completed_callback:
+            self._on_completed_callback()
+
+    @property
+    def on_completed_callback(self):
+        return self._on_completed_callback

ravendb/tests/database_changes/test_database_changes.py

@@ -35,7 +35,7 @@ def on_next(value):
                 event.set()
 
         all_observer = self.store.changes().for_all_documents()
-        close_method = all_observer.subscribe(ActionObserver(on_next=on_next))
+        close_method = all_observer.subscribe_with_observer(ActionObserver(on_next=on_next))
         all_observer.ensure_subscribe_now()
         with self.store.open_session() as session:
             session.store(User("Idan"))

ravendb/tests/jvm_migrated_tests/issues_tests/test_ravenDB_11703.py

@@ -61,7 +61,7 @@ def test_can_get_notification_about_counter_delete(self):
             session.counters_for("users/1").increment("likes")
             session.save_changes()
 
-        counter_change: CounterChange = changes_queue.get(timeout=20)
+        counter_change: CounterChange = changes_queue.get(timeout=10)
         self.assertIsNotNone(counter_change)
 
         self.assertEqual("users/1", counter_change.document_id)
@@ -75,7 +75,7 @@ def test_can_get_notification_about_counter_delete(self):
             session.counters_for("users/1").delete("likes")
             session.save_changes()
 
-        counter_change = changes_queue.get(timeout=20)
+        counter_change = changes_queue.get(timeout=10)
         self.assertIsNotNone(counter_change)
 
         self.assertEqual("users/1", counter_change.document_id)

ravendb/tests/jvm_migrated_tests/server_tests/documents/notifications/test_changes.py

@@ -42,8 +42,8 @@ def __ev(value: DocumentChange):
             document_change = value
             event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         with self.store.open_session() as session:
@@ -81,8 +81,8 @@ def __ev(value: DocumentChange):
             if len(changes_list) == 2:
                 event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         with self.store.open_session() as session:
@@ -117,8 +117,8 @@ def __ev(value: DocumentChange):
             document_change = value
             event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         with self.store.open_session() as session:
@@ -160,8 +160,8 @@ def __ev(value: IndexChange):
             index_change = value
             event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         operation = SetIndexesPriorityOperation(IndexPriority.LOW, index.index_name)
@@ -184,8 +184,8 @@ def __ev(value: IndexChange):
             index_change = value
             event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         operation = SetIndexesPriorityOperation(IndexPriority.LOW, index.index_name)
@@ -208,8 +208,8 @@ def __ev(value: DocumentChange):
             if len(document_changes) == 2:
                 event.set()
 
-        subscription = ActionObserver(__ev)
-        close_action = observable.subscribe(subscription)
+        observer = ActionObserver(__ev)
+        close_action = observable.subscribe_with_observer(observer)
         observable.ensure_subscribe_now()
 
         with self.store.open_session() as session: