Fix mypy errors: Observer, Scheduler, PriorityQueue, SchedulerItem, AsyncSubject (#411)

* fix PriorityQueue: correct items type hint

* fix ScheduledItem: relax __eq__ type hints

* fix Observer: remove monkey patch of _xxx_core functions by wrapping the given handler functions in __init__

* fix AsyncSubject: add type hints + set the list of observers to an empty list on dispose

* Observer: use default on_xxx function exposed in rx.core.internal instead

Author: jcafhe

Diff for: rx/core/observer/observer.py

@@ -1,6 +1,7 @@
 from typing import Any, Callable, Optional
 
 from .. import typing
+from rx.internal import noop, default_error
 
 
 class Observer(typing.Observer, typing.Disposable):
@@ -15,20 +16,20 @@ def __init__(self,
                  on_completed: Optional[typing.OnCompleted] = None
                  ) -> None:
         self.is_stopped = False
-        if on_next is not None:
-            self._on_next_core = on_next
-        if on_error is not None:
-            self._on_error_core = on_error
-        if on_completed is not None:
-            self._on_completed_core = on_completed
+        self._handler_on_next = on_next or noop
+        self._handler_on_error = on_error or default_error
+        self._handler_on_completed = on_completed or noop
 
     def on_next(self, value: Any) -> None:
         """Notify the observer of a new element in the sequence."""
         if not self.is_stopped:
             self._on_next_core(value)
 
     def _on_next_core(self, value: Any) -> None:
-        pass
+        """For Subclassing purpose. This method is called by `on_next()`
+        method until the observer is stopped.
+        """
+        self._handler_on_next(value)
 
     def on_error(self, error: Exception) -> None:
         """Notify the observer that an exception has occurred.
@@ -42,10 +43,10 @@ def on_error(self, error: Exception) -> None:
             self._on_error_core(error)
 
     def _on_error_core(self, error: Exception) -> None:
-        if isinstance(error, BaseException):
-            raise error
-        else:
-            raise Exception(error)
+        """For Subclassing purpose. This method is called by `on_error()`
+        method until the observer is stopped.
+        """
+        self._handler_on_error(error)
 
     def on_completed(self) -> None:
         """Notifies the observer of the end of the sequence."""
@@ -55,7 +56,10 @@ def on_completed(self) -> None:
             self._on_completed_core()
 
     def _on_completed_core(self) -> None:
-        pass
+        """For Subclassing purpose. This method is called by `on_completed()`
+        method until the observer is stopped.
+        """
+        self._handler_on_completed()
 
     def dispose(self) -> None:
         """Disposes the observer, causing it to transition to the

Diff for: rx/internal/priorityqueue.py

@@ -1,6 +1,6 @@
 import heapq
 from sys import maxsize
-from typing import Generic, List
+from typing import Generic, List, Tuple
 
 from rx.core.typing import T1
 
@@ -11,7 +11,7 @@ class PriorityQueue(Generic[T1]):
     MIN_COUNT = ~maxsize
 
     def __init__(self) -> None:
-        self.items: List[T1] = []
+        self.items: List[Tuple[T1, int]] = []
         self.count = PriorityQueue.MIN_COUNT  # Monotonic increasing for sort stability
 
     def __len__(self):

Diff for: rx/scheduler/scheduleditem.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Generic, Optional
+from typing import Generic, Optional, Any
 
 from rx.core import typing
 from rx.disposable import SingleAssignmentDisposable
@@ -40,5 +40,8 @@ def __lt__(self, other: 'ScheduledItem') -> bool:
     def __gt__(self, other: 'ScheduledItem') -> bool:
         return self.duetime > other.duetime
 
-    def __eq__(self, other: 'ScheduledItem') -> bool:
-        return self.duetime == other.duetime
+    def __eq__(self, other: Any) -> bool:
+        try:
+            return self.duetime == other.duetime
+        except AttributeError:
+            return NotImplemented

Diff for: rx/subjects/asyncsubject.py

@@ -1,5 +1,5 @@
 import threading
-from typing import Any
+from typing import Any, List, Optional
 
 from rx.disposable import Disposable
 from rx.core import Observable
@@ -24,16 +24,16 @@ def __init__(self) -> None:
         self.is_stopped = False
         self.value = None
         self.has_value = False
-        self.observers = []
-        self.exception = None
+        self.observers: List[Observer] = []
+        self.exception: Optional[Exception] = None
 
         self.lock = threading.RLock()
 
     def check_disposed(self) -> None:
         if self.is_disposed:
             raise DisposedException()
 
-    def _subscribe_core(self, observer, scheduler=None):
+    def _subscribe_core(self, observer: Observer, scheduler=None):
         with self.lock:
             self.check_disposed()
             if not self.is_stopped:
@@ -103,6 +103,6 @@ def on_next(self, value: Any) -> None:
     def dispose(self) -> None:
         with self.lock:
             self.is_disposed = True
-            self.observers = None
+            self.observers = []
             self.exception = None
             self.value = None