source: geniuswebcore/geniusweb/party/DefaultParty.py@ 99

import threading
from typing import TypeVar, List, cast, Optional

from tudelft.utilities.listener.Listener import Listener
from tudelft_utilities_logging.ReportToLogger import ReportToLogger
from tudelft_utilities_logging.Reporter import Reporter

from geniusweb.actions.Action import Action
from geniusweb.connection.ConnectionEnd import ConnectionEnd
from geniusweb.inform.Inform import Inform
from geniusweb.party.Party import Party



class DefaultParty (Party, Listener[Inform]):
        '''
        Party with default implementation to handle the connection.
        '''

        def __init__(self, reporter:Reporter=None):
                '''
                @param reporter the reporter to use. Generated automatically if None
                '''
                if not reporter:
                        reporter = ReportToLogger(self.__class__.__name__)
                if not isinstance(reporter, Reporter):
                        raise ValueError("reporter must be instance of reporter, got "+str(reporter))
                self._reporter=reporter
                self._connection: Optional[ConnectionEnd[Inform,Action]] = None
                self._lock = threading.Lock()

        def connect(self, connection: ConnectionEnd[Inform, Action] ):
                try:
                        self._lock.acquire()
                        if self._connection:
                                raise ValueError("Already connected")
                        self._connection = connection
                        self._connection.addListener(self)
                finally:
                        self._lock.release()

        #Override
        def disconnect(self):
                self._lock.acquire()
                if self._connection:
                        self._connection.removeListener(self)
                        self._connection.close()
                        self._connection = None #type:ignore
                self._lock.release()
        

        #Override
        def terminate(self):
                self.disconnect();
        

        def getConnection(self) -> Optional[ConnectionEnd[Inform, Action]]:
                '''
                @return currently available connection, or null if not currently
        connected.
                '''
                return self._connection;
        

        def getReporter(self)->Reporter:
                return self._reporter;