observersubject_numberproducer.py

Go to the documentation of this file.

 
 
from abc import ABC, abstractmethod
 
 
 
class IObserverNumberChanged(ABC):
    
    @abstractmethod
    def NumberChanged(self) -> None:
        pass
 
 
 
 
 
 
class IEventNotifications(ABC):
    
    @abstractmethod
    def SubscribeToNumberChanged(self, observer : IObserverNumberChanged) -> None:
        pass
 
    
    @abstractmethod
    def UnsubscribeFromNumberChanged(self, observer : IObserverNumberChanged) -> None:
        pass
 
 
 
 
 
 
class INumberProducer(ABC):
 
    
    @abstractmethod
    def Update(self) -> None:
        pass
 
    
    def FetchNumber(self) -> int:
        pass
 
 
 
 
 
 
class ObserverSubject_NumberProducer(IEventNotifications, INumberProducer):
 
    
    def _NotifyNumberChanged(self) -> None:
        # Copy the list so observers can change the original observers
        # during the notification (this isn't strictly needed in this
        # example but it is good practice for any notification system
        # that handles multiple observers where multiple threads might
        # be in play or observers can unsubscribe at any time, even in
        # the event notification).
        observers = self._observers.copy()
        for observer in observers:
            observer.NumberChanged();
 
 
    
    def _FindObserver(self, observer : IObserverNumberChanged) -> int:
        foundIndex = -1
        try:
            foundIndex = self._observers.index(observer)
        except ValueError:
            # observer was not found
            pass
        return foundIndex
 
 
    
    def _ContainsObserver(self, observer : IObserverNumberChanged) -> bool:
        foundObserver = self._FindObserver(observer)
        return foundObserver != -1
 
 
    
    def __init__(self) -> None:
        self._observers = [] # type: list[IObserverNumberChanged]
        self._number = 0
 
    
 
 
    
    def Update(self) -> None:
        self._number += 1
        self._NotifyNumberChanged()
 
 
    #++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    # Implementation of the INumberProducer interface.
    #++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
    
    def FetchNumber(self) -> int:
        return self._number
 
 
    #++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    # Implementation of the IEventNotifications interface.
    #++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
    
    def SubscribeToNumberChanged(self, observer : IObserverNumberChanged) -> None:
        if not self._ContainsObserver(observer):
            self._observers.append(observer)
 
 
    
    def UnsubscribeFromNumberChanged(self, observer : IObserverNumberChanged) -> None:
        if self._ContainsObserver(observer):
            self._observers.remove(observer)