class RemotePersistenceProtocol(metaclass=RequestResponseAPI,
                                error_codes=ERROR_CODES,
                                default_version=b'\x00\x00'):
    ''' Defines the protocol for remote persisters.
    '''
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _postman = weak_property('__postman')
    _salmonator = weak_property('__salmonator')
    
    @fixture_api
    def __init__(self, percore=None, librarian=None, *args, **kwargs):
        super(RemotePersistenceProtocol.__fixture__, self).__init__(
            *args,
            **kwargs
        )
        if percore is not None:
            self._percore = percore
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, percore, librarian, postman, salmonator=None):
        # Link to the remote core.
        self._percore = percore
        self._postman = postman
        self._librarian = librarian
        
        if salmonator is not None:
            self._salmonator = salmonator
        
    @request(b'??')
    async def ping(self, connection):
        ''' Check a remote for availability.
        '''
        return b''
        
    @ping.request_handler
    async def ping(self, connection, body):
        # Really not much to see here.
        return b'\x01'
        
    @ping.response_handler
    async def ping(self, connection, response, exc):
        # This will suppress any errors during pinging.
        if response == b'\x01':
            return True
        else:
            return False
    
    @public_api
    @request(b'PB')
    async def publish(self, connection, packed):
        ''' Publish a packed Golix object.
        '''
        return packed
        
    @publish.fixture
    async def publish(self, connection, packed):
        ''' Just slap the thing into librarian with no checking for
        fixtures.
        '''
        obj = await self._percore.attempt_load(packed)
        await self._librarian.store(obj, packed)
        
    @publish.request_handler
    async def publish(self, connection, body):
        ''' Handle a published object.
        '''
        obj = await self._percore.ingest(
            packed = body,
            remotable = False,
            skip_conn = weakref.ref(connection)
        )
        
        # Object already existed.
        if obj is None:
            response = b'\x00'
        
        # Object is new
        else:
            response = b'\x01'
            
        # We don't need to wait for the mail run to have a successful return
        return response
        
    @publish.response_handler
    async def publish(self, connection, response, exc):
        ''' Handle responses to publish requests.
        '''
        if exc is not None:
            raise exc
        else:
            return True
    
    @public_api
    @request(b'GT')
    async def get(self, connection, ghid):
        ''' Request an object from the persistence provider.
        '''
        return bytes(ghid)
        
    @get.fixture
    async def get(self, connection, ghid):
        ''' Fixture to just pull directly from librarian.
        '''
        return (await self._librarian.retrieve(ghid))
        
    @get.request_handler
    async def get(self, connection, body):
        ''' Handle get requests.
        '''
        ghid = Ghid.from_bytes(body)
        return (await self._librarian.retrieve(ghid))
    
    @public_api
    @request(b'+S')
    async def subscribe(self, connection, ghid):
        ''' Subscribe to updates from the remote.
        '''
        return bytes(ghid)
        
    @subscribe.fixture
    async def subscribe(self, connection, ghid):
        ''' Manual noop.
        '''
        
    @subscribe.request_handler
    async def subscribe(self, connection, body):
        ''' Handle subscription requests.
        '''
        ghid = Ghid.from_bytes(body)
        await self._postman.subscribe(connection, ghid)
        return b'\x01'
        
    @request(b'-S')
    async def unsubscribe(self, connection, ghid):
        ''' Unsubscribe from updates at a remote.
        '''
        return bytes(ghid)
        
    @unsubscribe.request_handler
    async def unsubscribe(self, connection, body):
        ''' Handle unsubscription requests.
        '''
        ghid = Ghid.from_bytes(body)
        had_subscription = await self._postman.unsubscribe(connection, ghid)
        
        if had_subscription:
            return b'\x01'
        
        # Still successful, but idempotent
        else:
            return b'\x00'
        
    @unsubscribe.response_handler
    async def unsubscribe(self, connection, response, exc):
        ''' Handle responses to unsubscription requests.
        '''
        if exc is not None:
            raise exc
            
        # For now, ignore (success & UNsub) vs (success & NOsub)
        else:
            return True
    
    @public_api
    @request(b'!!')
    async def subscription_update(self, connection, subscription_ghid,
                                  notification_ghid):
        ''' Send a subscription update to the connection.
        '''
        payload = await self._librarian.retrieve(notification_ghid)
        return bytes(subscription_ghid) + payload
        
    @subscription_update.fixture
    async def subscription_update(self, connection, subscription_ghid,
                                  notification_ghid, timeout=None):
        ''' Make a manual no-op fixture, since inspect signatures
        apparently don't from_callable on a descriptor... (grrr). Also,
        because timeout isn't appropriately wrapped.
        '''
        
    @subscription_update.request_handler
    async def subscription_update(self, connection, body):
        ''' Handles an incoming subscription update.
        '''
        subscribed_ghid = Ghid.from_bytes(body[0:65])
        notification = body[65:]
        
        try:
            # Note that this handles postman scheduling as well.
            ingested = await self._percore.ingest(
                notification,
                remotable = False,
                skip_conn = connection
            )
            
        except AlreadyDebound as exc:
            vomitus = exc.ghid
            debindings = await self._librarian.debind_status(vomitus)
            debinding_ghid = next(debinding for debinding in debindings)
            logger.info(str(subscribed_ghid) +
                        ' subscription update already debound: ' +
                        str(vomitus) + '. Pushing debinding upstream: ' +
                        str(debinding_ghid))
            
            make_background_future(
                self._salmonator.push(debinding_ghid)
            )
            
        else:
            # But, if ingested, we need to notify the salmonator, so it can (if
            # needed) also acquire the target
            if ingested:
                make_background_future(
                    self._salmonator.notify(subscribed_ghid)
                )
        
        return b'\x01'
        
    @request(b'?S')
    async def query_subscriptions(self, connection):
        ''' Request a list of all currently subscribed ghids.
        '''
        return b''
        
    @query_subscriptions.request_handler
    async def query_subscriptions(self, connection, body):
        ''' Handle subscription query requests.
        '''
        ghidlist = list(await self._postman.list_subs(connection))
        parser = generate_ghidlist_parser()
        return parser.pack(ghidlist)
        
    @query_subscriptions.response_handler
    async def query_subscriptions(self, connection, response, exc):
        ''' Handle responses to subscription queries.
        '''
        if exc is not None:
            raise exc
        
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
        
    @request(b'?B')
    async def query_bindings(self, connection, ghid):
        ''' Get a list of all bindings for the ghid.
        '''
        return bytes(ghid)
        
    @query_bindings.request_handler
    async def query_bindings(self, connection, body):
        ''' Handle binding query requests.
        '''
        ghid = Ghid.from_bytes(body)
        ghidlist = await self._librarian.bind_status(ghid)
        parser = generate_ghidlist_parser()
        return parser.pack(list(ghidlist))
        
    @query_bindings.response_handler
    async def query_bindings(self, connection, response, exc):
        ''' Handle responses to binding queries.
        '''
        if exc is not None:
            raise exc
            
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
        
    @request(b'?D')
    async def query_debindings(self, connection, ghid):
        ''' Query which, if any, ghid(s) have debindings for <ghid>.
        '''
        return bytes(ghid)
        
    @query_debindings.request_handler
    async def query_debindings(self, connection, body):
        ''' Handles debinding query requests.
        '''
        ghid = Ghid.from_bytes(body)
        ghidlist = await self._librarian.debind_status(ghid)
        parser = generate_ghidlist_parser()
        return parser.pack(list(ghidlist))
        
    @query_debindings.response_handler
    async def query_debindings(self, connection, response, exc):
        ''' Handle responses to debinding queries.
        '''
        if exc is not None:
            raise exc
            
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
    
    @public_api
    @request(b'?E')
    async def query_existence(self, connection, ghid):
        ''' Checks if the passed <ghid> exists at the remote.
        '''
        return bytes(ghid)
        
    @query_existence.fixture
    async def query_existence(self, connection, ghid):
        ''' Just always return false.
        '''
        return (await self._librarian.contains(ghid))
        
    @query_existence.request_handler
    async def query_existence(self, connection, body):
        ''' Handle existence queries.
        '''
        ghid = Ghid.from_bytes(body)
        if (await self._librarian.contains(ghid)):
            return b'\x01'
        else:
            return b'\x00'
        
    @query_existence.response_handler
    async def query_existence(self, connection, response, exc):
        ''' Handle responses to existence queries.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x00':
            return False
        else:
            return True
        
    @request(b'XX')
    async def disconnect(self, connection):
        ''' Terminates all subscriptions and requests.
        '''
        return b''
        
    @disconnect.request_handler
    async def disconnect(self, connection, body):
        ''' Handle disconnect requests.
        '''
        await self._postman.clear_subs(connection)
        return b'\x01' class Salmonator(loopa.TaskLooper, metaclass=API):
    ''' Responsible for disseminating Golix objects upstream and
    downstream. Handles all comms with them as well.
    '''
    _golcore = weak_property('__golcore')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _remote_protocol = weak_property('__remote_protocol')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Yarp.
        '''
        super().__init__(*args, **kwargs)
        
        self._deferred = ListMap()
        self._clear_q = None
        
        self._upstream_remotes = set()
        self._downstream_remotes = set()
        
        # Lookup for <registered ghid>
        self._registered = set()
        
        self._bootstrapped = False
        
    @fixture_api
    def RESET(self):
        ''' Clear stuff out, yo.
        '''
        self._upstream_remotes.clear()
        self._downstream_remotes.clear()
        self._registered.clear()
        
    def assemble(self, golcore, percore, librarian, remote_protocol):
        self._golcore = golcore
        self._percore = percore
        self._librarian = librarian
        self._remote_protocol = remote_protocol
    
    @fixture_noop
    @public_api
    def add_upstream_remote(self, task_commander, connection_cls, *args,
                            **kwargs):
        ''' Adds an upstream remote persister.
        *args and **kwargs will be passed to the task_commander task.
        '''
        remote = ConnectionManager(
            connection_cls = connection_cls,
            msg_handler = self._remote_protocol,
            conn_init = self.restore_connection,
        )
        # We have to insert the remote before us at the commander, or shutdown
        # will kill the connections before we can clean them up.
        task_commander.register_task(remote, *args, before_task=self, **kwargs)
        self._upstream_remotes.add(remote)
        
    def add_downstream_remote(self, persister):
        ''' Adds a downstream persister.
        
        PersistenceCore will not attempt to keep a consistent state with
        downstream persisters. Instead, it will simply push updates to
        local objects downstream. It will not, however, look to them for
        updates.
        
        Therefore, to create synchronization **between** multiple
        upstream remotes, also add them as downstream remotes.
        '''
        raise NotImplementedError()
        self._downstream_remotes.add(persister)
        
    async def loop_init(self, *args, **kwargs):
        ''' On top of the usual stuff, set up our queues.
        '''
        await super().loop_init(*args, **kwargs)
        # This "must" be unbounded, because we don't have any control over
        # finalization of objects.
        self._clear_q = asyncio.Queue()
        
    async def loop_stop(self, *args, **kwargs):
        ''' On top of the usual stuff, clear our queues.
        '''
        # Would be good to do, but not currently working
        # for remote in self._persisters:
        #     await self._stop_persister(remote)
        
        await super().loop_stop(*args, **kwargs)
        self._clear_q = None
        
        disconnections = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                disconnections.add(
                    make_background_future(remote.disconnect())
                )
        
        # Need to make sure it's not empty
        if disconnections:
            await asyncio.wait(
                fs = disconnections,
                return_when = asyncio.ALL_COMPLETED
            )
        
    async def loop_run(self, *args, **kwargs):
        ''' Wait for object finalizers, and then immediately unsub the
        remotes when the objects are removed from memory.
        '''
        to_clear = await self._clear_q.get()
        await self.deregister(to_clear)
        
    @fixture_noop
    @public_api
    async def notify(self, ghid):
        ''' Notify the salmonator that an upstream subscription
        notification was just successfully completed.
        '''
        # If that was a dynamic object, we also need to pull its target.
        obj = await self._librarian.summarize(ghid)
        if isinstance(obj, _GobdLite):
            if not (await self._librarian.contains(obj.target)):
                await self.pull(obj.target)
    
    @fixture_noop
    @public_api
    async def register(self, ghid):
        ''' Tells the Salmonator to listen upstream for any updates.
        Consumers must independently register an object finalizer for
        salmonator._deregister (not to be called atexit) to remove the
        subscription when the object leaves local memory. TODO: fix that
        leaky abstraction.
        '''
        obj = await self._librarian.summarize(ghid)
        
        if isinstance(obj, _GobdLite):
            logger.info(
                'GAO ' + str(obj.ghid) + ' upstream registration starting.'
            )
            self._registered.add(obj.ghid)
    
            subscriptions = set()
            for remote in self._upstream_remotes:
                if remote.has_connection:
                    subscriptions.add(
                        make_background_future(remote.subscribe(obj.ghid))
                    )
                
            # Need to make sure it's not empty
            if subscriptions:
                await asyncio.wait(
                    fs = subscriptions,
                    return_when = asyncio.ALL_COMPLETED
                )
        
        else:
            logger.debug(
                'GAO ' + str(obj.ghid) + ' cannot be registered upstream: ' +
                'invalid object type: ' + str(type(obj))
            )
    
    @fixture_noop
    @public_api
    def _deregister(self, ghid):
        ''' Finalizer for GAO objects that executes async deregister()
        from within the event loop. Must be called from within our own
        thread, which it should be (finalizers are called from object
        thread, and all gao must be created within event loop's thread).
        '''
        logger.debug('_deregister finalizer called for ' + str(ghid))
        if self._clear_q is not None:
            # This needs to be a function, not a coro, so use nowait.
            self._clear_q.put_nowait(ghid)
    
    @fixture_noop
    @public_api
    async def deregister(self, ghid):
        ''' Tells the salmonator to stop listening for upstream object
        updates.
        '''
        logger.debug('Deregistering updates for ' + str(ghid))
        # This should maybe use remove instead of discard?
        self._registered.discard(ghid)
    
        unsubscriptions = set()
        for remote in self._upstream_remotes:
            unsubscriptions.add(
                make_background_future(remote.unsubscribe(ghid))
            )
        
        # Need to make sure it's not empty
        if unsubscriptions:
            await asyncio.wait(
                fs = unsubscriptions,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def push(self, ghid):
        ''' Push a single ghid to all remotes.
        
        TODO: smartly defer any pushes to unavailable remotes using
        "if remote.has_connection()"
        '''
        data = await self._librarian.retrieve(ghid)
        
        tasks = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                tasks.add(
                    make_background_future(remote.publish(data))
                )
            else:
                self._deferred.append(remote, data)
            
        # Need to make sure it's not empty
        if tasks:
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def pull(self, ghid):
        ''' Gets a ghid from upstream. Returns on the first result. Note
        that this is not meant to be called on a dynamic address, as a
        subs update from a slower remote would always be overridden by
        the faster one.
        '''
        pull_complete = None
        tasks_available = set()
        task_to_remote = {}
        for remote in self._upstream_remotes:
            if remote.has_connection:
                task = asyncio.ensure_future(
                    self._attempt_pull_single(ghid, remote)
                )
                tasks_available.add(task)
                task_to_remote[task] = remote
            
        # Wait until the first successful task completion
        # Note that this also shields us against having no tasks
        while tasks_available and not pull_complete:
            finished, pending = await asyncio.wait(
                fs = tasks_available,
                return_when = asyncio.FIRST_COMPLETED
            )
        
            # Despite FIRST_COMPLETED, asyncio may return more than one task
            for task in finished:
                # The task finished, so discard it from the available and reap
                # any exception
                tasks_available.discard(task)
                exc = task.exception()
                
                # If there's been an exception, continue waiting for the rest.
                # Downgrade this to info, since things may be missing from
                # persisters all the time. Let the parent log if, for example,
                # it's missing everywhere.
                if exc is not None:
                    remote = task_to_remote[task]
                    logger.info(
                        'Error while pulling from remote at ' +
                        remote._conn_desc + ':\n' +
                        ''.join(traceback.format_tb(exc.__traceback__)) +
                        repr(exc)
                    )
                    pull_complete = None
                
                # Completed successfully, but it could be a 404 (or other
                # error), which would present as result() = False.
                # Instead of letting the while loop handle this, since more
                # than one task can complete simultaneously, make sure we don't
                # already have a finalized result before blindly assigning the
                # task's result.
                elif not pull_complete:
                    pull_complete = task.result()
                    
                # Multiple tasks completed at once. An earlier one was
                # successful. We need to grab the result to suppress asyncio
                # complaints.
                else:
                    task.result()
                
        # No dice. Either finished is None (no remotes), None (no successful
        # pulls), or False (exactly one remote had the object, but it was
        # unloadable). Raise.
        if not pull_complete:
            raise UnavailableUpstream(
                'Object was unavailable or unacceptable at all '
                'currently-registered remotes.'
            )
            
        # Log success.
        else:
            logger.debug(
                'Successful remote pull for {!s}. Handling...'.format(ghid)
            )
        
        # We may still have some pending tasks. Cancel them. Note that we have
        # not yielded control to the event loop, so there is no race.
        for task in pending:
            logger.debug('Cancelling pending pulls.')
            task.cancel()
            
        # Now handle the result.
        await self._handle_successful_pull(pull_complete)
            
    async def _handle_successful_pull(self, maybe_obj):
        ''' Dispatches the object in the successful pull.
        
        maybe_obj can be either True (if the object already existed
        locally), or the object itself (if the object was new).
        '''
        # If we got a gobd, make sure its target is in the librarian
        if isinstance(maybe_obj, _GobdLite):
            if not (await self._librarian.contains(maybe_obj.target)):
                # Catch unavailableupstream and log a warning.
                # TODO: add logic to retry a few times and then discard
                try:
                    await self.pull(maybe_obj.target)
                    
                except UnavailableUpstream:
                    logger.warning(
                        'Received a subscription notification for ' +
                        str(maybe_obj.ghid) + ', but the sub\'s target was '
                        'missing both locally and upstream.'
                    )
        
        logger.debug('Successful pull handled.')
    
    @fixture_noop
    @public_api
    async def attempt_pull(self, ghid, quiet=False):
        ''' Grabs the ghid from remotes, if available, and puts it into
        the ingestion pipeline.
        '''
        # TODO: check locally, run _inspect, check if mutable before blindly
        # pulling.
        try:
            await self.pull(ghid)
            
        except UnavailableUpstream:
            if not quiet:
                raise
            # Suppress errors if we were called quietly.
            else:
                logger.info(
                    'Object was unavailable or unacceptable upstream, but '
                    'pull was called quietly: ' + str(ghid)
                )
        
    async def _attempt_pull_single(self, ghid, remote):
        ''' Attempt to fetch a single object from a single remote. If
        successful, put it into the ingestion pipeline.
        '''
        # This may error, but any errors here will be caught by the parent.
        data = await remote.get(ghid)
        
        # Call as remotable=False to avoid infinite loops.
        ingested = await self._percore.ingest(data, remotable=False)
        
        # Note that ingest can either return None, if we already have
        # the object, or the object itself, if it's new.
        if ingested:
            return ingested
        else:
            return True
            
    async def bootstrap(self, account):
        ''' Bootstrapping publishes our identity upstream.
        
        ONLY do this if we already have a connection available there, or
        we will hang.
        '''
        for remote in self._upstream_remotes:
            if remote.has_connection:
                await remote.publish(account._identity.second_party.packed)
                await remote.subscribe(account._identity.ghid)
            
        self._bootstrapped = True
    
    @fixture_noop
    @public_api
    async def restore_connection(self, remote, connection):
        ''' Start or re-start a connection.
        '''
        # We need to subscribe to our identity if we're restoring a terminated
        # connection. If we haven't bootstrapped though, that will be handled
        # there.
        if self._bootstrapped:
            await remote.publish(self._golcore._identity.second_party.packed)
            await self._remote_protocol.subscribe(
                connection,
                self._golcore.whoami
            )
        
        # For every every active (salmonator-registered) GAO's ghid...
        tasks = set()
        for registrant in self._registered:
            # Record that we need to perform...
            tasks.add(
                # ...as a background future (which handles its own errors)...
                make_background_future(
                    # ...a subscription call with our connection at protocol
                    self._remote_protocol.subscribe(connection, registrant)
                )
            )
        
        # We need to make sure there's at least one task.
        if tasks:
            # And now just wait for all of that to complete.
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
        
        # Now, we need to destructively iterate over our deferreds until the
        # remote list is exhausted. We have to do this in order, sequentially,
        # and serially, because eg. containers require bindings, etc.
        to_push = collections.deque(self._deferred.pop_key(remote))
        try:
            while to_push:
                data = to_push.popleft()
                await self._remote_protocol.publish(connection, data)
        
        # Restore the last to_push, put it back into the deferred list, and
        # then re-raise
        except Exception:
            to_push.appendleft(data)
            self._deferred.extend(remote, to_push)
            raise class RemotePersistenceServer(loopa.TaskCommander):
    ''' Simple persistence server.
    Expected defaults:
    host:       'localhost'
    port:       7770
    logfile:    None
    verbosity:  'warning'
    debug:      False
    traceur:    False
    '''
    
    def __init__(self, cache_dir, host, port, *args, **kwargs):
        ''' Do all of that other smart setup while we're at it.
        '''
        super().__init__(*args, **kwargs)
        
        self.executor = concurrent.futures.ThreadPoolExecutor()
        
        # Persistence stuff
        self.percore = PersistenceCore(self._loop)
        self.doorman = Doorman(self.executor, self._loop)
        self.enforcer = Enforcer()
        self.bookie = Bookie()
        self.lawyer = LawyerCore()
        self.librarian = DiskLibrarian(cache_dir, self.executor, self._loop)
        self.postman = PostOffice()
        self.undertaker = UndertakerCore()
        # I mean, this won't be used unless we set up peering, but it saves us
        # needing to do a modal switch for remote persistence servers
        self.salmonator = Salmonator.__fixture__()
        self.remote_protocol = RemotePersistenceProtocol()
        
        self.percore.assemble(
            doorman = self.doorman,
            enforcer = self.enforcer,
            lawyer = self.lawyer,
            bookie = self.bookie,
            librarian = self.librarian,
            postman = self.postman,
            undertaker = self.undertaker,
            salmonator = self.salmonator
        )
        self.doorman.assemble(librarian=self.librarian)
        self.enforcer.assemble(librarian=self.librarian)
        self.bookie.assemble(librarian=self.librarian)
        self.lawyer.assemble(librarian=self.librarian)
        self.librarian.assemble(
            enforcer = self.enforcer,
            lawyer = self.lawyer,
            percore = self.percore
        )
        self.postman.assemble(
            librarian = self.librarian,
            remote_protocol = self.remote_protocol
        )
        self.undertaker.assemble(
            librarian = self.librarian,
            postman = self.postman
        )
        self.remote_protocol.assemble(
            percore = self.percore,
            librarian = self.librarian,
            postman = self.postman
        )
        
        self.server = BasicServer(connection_cls=WSConnection)
        self.register_task(
            self.server,
            msg_handler = self.remote_protocol,
            host = host,
            port = port,
            tls = False
        )
        self.register_task(self.postman)
        self.register_task(self.undertaker)
        
    async def setup(self):
        ''' Once booted, restore the librarian.
        '''
        await self.librarian.restore() class BasicServer(loopa.ManagedTask):
    ''' Generic server. This isn't really a looped task, just a managed
    one.
    '''
    
    def __init__(self, connection_cls, *args, **kwargs):
        self.connection_cls = connection_cls
        super().__init__(*args, **kwargs)
        
    async def task_run(self, *args, **kwargs):
        await self.connection_cls.serve_forever(*args, **kwargs) class WSConnection(_ConnectionBase):
    ''' Bookkeeping object for a single websocket connection (client or
    server).
    
    This should definitely use slots, to save on server memory usage.
    '''
    
    def __init__(self, websocket, path=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        self.websocket = websocket
        self.path = path
    
    @classmethod
    def desc_str(cls, host, port, tls):
        ''' Override this to explain where the connection is supposed to
        go, for use in debug scenarios.
        '''
        loc = _WSLoc(host, int(port), bool(tls))
        return cls.__name__ + '(' + str(loc) + ')'
        
    @classmethod
    async def serve_forever(cls, msg_handler, host, port, tls=False):
        ''' Starts a server for this kind of connection. Should handle
        its own return, and be cancellable via task cancellation.
        '''
        async def wrapped_msg_handler(websocket, path):
            ''' We need an intermediary that will feed the conn_handler
            actual _WSConnection objects.
            '''
            self = weakref.proxy(cls(websocket, path))
            # Make sure we don't take a strong reference to the connection!
            await self.listen_forever(msg_handler)
        
        try:
            server = await websockets.server.serve(
                wrapped_msg_handler,
                host,
                port,
                max_size = 10 * (2 ** 20)    # Max incoming msg size 10 MiB
            )
            
            try:
                await server.wait_closed()
            
            except Exception:
                server.close()
                await server.wait_closed()
                raise
                
        except asyncio.CancelledError:
            # Don't log being cancelled; it's expected close behavior.
            raise
            
        except Exception as exc:
            logger.error(
                'INTERNAL SERVER ERROR. Closing server w/ traceback:\n' +
                ''.join(traceback.format_exc())
            )
            logger.debug('Error args:' + str(exc.args))
        
    @classmethod
    async def new(cls, host, port, tls):
        ''' Creates and returns a new connection. Intended to be called
        by clients; servers may call __init__ directly.
        '''
        loc = _WSLoc(host, int(port), bool(tls))
        
        # If this raises, we don't need to worry about closing, because the
        # websocket won't exist.
        if tls:
            websocket = await websockets.client.connect(
                str(loc),
                ssl = SSL_VERIFICATION_CONTEXT,
                max_size = 10 * (2 ** 20)    # Max incoming msg size 10 MiB
            )
        
        else:
            websocket = await websockets.client.connect(
                str(loc),
                max_size = 10 * (2 ** 20)    # Max incoming msg size 10 MiB
            )
        
        return cls(websocket=websocket)
        
    async def close(self):
        ''' Wraps websocket.close and calls self.terminate().
        '''
        try:
            # The close() call is idempotent, but it will raise (for example) a
            # ConnectionResetError if something went screwy. So check for
            # availability first.
            if self.websocket.open:
                await self.websocket.close()
            
        finally:
            # And force us to be GC'd
            self.terminate()
        
    async def send(self, msg):
        ''' Send from the same event loop as the websocket.
        '''
        try:
            return (await self.websocket.send(msg))
        
        # If the connection is closed, call our own close (and therefore
        # self-destruct)
        except websockets.exceptions.ConnectionClosed as exc:
            try:
                # await self.close()
                raise ConnectionClosed(
                    'code: ' + str(exc.code) + '; reason: ' + exc.reason
                ) from exc
                
            finally:
                self.terminate()
        
    async def recv(self):
        ''' Receive from the same event loop as the websocket.
        '''
        try:
            return (await self.websocket.recv())
        
        # If the connection is closed, call our own close (and therefore
        # self-destruct)
        except websockets.exceptions.ConnectionClosed as exc:
            try:
                # await self.close()
                raise ConnectionClosed(
                    'code: ' + str(exc.code) + '; reason: ' + exc.reason
                ) from exc
                
            finally:
                self.terminate() class ConnectionManager(loopa.TaskLooper, metaclass=loopa.utils.Triplicate):
    ''' Use this client-side to automatically establish (and, whenever
    necessary, re-establish) a connection with a server. This adds
    exponential backoff for every consecutive failed connection attempt.
    
    ConnectionManagers will handle listening to the connection, and will
    dispatch any incoming requests to protocol_def. They do not buffer
    the message processing; ie, they will only handle one incoming
    message at a time. They also handle closing connections.
    
    Finally, ConnectionManagers may be used to invoke any requests that
    are defined at the msg_handler, and will automatically pass them the
    current connection.
    '''
    # Minimum delay before retrying a connection, in seconds
    MIN_RETRY_DELAY = .01
    # Maximum delay before retrying a connection, in seconds (1 hour)
    MAX_RETRY_DELAY = 3600
    
    def __init__(self, connection_cls, msg_handler, conn_init=None,
                 conn_close=None, *args, autoretry=True, **kwargs):
        ''' We need to assign our connection class.
        
        conn_init, if defined, will be awaited (as a background task)
        every time the connection itself is created.
        
        conn_close, if defined, will be awaited (but not as a background
        task) every time the connection itself has terminated -- AFTER
        the closure.
        '''
        super().__init__(*args, **kwargs)
        
        self.connection_cls = connection_cls
        self.protocol_def = msg_handler
        self.conn_init = conn_init
        self.conn_close = conn_close
        self._connection = None
        self._conn_available = None
        self._conn_args = None
        self._conn_kwargs = None
        self._consecutive_attempts = 0
        # This determines whether or not we even try to reestablish the
        # connection.
        self.autoretry = autoretry
        
        # Very quick and easy way of injecting all of the handler methods into
        # self. Short of having one queue per method, we need to wrap it
        # anyways to buffer the actual method call.
        for name in msg_handler._RESPONDERS:
            async def wrap_request(*args, _method=name, **kwargs):
                ''' Pass all requests to our perform_request method.
                '''
                return (await self.perform_request(_method, args, kwargs))
            
            setattr(self, name, wrap_request)
            
    @property
    def _conn_desc(self):
        ''' Make (and cache) a description of the connection.
        '''
        conn_desc = getattr(self, '__conn_desc', None)
        
        if conn_desc is None:
            conn_desc = self.connection_cls.desc_str(*self._conn_args,
                                                     **self._conn_kwargs)
            setattr(self, '__conn_desc', conn_desc)
            
        return conn_desc
    
    async def loop_init(self, *args, **kwargs):
        ''' *args and **kwargs will be passed to the connection class.
        '''
        self._send_q = asyncio.Queue()
        self._conn_available = asyncio.Event()
        self._conn_args = args
        self._conn_kwargs = kwargs
        self._consecutive_attempts = 0
        self._connection = None
        
    async def loop_stop(self):
        ''' Reset whether or not we have a connection available.
        '''
        self._send_q = None
        self._conn_available = None
        self._conn_args = None
        self._conn_kwargs = None
        self._connection = None
        
    async def loop_run(self):
        ''' Creates a connection and listens forever.
        '''
        # Attempt to connect.
        try:
            logger.debug('Establishing connection to ' + self._conn_desc)
            # Technically this violates the idea that connections should be
            # wholly self-sustained, but we want to be able to explicitly close
            # them and pass some strong references to them later. Maybe. I
            # think, anyways. Okay, not totally sure.
            connection = await self.connection_cls.new(
                *self._conn_args,
                **self._conn_kwargs
            )
            
        # Need to catch this specifically, lest we accidentally do this forever
        except asyncio.CancelledError:
            logger.debug('Connection manager cancelled to ' + self._conn_desc)
            raise
            
        # The connection failed. Wait before reattempting it.
        except Exception as exc:
            logger.error('Failed to establish connection at ' +
                         self._conn_desc)
            logger.info('Failed connection traceback:\n' +
                        ''.join(traceback.format_exc()))
            # Do this first, because otherwise randrange errors (and also
            # otherwise it isn't technically binary exponential backoff)
            self._consecutive_attempts += 1
            backoff = random.randrange((2 ** self._consecutive_attempts) - 1)
            backoff = max(self.MIN_RETRY_DELAY, backoff)
            backoff = min(self.MAX_RETRY_DELAY, backoff)
            
            if self.autoretry:
                await asyncio.sleep(backoff)
            else:
                raise exc
            
        # We successfully connected. Awesome.
        else:
            logger.info('Connected to ' + self._conn_desc)
            # Reset the attempts counter and set that a connection is available
            self._consecutive_attempts = 0
            # See note above re: strong/weak references
            self._connection = connection
            self._conn_available.set()
            
            try:
                # If we need to do something to establish the connection, do
                # that here. We don't need to worry about race conditions with
                # task starting, because we don't cede execution flow control
                # to the loop until we hit the next await.
                if self.conn_init is not None:
                    loopa.utils.make_background_future(
                        self.conn_init(self, connection)
                    )
                
                logger.debug('Listening for messages at ' + self._conn_desc)
                
                # We don't expect clients to have a high enough message volume
                # to justify a buffer, so directly invoke the message handler
                await connection.listen_forever(receiver=self.protocol_def)
                
            # We want to log, but swallow-and-attempt-reconnect, errors in
            # connections.
            except ConnectionError:
                logger.warning('Connection errored: ' + self._conn_desc +
                               ' w/ traceback:\n' +
                               ''.join(traceback.format_exc()))
            
            # No matter what happens, when this dies we need to clean up the
            # connection and tell downstream that we cannot send anymore.
            finally:
                self._conn_available.clear()
                await connection.close()
                logger.info('Connection closed: ' + self._conn_desc)
                
                if self.conn_close is not None:
                    await self.conn_close(self, connection)
                    
    def __str__(self):
        ''' Make a better, compact representation of self.
        '''
        return type(self).__name__ + '(' + self._conn_desc + ')'
            
    async def perform_request(self, request_name, args, kwargs):
        ''' Make the given request using the protocol_def, but wait
        until a connection exists.
        '''
        # Wait for the connection to be available.
        method = getattr(self.protocol_def, request_name)
        await self.await_connection()
        return (await method(self._connection, *args, **kwargs))
    
    @property
    def has_connection(self):
        # Hmmm
        return self._conn_available.is_set()
    
    @loopa.utils.triplicated
    async def await_connection(self):
        ''' Wait for a connection to be available.
        '''
        # There is a potential race condition here with loop_init. The quick
        # and dirty solution to it is to yield control to the event loop
        # repeatedly until it appears. Also, go ahead and add in a bit of a
        # delay for good measure.
        while self._conn_available is None:
            await asyncio.sleep(.01)
            
        await self._conn_available.wait() class _ConnectionBase(metaclass=API):
    ''' Defines common interface for all connections.
    
    TODO: modify this such that the return is ALWAYS weakly referenced,
    but still hashable, etc. Within that proxy, wrap all ReferenceErrors
    with ConnectionClosed errors.
    '''
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Log the creation of the connection.
        '''
        super().__init__(*args, **kwargs)
        # Create a reference to ourselves so that we can manage our own
        # lifetime through self.terminate()
        self._ref = self
        # If termination gets any more complicated, add this in so we can
        # register self.terminate in an atexit call.
        # ALL_CONNECTIONS.add(self)
        
        # Memoize our repr as the hex repr of our id(self)
        self._repr = '<' + type(self).__name__ + ' ' + hex(id(self)) + '>'
        # Memoize a urlsafe base64 string of our id(self) for str
        self._str = str(
            # Change the bytes int to base64
            base64.urlsafe_b64encode(
                # Convert our id(self) to a bytes integer equal to the maximum
                # length of ID on a 64-bit system
                id(self).to_bytes(byteorder='big', length=8)
            ),
            # And then re-cast as a native str
            'utf-8'
        )
        # And log our creation.
        logger.info('CONN ' + str(self) + ' CREATED.')
        
    @__init__.fixture
    def __init__(self, msg_iterator=None, *args, **kwargs):
        ''' Add in an isalive flag and an iterator to simulate message
        reciept.
        '''
        super(_ConnectionBase.__fixture__, self).__init__(*args, **kwargs)
        self._isalive = True
        self._msg_iterator = msg_iterator
        
    def terminate(self):
        ''' Remove our circular reference, which (assuming all other
        refs were weak) will result in our garbage collection.
        Idempotent.
        '''
        # Note: if this gets any more complicated, then we should register an
        # atext cleanup to call terminate. For now, just let GC remove it.
        try:
            del self._ref
            
        except AttributeError:
            logger.debug(
                'Attempted to terminate connection twice. Check for other ' +
                'strong references to the connection.'
            )
            
    def __bool__(self):
        ''' Return True if the connection is still active and False
        otherwise.
        '''
        return hasattr(self, '_ref')
        
    def __repr__(self):
        ''' Make a slightly nicer version of repr, since connections are
        ephemeral and we cannot reproduce them anyways.
        '''
        # Example: <_AutoresponderSession 0x52b2978>
        return self._repr
        
    def __str__(self):
        ''' Make an even more concise version of str, just giving the
        id, as a constant-length base64 string.
        '''
        return self._str
            
    @classmethod
    def desc_str(cls, *args, **kwargs):
        ''' Override this to explain where the connection is supposed to
        go, for use in debug scenarios.
        
        *args and **kwargs should match those used in cls.new()
        '''
        return cls.__name__ + '(*' + str(args) + ', **' + str(kwargs) + ')'
        
    async def listener(self, receiver):
        ''' Once a connection has been created, call this to listen for
        a message until the connection is terminated. Put it in a loop
        to listen forever.
        '''
        try:
            msg = await self.recv()
        
        except ConnectionClosed:
            logger.info(
                'CONN ' + str(self) + ' closed at listener.'
            )
            raise
            
        else:
            logger.debug('CONN ' + str(self) + ' message received.')
            
            try:
                # When we pass to the receiver, make sure we give them a strong
                # reference, so hashing and stuff continues to work.
                await receiver(self, msg)
                
            except asyncio.CancelledError:
                logger.debug('CONN ' + str(self) + ' receive cancelled.')
                raise
            
            except Exception:
                logger.error(
                    'CONN ' + str(self) + ' Listener receiver ' +
                    'raised w/ traceback:\n' + ''.join(traceback.format_exc())
                )
                raise
                
    async def listen_forever(self, receiver):
        ''' Listens until the connection terminates.
        '''
        # Wait until terminate is called.
        try:
            while self:
                # Juuust in case things go south, add this to help
                # cancellation.
                await asyncio.sleep(0)
                await self.listener(receiver)
        
        # Catch this so we don't log a huge traceback on it.
        except ConnectionClosed as exc:
            logger.debug('CONN ' + str(self) + ' close message: ' + str(exc))
        
    @classmethod
    @fixture_api
    async def serve_forever(cls, msg_handler, *args, **kwargs):
        ''' Starts a server for this kind of connection. Should handle
        its own return, and be cancellable via task cancellation.
        '''
        
    @classmethod
    @fixture_api
    async def new(cls, *args, **kwargs):
        ''' Creates and returns a new connection. Intended to be called
        by clients; servers may call __init__ directly.
        '''
        return cls(*args, **kwargs)
    
    @fixture_api
    async def close(self):
        ''' Closes the existing connection, performing any necessary
        cleanup.
        
        Close MUST be idempotent.
        '''
        self._isalive = False
    
    @fixture_api
    async def send(self, msg):
        ''' Does whatever is needed to send a message.
        '''
        # Noop fixture.
    
    @fixture_api
    async def recv(self):
        ''' Waits for first available message and returns it.
        '''
        # Floating point [0.0, 1.0]
        scalar = random.random()
        # Wait somewhere between 0 and .1 seconds
        delay = scalar * .1
        
        if self._msg_iterator is None:
            # Figure out how long to make the message -- somewhere between
            # 32 bytes and a kilobibite, inclusive
            length = random.randrange(32, 1024)
            # Make the message itself
            msg = bytes([random.randint(0, 255) for i in range(length)])
        else:
            msg = next(self._msg_iterator)
            
        if self._isalive:
            # Wait the delay and then return
            await asyncio.sleep(delay)
            return msg
        
        else:
            raise ConnectionClosed() class Aengel:
    ''' Watches for completion of the main thread and then automatically
    closes any other threaded objects (that have been registered with 
    the Aengel) by calling their close methods.
    '''
    def __init__(self, threadname='aengel', guardlings=None):
        ''' Creates an aengel.
        
        Uses threadname as the thread name.
        
        guardlings is an iterable of threaded objects to watch. Each 
        must have a stop_threadsafe() method, which will be invoked upon 
        completion of the main thread, from the aengel's own thread. The
        aengel WILL NOT prevent garbage collection of the guardling 
        objects; they are internally referenced weakly.
        
        They will be called **in the order that they were added.**
        '''
        # I would really prefer this to be an orderedset, but oh well.
        # That would actually break weakref proxies anyways.
        self._guardlings = collections.deque()
        self._dead = False
        self._stoplock = threading.RLock()
        
        if guardlings is not None:
            for guardling in guardlings:
                self.append_guardling(guardling)
            
        self._thread = threading.Thread(
            target = self._watcher,
            daemon = True,
            name = threadname,
        )
        self._thread.start()
        
    def append_guardling(self, guardling):
        if not isinstance(guardling, weakref.ProxyTypes):
            guardling = weakref.proxy(guardling)
            
        self._guardlings.append(guardling)
        
    def prepend_guardling(self, guardling):
        if not isinstance(guardling, weakref.ProxyTypes):
            guardling = weakref.proxy(guardling)
            
        self._guardlings.appendleft(guardling)
        
    def remove_guardling(self, guardling):
        ''' Attempts to remove the first occurrence of the guardling.
        Raises ValueError if guardling is unknown.
        '''
        try:
            self._guardlings.remove(guardling)
        except ValueError:
            logger.error('Missing guardling ' + repr(guardling))
            logger.error('State: ' + repr(self._guardlings))
            raise
    
    def _watcher(self):
        ''' Automatically watches for termination of the main thread and
        then closes the autoresponder and server gracefully.
        '''
        main = threading.main_thread()
        main.join()
        self.stop()
        
    def stop(self, *args, **kwargs):
        ''' Call stop_threadsafe on all guardlings.
        '''
        with self._stoplock:
            if not self._dead:
                for guardling in self._guardlings:
                    try:
                        guardling.stop_threadsafe_nowait()
                    except Exception:
                        # This is very precarious. Swallow all exceptions.
                        logger.error(
                            'Swallowed exception while closing ' + 
                            repr(guardling) + '.\n' + 
                            ''.join(traceback.format_exc())
                        )
                self._dead = True class PersistenceCore(metaclass=API):
    ''' Provides the core functions for storing Golix objects. Required
    for the hypergolix service to start.
    
    Can coordinate with both "upstream" and "downstream" persisters.
    Other persisters should pass through the "ingestive tract". Local
    objects can be published directly through calling the ingest_<type>
    methods.
    
    TODO: add librarian validation, so that attempting to update an
    object we already have an identical copy to silently exits.
    '''
    _doorman = weak_property('__doorman')
    _enforcer = weak_property('__enforcer')
    _lawyer = weak_property('__lawyer')
    _bookie = weak_property('__bookie')
    _postman = weak_property('__postman')
    _undertaker = weak_property('__undertaker')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    
    # This is a messy way of getting the suffix for validation and stuff but
    # it's getting the job done.
    _ATTR_LOOKUP = {
        _GidcLite: 'gidc',
        _GeocLite: 'geoc',
        _GobsLite: 'gobs',
        _GobdLite: 'gobd',
        _GdxxLite: 'gdxx',
        _GarqLite: 'garq'
    }
    
    @public_api
    def __init__(self, loop=None, *args, **kwargs):
        ''' Create a KeyedAsyncioLock for ingestion.
        '''
        self._ingestion_mutex = KeyedAsyncioLock(loop)
        
    @__init__.fixture
    def __init__(self, librarian=None, *args, **kwargs):
        ''' Right now, well, this is basically a noop. Anticipating
        substantial changes!
        '''
        super(PersistenceCore.__fixture__, self).__init__(
            *args,
            **kwargs
        )
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, doorman, enforcer, lawyer, bookie, librarian, postman,
                 undertaker, salmonator):
        self._doorman = doorman
        self._enforcer = enforcer
        self._lawyer = lawyer
        self._bookie = bookie
        self._postman = postman
        self._undertaker = undertaker
        self._librarian = librarian
        self._salmonator = salmonator
    
    @public_api
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Standard ingestion flow for stuff. To be called from ingest
        above, or directly (for objects created "in-house").
        '''
        # Check for a redundant object, which will immediately short-circuit.
        if isinstance(obj, _GobdLite):
            check_ghid = obj.frame_ghid
            log_frame = True
            counter = obj.counter
            target = obj.target
        else:
            check_ghid = obj.ghid
            log_frame = False
            counter = 0
            target = None
        
        async with self._ingestion_mutex(check_ghid):
            if (await self._librarian.contains(check_ghid)):
                logger.debug(
                    str(check_ghid) + ' not ingested: already exists.'
                )
                return False
            
            else:
                logger.info(
                    'Ingesting ' + str(obj) +
                    (
                        ' frame ' + str(counter) + ': ' + str(check_ghid) +
                        ', target: ' + str(target)
                    ) * log_frame + '...'
                )
                # Calculate "gidc", etc
                suffix = self._ATTR_LOOKUP[type(obj)]
                validation_method = 'validate_' + suffix
                
                # Validate the object... (will raise for invalid)
                # ########################
                # Enforce target selection
                await getattr(self._enforcer, validation_method)(obj)
                # Now make sure authorship requirements are satisfied
                await getattr(self._lawyer, validation_method)(obj)
                # Finally make sure persistence rules are followed
                await getattr(self._bookie, validation_method)(obj)
                
                # Ingest the object
                # ########################
                # Alert the undertaker for any necessary GC of targets, etc. Do
                # that before storing at the librarian, so that the undertaker
                # has access to the old state.
                await getattr(self._undertaker, 'alert_' + suffix)(obj,
                                                                   skip_conn)
                # Finally, add it to the librarian.
                await self._librarian.store(obj, packed)
                
                if remotable:
                    await self._salmonator.push(obj.ghid)
            
                return True
    
    @direct_ingest.fixture
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Fixture direct ingest to just store at librarian directly
        with no verification (if librarian exists), otherwise simply
        return None.
        '''
        try:
            librarian = self._librarian
        except AttributeError:
            return None
        else:
            await librarian.store(obj, packed)
    
    @public_api
    async def attempt_load(self, packed, quiet=True):
        ''' Attempt to load a packed golix object. If quiet=False, it
        will raise if there is no loader.
        
        TODO: move into doorman.
        '''
        # This is kinda silly, but instead of spewing off a million different
        # threads that will immediately die, let's pre-select the loader
        magic = bytes(packed[:4])
        loaders = {
            b'GIDC': self._doorman.load_gidc,
            b'GEOC': self._doorman.load_geoc,
            b'GOBS': self._doorman.load_gobs,
            b'GOBD': self._doorman.load_gobd,
            b'GDXX': self._doorman.load_gdxx,
            b'GARQ': self._doorman.load_garq
        }
        
        try:
            loader = loaders[magic]
            
        # If no successful loader was found, return None, and allow the parent
        # to raise.
        except KeyError as exc:
            if quiet:
                return None
            else:
                raise MalformedGolixPrimitive('No loader found for magic: ' +
                                              str(magic)) from exc
        
        obj = await loader(packed)
            
        return obj
        
    @attempt_load.fixture
    async def attempt_load(self, packed):
        ''' Create an ad-hoc doorman fixture.
        '''
        # Note that, because of the weak ref, we need to actually hold this
        # here to prevent the doorman from being gc'd. So, just create a
        # disposable one.
        doorman = Doorman.__fixture__()
        self._doorman = doorman
        result = \
            await super(PersistenceCore.__fixture__, self).attempt_load(packed)
        del doorman
        return result
    
    @fixture_noop
    @public_api
    async def ingest(self, packed, remotable=True, skip_conn=None):
        ''' Called on an untrusted and unknown object. May be bypassed
        by locally-created, trusted objects (by calling the individual
        ingest methods directly). Parses, validates, and stores the
        object, and returns True; or, raises an error.
        '''
        # This may return None, but that will be caught by the KeyError below.
        obj = await self.attempt_load(packed)
        if obj is None:
            raise MalformedGolixPrimitive(
                'Packed bytes do not appear to be a Golix primitive.'
            )
        
        ingested = await self.direct_ingest(obj, packed, remotable)
        
        # Note that individual ingest methods are only called directly for
        # locally-built objects, which do not need a mail run.
        # If the object is identical to what we already have, the ingester
        # will return None.
        if ingested:
            await self._postman.schedule(obj, skip_conn=skip_conn)
            
        else:
            logger.debug(str(obj.ghid) + ' postal scheduling aborted on ' +
                         'unchanged object.')
        
        # Note: this is not the place for salmonator pushing! Locally
        # created/updated objects call the individual ingest methods
        # directly, so they have to be the ones that actually deal with it.
        
        return ingested class LibrarianCore(metaclass=API):
    ''' Base class for caching systems common to non-volatile librarians
    such as DiskLibrarian, S3Librarian, etc.
    
    TODO: make ghid vs frame ghid usage more consistent across things.
    
    NOTE: ideally, everything stateful should be in the librarian. In
    other words, you shouldn't use a bookie to track binding status and
    stuff; that should all be handled within librarian. That would make
    the librarian the single source of truth. Unfortunately, there's not
    really a good way of doing that right now ops-wise; you'd need some
    kind of relational filesystem.
    
    TODO: instead, should create a dedicated "cache" system. It can be
    responsible for restoration of stuff, atomic state updates, etc.
    '''
    _enforcer = weak_property('__enforcer')
    _lawyer = weak_property('__lawyer')
    _percore = weak_property('__percore')
    
    @public_api
    def __init__(self, *args, memory_cache=10000, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Lookup for ghid -> hypergolix description
        # This may be GC'd by the python process.
        self._catalog = FiniteDict(maxlen=memory_cache)
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Construct an in-memory-only version of librarian.
        '''
        super(LibrarianCore.__fixture__, self).__init__(*args, **kwargs)
        
        # Lookup <ghid>: <raw data>
        self._shelf = {}
        
        # Lookup for dynamic ghid -> frame ghid
        self._dyn_resolver = {}
        
        # Lookup <bound ghid>: set(<binding obj>)
        self._bound_by_ghid = SetMap()
        
        # Lookup <debound ghid>: set(<debinding ghid>)
        self._debound_by_ghid = SetMap()
        
        # Lookup <recipient>: set(<request ghid>)
        self._requests_for_recipient = SetMap()
        
    @fixture_api
    def RESET(self):
        ''' Reset all of the librarian.
        '''
        self._catalog.clear()
        self._shelf.clear()
        self._dyn_resolver.clear()
        self._bound_by_ghid.clear_all()
        self._debound_by_ghid.clear_all()
        self._requests_for_recipient.clear_all()
        
    def assemble(self, enforcer, lawyer, percore):
        ''' Assign stuff hereto avoid circuitous problems.
        '''
        # We need to be able to check the legality of debindings whose targets
        # are uploaded after the debinding itself is.
        self._enforcer = enforcer
        self._lawyer = lawyer
        # And we need to be able to load things back up from the cache
        self._percore = percore
    
    @public_api
    async def is_bound(self, obj):
        ''' Check to see if the object has been bound.
        '''
        bindings = await self.bind_status(obj.ghid)
        return bool(bindings)
    
    @public_api
    async def is_debound(self, obj):
        # We need to validate any unvalidated bindings. For now, just always
        # validate debindings when an object is being checked.
        debindings = await self.debind_status(obj.ghid)
        invalidated = 0
        for debinding_ghid in debindings:
            # Get the existing debinding object
            debinding = await self.summarize(debinding_ghid)
            
            # Validate existing binding against newly-known target
            try:
                await self._enforcer.validate_gdxx(debinding, target_obj=obj)
                await self._lawyer.validate_gdxx(debinding, target_obj=obj)
                
            # Validation failed. Remove illegal debinding.
            except (InvalidTarget, InconsistentAuthor):
                logger.error(''.join((
                    'Removed invalid existing debinding.\n',
                    '    Debinding author:     ', str(debinding.author), '\n',
                    '    Valid object author:  ', str(obj.author), '\n',
                    '    Debinding target:     ', str(debinding.target), '\n',
                    '    Target type:          ', str(type(obj))
                )))
                # There's no need to do anything with the undertaker, because
                # debindings don't get subscribed to, and we (the librarian)
                # are the sole source of truth.
                await self.abandon(debinding)
                
                # Record that this was invalidated
                invalidated += 1
                
            # If we were differentiating between validated debindings and
            # provisional (unvalidated) debindings, this would be the place to
            # commit the provisional to the validated.
        
        # Instead of making another call to debind_status, we can calculate it
        # directly. NOTE! There is a race condition between the initial
        # validity check on upload, and the commit call.
        
        return bool(len(debindings) - invalidated)
        
    @is_debound.fixture
    async def is_debound(self, obj):
        ''' Create ad-hoc enforcers and lawyers when fixturing self.
        '''
        enforcer = Enforcer.__fixture__(librarian=self)
        lawyer = LawyerCore.__fixture__(librarian=self)
        self._enforcer = enforcer
        self._lawyer = lawyer
        result = await super(LibrarianCore.__fixture__, self).is_debound(obj)
        del enforcer
        del lawyer
        return result
    
    @public_api
    async def store(self, obj, data):
        ''' Starts tracking an object.
        obj is a hypergolix representation object.
        raw is bytes-like.
        '''
        if isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            # If we have an existing frame, get ITS frame ghid, so we can clear
            # it if the storage succeeds.
            try:
                old_ghid = await self.resolve_frame(obj.ghid)
            except KeyError:
                old_ghid = None
            
        else:
            reference_ghid = obj.ghid
            old_ghid = None
            
        await self.add_to_cache(obj, data)
        self._catalog[reference_ghid] = obj
        
        # If successful (which is any time we get to here), we also need to get
        # rid of any old dynamic frames and pop them from the catalog.
        if old_ghid is not None:
            await self.remove_from_cache(old_ghid)
            # We need the None regardless of bugs, in case the old frame is
            # "stale" enough to have been released from memory
            self._catalog.pop(old_ghid, None)
    
    @public_api
    async def retrieve(self, ghid):
        ''' Returns the raw data associated with the ghid, checking only
        locally.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
            
        return (await self.get_from_cache(ghid))
    
    @public_api
    async def summarize(self, ghid):
        ''' Returns a lightweight Hypergolix description of the object.
        Checks only locally.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        try:
            obj = self._catalog[ghid]
        
        except KeyError:
            logger.debug('Attempting lazy-load for ' + str(ghid))
            # This will raise DoesNotExist if missing.
            data = await self.get_from_cache(ghid)
            # This does NOT ingest the data into the persistence system!
            obj = await self._percore.attempt_load(data, quiet=False)
            self._catalog[ghid] = obj
            
        return obj
        
    @public_api
    async def abandon(self, obj):
        ''' Forces erasure of an object without notifying anyone else.
        Idempotent. Should never raise KeyError.
        '''
        if isinstance(obj, _GobdLite):
            ghid = obj.frame_ghid
        else:
            ghid = obj.ghid
        
        await self.remove_from_cache(ghid)
        
        # Delete it from the catalog (if it exists there)
        self._catalog.pop(ghid, None)
    
    # Subclasses MAY define this, but are not required to do so.
    @fixture_api
    async def restore(self):
        ''' For LibrarianCore, do nothing. This is just here as an
        endpoint for restore calls. If subclasses want/need to support
        restoration, they should override this.
        '''
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def contains(self, ghid):
        ''' Checks the ghidcache for the ghid.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        # Catalog may only be accurate locally. Shelf is accurate globally.
        return ghid in self._shelf
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def resolve_frame(self, ghid):
        ''' Get the current frame ghid from the dynamic ghid.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('Ghid must be a Ghid.')
            
        if ghid in self._dyn_resolver:
            return self._dyn_resolver[ghid]
        else:
            raise KeyError(str(ghid) + ' not known as dynamic ghid.')
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def recipient_status(self, ghid):
        ''' Return a frozenset of ghids assigned to the passed ghid as
        a recipient.
        '''
        return self._requests_for_recipient.get_any(ghid)
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def bind_status(self, ghid):
        ''' Return a frozenset of ghids binding the passed ghid.
        '''
        return self._bound_by_ghid.get_any(ghid)
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def debind_status(self, ghid):
        ''' Return either a ghid, or None.
        '''
        # Note that any particular object can have exactly zero or one VALID
        # debinds, but that a malicious actor could find a race condition and
        # debind something FOR SOMEONE ELSE before the bookie knows about the
        # original object authorship.
        return self._debound_by_ghid.get_any(ghid)
            
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def add_to_cache(self, obj, data):
        ''' Adds the passed raw data to the cache.
        '''
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            try:
                existing = await self.summarize(obj.ghid)
            except KeyError:
                pass
            else:
                self._bound_by_ghid.remove(existing.target, obj.ghid)
                
            # Now we have a clean slate and need to update things accordingly.
            self._bound_by_ghid.add(obj.target, obj.ghid)
            self._dyn_resolver[obj.ghid] = obj.frame_ghid
                
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            # NOTE: this is only necessary for a persistence server
            self._requests_for_recipient.add(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
            
        self._shelf[reference_ghid] = data
        
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def remove_from_cache(self, ghid):
        ''' Removes the data associated with the passed ghid from the
        cache.
        '''
        obj = await self.summarize(ghid)
        
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
            if self._dyn_resolver.get(obj.ghid, None) == obj.frame_ghid:
                self._dyn_resolver.pop(obj.ghid, None)
            
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            self._requests_for_recipient.discard(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
        
        try:
            self._shelf.pop(reference_ghid)
        except KeyError as exc:
            # Protect the full GHID from accidental exposure
            raise DoesNotExist(str(ghid)) from None
        
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def get_from_cache(self, ghid):
        ''' Returns the raw data associated with the ghid.
        '''
        try:
            return self._shelf[ghid]
        except KeyError as exc:
            # Protect the full GHID from accidental exposure
            raise DoesNotExist(str(ghid)) from None class PostalCore(loopa.TaskLooper, metaclass=API):
    ''' Tracks, delivers notifications about objects using **only weak
    references** to them. Threadsafe.
    
    ♫ Please Mister Postman... ♫
    
    Question: should the distributed state management of GARQ recipients
    be managed here, or in the bookie (where it currently is)?
    '''
    _librarian = weak_property('__librarian')
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # The scheduling queue is created at loop init.
        self._scheduled = None
        # The delayed lookup. <awaiting ghid>: set(<subscribed ghids>)
        self._deferred = SetMap()
        
        # Resolve primitives into their schedulers.
        self._scheduler_lookup = {
            _GidcLite: self._schedule_gidc,
            _GeocLite: self._schedule_geoc,
            _GobsLite: self._schedule_gobs,
            _GobdLite: self._schedule_gobd,
            _GdxxLite: self._schedule_gdxx,
            _GarqLite: self._schedule_garq
        }
        
    @fixture_api
    def RESET(self):
        ''' In general this would be where you'd reset self._scheduled,
        but since self.schedule() is fixtured as a NOOP, there's no
        real reason to do anything here.
        '''
        
    def assemble(self, librarian):
        # Links the librarian.
        self._librarian = librarian
        
    async def await_idle(self):
        ''' Wait until the postman has no more deliveries to perform.
        '''
        while self._scheduled is None:
            await asyncio.sleep(.01)
        
        await self._scheduled.join()
        
    async def loop_init(self):
        ''' Init all of the needed async primitives.
        '''
        self._scheduled = asyncio.Queue()
        
    async def loop_run(self):
        ''' Deliver notifications as soon as they are available.
        TODO: support parallel sending.
        '''
        subscription, notification, skip_conn = await self._scheduled.get()
        
        try:
            logger.info(str(subscription) + ' subscription out for delivery.')
            # We can't spin this out into a thread because some of our
            # delivery mechanisms want this to have an event loop.
            await self._deliver(subscription, notification, skip_conn)
            
        except asyncio.CancelledError:
            logger.debug('PostalCore cancelled.')
            raise
        
        except Exception:
            logger.error(str(subscription) + ' subscription FAILED for ' +
                         'notification ' + str(notification) +
                         ' w/ traceback:\n' +
                         ''.join(traceback.format_exc()))
            
        finally:
            self._scheduled.task_done()
        
    async def loop_stop(self):
        ''' Clear the async primitives.
        '''
        # Ehhhhh, should the queue be emptied before being destroyed?
        self._scheduled = None
        
    @fixture_return(True)
    @public_api
    async def schedule(self, obj, removed=False, skip_conn=None):
        ''' Schedules update delivery for the passed object.
        '''
        # It's possible we're being told to schedule nothing, so catch that
        # here.
        if obj is None:
            return
        
        else:
            try:
                scheduler = self._scheduler_lookup[type(obj)]
            
            except KeyError:
                raise TypeError('Postman scheduling failed for ' +
                                type(obj).__name__) from None
            
            else:
                await scheduler(obj, removed, skip_conn)
                
            return True
        
    async def _schedule_gidc(self, obj, removed, skip_conn):
        # GIDC will never trigger a subscription.
        pass
        
    async def _schedule_geoc(self, obj, removed, skip_conn):
        # GEOC will never trigger a subscription directly, though they might
        # have deferred updates.
        # Note that these have already been put into _SubsUpdate form.
        for deferred in self._deferred.pop_any(obj.ghid):
            await self._scheduled.put(deferred)
        
    async def _schedule_gobs(self, obj, removed, skip_conn):
        # GOBS will never trigger a subscription.
        pass
        
    async def _schedule_gobd(self, obj, removed, skip_conn):
        # GOBD might trigger a subscription! But, we also might to need to
        # defer it. Or, we might be removing it.
        if removed:
            debinding_ghids = await self._librarian.debind_status(obj.ghid)
            
            # Check to see that there are the proper number of debindings
            num_debindings = len(debinding_ghids)
            if num_debindings != 1:
                raise ValueError('Improper debinding number: ' +
                                 str(num_debindings) + ' for ' + str(obj))
                
            # Debinding_ghids is a frozenset; this is the fastest way of
            # getting the single element from it.
            debinding_ghid = next(iter(debinding_ghids))
            await self._scheduled.put(
                _SubsUpdate(obj.ghid, debinding_ghid, skip_conn)
            )
            
        else:
            notifier = _SubsUpdate(obj.ghid, obj.frame_ghid, skip_conn)
            if (await self._librarian.contains(obj.target)):
                logger.debug(str(obj) +
                             ' subscription notification scheduled for: ' +
                             str(obj.target))
                await self._scheduled.put(notifier)
            
            else:
                self._deferred.add(obj.target, notifier)
                logger.debug(str(obj) +
                             ' subscription notification deferred for: ' +
                             str(obj.target))
        
    async def _schedule_gdxx(self, obj, removed, skip_conn):
        # GDXX will never directly trigger a subscription. If they are removing
        # a subscribed object, the actual removal (in the undertaker GC) will
        # trigger a subscription without us.
        pass
        
    async def _schedule_garq(self, obj, removed, skip_conn):
        # GARQ might trigger a subscription! Or we might be removing it.
        if removed:
            debinding_ghids = await self._librarian.debind_status(obj.ghid)
            
            # Check to see that there are the proper number of debindings
            num_debindings = len(debinding_ghids)
            if num_debindings != 1:
                raise ValueError('Improper debinding number: ' +
                                 str(num_debindings) + ' for ' + str(obj))
                
            # Debinding_ghids is a frozenset; this is the fastest way of
            # getting the single element from it.
            debinding_ghid = next(iter(debinding_ghids))
            await self._scheduled.put(
                _SubsUpdate(obj.recipient, debinding_ghid, skip_conn)
            )
        else:
            await self._scheduled.put(
                _SubsUpdate(obj.recipient, obj.ghid, skip_conn)
            )
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        '''
        # We need to freeze the listeners before we operate on them, but we
        # don't need to lock them while we go through all of the callbacks.
        # Instead, just sacrifice any subs being added concurrently to the
        # current delivery run.
        pass class PostOffice(PostalCore, metaclass=API):
    ''' Postman to use for remote persistence servers.
    '''
    _remote_protocol = weak_property('__remote_protocol')
    
    def __init__(self, *args, subs_timeout=30, **kwargs):
        super().__init__(*args, **kwargs)
        # By using WeakSetMap we can automatically handle dropped connections
        # Lookup <subscribed ghid>: set(<subscribed callbacks>)
        self._connections = WeakSetMap()
        self._subscriptions = WeakKeySetMap()
        
        self._subs_timeout = subs_timeout
        
    def assemble(self, librarian, remote_protocol):
        super().assemble(librarian)
        self._remote_protocol = remote_protocol
    
    @fixture_noop
    @public_api
    async def subscribe(self, connection, ghid):
        ''' Tells the postman that the connection would like to be
        updated about ghid.
        '''
        logger.debug('CONN ' + str(connection) + ' subscribed to ' + str(ghid))
        
        # First add the subscription listeners
        self._connections.add(ghid, connection)
        self._subscriptions.add(connection, ghid)
        
        # Now manually reinstate any desired notifications for garq requests
        # that have yet to be handled
        for existing_mail in (await self._librarian.recipient_status(ghid)):
            obj = await self._librarian.summarize(existing_mail)
            await self.schedule(obj)
    
    @fixture_noop
    @public_api
    async def unsubscribe(self, connection, ghid):
        ''' Remove the callback for ghid. Indempotent; will never raise
        a keyerror.
        '''
        try:
            # Discard the subscription (no keyerror)
            self._subscriptions.discard(connection, ghid)
            # Remove the connection (allowing it to raise if missing)
            self._connections.remove(ghid, connection)
            
        except KeyError:
            logger.debug('CONN ' + str(connection) + ' never subscribed to ' +
                         str(ghid))
            return False
            
        else:
            logger.debug('CONN ' + str(connection) + ' unsubscribed to ' +
                         str(ghid))
            return True
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        
        NOTE THAT SKIP_CONN is a weakref.ref.
        '''
        pkg_label = (str(subscription) + ' subscription, notification ' +
                     str(notification))
        # We need to freeze the listeners before we operate on them, but we
        # don't need to lock them while we go through all of the callbacks.
        # Instead, just sacrifice any subs being added concurrently to the
        # current delivery run.
        connections = self._connections.get_any(subscription)
        logger.debug(
            pkg_label + ' has ' + str(len(connections)) + ' listeners.'
        )
        
        # Resolve the weak reference to the connection
        if skip_conn is not None:
            # This could still be None, but that won't affect our comparison
            skip_conn = skip_conn()
            
        for connection in connections:
            if connection is not skip_conn:
                # Make this a background task, or one blocking connection can
                # hold up the entire subscription queue
                make_background_future(
                    self._remote_protocol.subscription_update(
                        connection,
                        subscription,
                        notification,
                        timeout = self._subs_timeout
                    )
                )
            else:
                logger.debug(pkg_label + ' skipped one connection.')
    
    @fixture_return(frozenset())
    @public_api
    async def list_subs(self, connection):
        ''' List all subscriptions for the connection.
        '''
        return self._subscriptions.get_any(connection)
    
    @fixture_noop
    @public_api
    async def clear_subs(self, connection):
        ''' Clear all subscriptions for the connection.
        '''
        subscriptions = self._subscriptions.pop_any(connection)
        for ghid in subscriptions:
            self._connections.discard(ghid, connection) class GolixCore(metaclass=API):
    ''' Wrapper around Golix library that automates much of the state
    management, holds the Agent's identity, etc etc.
    '''
    _librarian = weak_property('__librarian')
    DEFAULT_LEGROOM = 7
    
    # Async stuff
    _executor = readonly_property('__executor')
    _loop = readonly_property('__loop')
    
    @public_api
    def __init__(self, executor, loop, *args, **kwargs):
        ''' Create a new agent. Persister should subclass _PersisterBase
        (eventually this requirement may be changed).
        
        persister isinstance _PersisterBase
        dispatcher isinstance DispatcherBase
        _identity isinstance golix.FirstParty
        '''
        super().__init__(*args, **kwargs)
        self._mutex_request = threading.Lock()
        self._mutex_container = threading.Lock()
        self._mutex_sbinding = threading.Lock()
        self._mutex_dbinding = threading.Lock()
        self._mutex_xbinding = threading.Lock()
        
        # Added during bootstrap
        self.__identity = None
        
        # Async-specific stuff
        setattr(self, '__executor', executor)
        setattr(self, '__loop', loop)
        
    @__init__.fixture
    def __init__(self, test_agent, *args, librarian=None, **kwargs):
        ''' Just, yknow, throw in the towel. Err, agent. Whatever.
        '''
        super(GolixCore.__fixture__, self).__init__(
            executor = None,
            loop = None,
            *args,
            **kwargs
        )
        self._identity = test_agent
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, librarian):
        # Chicken, meet egg.
        self._librarian = librarian
        
    def bootstrap(self, account):
        # This must be done ASAGDFP. Must be absolute first thing to bootstrap.
        self._identity = account._identity
        
    @property
    def _identity(self):
        ''' Wrap our identity in such a way that, if it's not defined,
        we can fall back on the FirstParty class.
        '''
        if self.__identity is None:
            return FirstParty
        else:
            return self.__identity
            
    @_identity.setter
    def _identity(self, identity):
        ''' Set the identity.
        '''
        self.__identity = identity
        
    @property
    @public_api
    def whoami(self):
        ''' Return the Agent's Ghid.
        '''
        return self._identity.ghid
    
    @public_api
    async def unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._unpack_request,
            request
        ))
        
    @unpack_request.fixture
    async def unpack_request(self, request):
        ''' Bypass executor for the fixture.
        '''
        return self._unpack_request(request)
        
    def _unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        with self._mutex_request:
            unpacked = self._identity.unpack_request(request)
        return unpacked
        
    @public_api
    async def open_request(self, unpacked):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        try:
            requestor = SecondParty.from_packed(
                await self._librarian.retrieve(unpacked.author)
            )
            
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(unpacked.author)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_request,
            unpacked,
            requestor
        ))
        
    @open_request.fixture
    async def open_request(self, unpacked):
        ''' Also bypass executor here.
        '''
        requestor = SecondParty.from_packed(
            await self._librarian.retrieve(unpacked.author)
        )
        return self._open_request(unpacked, requestor)
        
    def _open_request(self, unpacked, requestor):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        return self._identity.receive_request(requestor, unpacked)
    
    @public_api
    async def make_request(self, recipient, payload):
        # Just like it says on the label...
        try:
            recipient = SecondParty.from_packed(
                await self._librarian.retrieve(recipient)
            )
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(recipient)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_request,
            recipient,
            payload
        ))
        
    @make_request.fixture
    async def make_request(self, recipient, payload):
        ''' Bypass the goddamn executor. Ffs.
        '''
        recipient = SecondParty.from_packed(
            await self._librarian.retrieve(recipient)
        )
        return self._make_request(recipient, payload)
        
    def _make_request(self, recipient, payload):
        # Just like it says on the label...
        with self._mutex_request:
            return self._identity.make_request(
                recipient = recipient,
                request = payload,
            )
    
    @public_api
    async def open_container(self, container, secret):
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        # Wrapper around golix.FirstParty.receive_container.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_container,
            container,
            secret,
            author
        ))
        
    @open_container.fixture
    async def open_container(self, container, secret):
        ''' Bypass executor for fixture.
        '''
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        return self._open_container(container, secret, author)
        
    def _open_container(self, container, secret, author):
        # Wrapper around golix.FirstParty.receive_container.
        with self._mutex_container:
            return self._identity.receive_container(
                author = author,
                secret = secret,
                container = container
            )
    
    @public_api
    async def make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_container,
            data,
            secret
        ))
        
    @make_container.fixture
    async def make_container(self, data, secret):
        ''' Bypass executor.
        '''
        return self._make_container(data, secret)
        
    def _make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        with self._mutex_container:
            return self._identity.make_container(
                secret = secret,
                plaintext = data
            )

    @public_api
    async def make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_stat,
            target
        ))
        
    @make_binding_stat.fixture
    async def make_binding_stat(self, target):
        ''' Skip the executor for the fixture.
        '''
        return self._make_binding_stat(target)

    def _make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        with self._mutex_sbinding:
            return self._identity.make_bind_static(target)
    
    @public_api
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        # Either ghid AND history must be defined, XOR ghid AND history must be
        # undefined.
        if bool(ghid) ^ bool(history):
            raise ValueError('Mixed def of ghid/history while dyn binding.')
        
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
            
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_dyn,
            ghid,
            counter,
            target,
            history
        ))
        
    @make_binding_dyn.fixture
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Yep, just bypass the executor
        '''
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
        
        return self._make_binding_dyn(ghid, counter, target, history)
        
    def _make_binding_dyn(self, ghid, counter, target, history):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        if history is not None:
            target_vector = [target, *history]
        else:
            target_vector = [target]
        
        with self._mutex_dbinding:
            return self._identity.make_bind_dynamic(
                counter = counter,
                target_vector = target_vector,
                ghid_dynamic = ghid
            )
    
    @public_api
    async def make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_debinding,
            target
        ))
        
    @make_debinding.fixture
    async def make_debinding(self, target):
        ''' Bypass executor for the fixture.
        '''
        return self._make_debinding(target)
        
    def _make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        with self._mutex_xbinding:
            return self._identity.make_debind(target) class RemoteNak(PersistenceError):
    ''' This exception (or a subclass thereof) is raised for all failed
    operations with remote persister servers.
    '''
    pass class StillBoundWarning(HypergolixException, RuntimeWarning):
    ''' Raised when a debinding did not result in the removal of its
    target -- for example, if another binding remains on the target
    object.
    '''
    pass class _GidcLite(_BaseLite):
    ''' Lightweight description of a GIDC.
    '''
    __slots__ = [
        'identity'
    ]
    
    def __init__(self, ghid, identity):
        self.ghid = ghid
        self.identity = identity
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        identity = SecondParty.from_identity(golix_obj)
        return cls(
            ghid = golix_obj.ghid,
            identity = identity,
        ) class _GeocLite(_BaseLite):
    ''' Lightweight description of a GEOC.
    '''
    __slots__ = [
        'author',
    ]
    
    def __init__(self, ghid, author):
        self.ghid = ghid
        self.author = author
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author
            )
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.author,
        ) class _GobsLite(_BaseLite):
    ''' Lightweight description of a GOBS.
    '''
    __slots__ = [
        'author',
        'target',
    ]
    
    def __init__(self, ghid, author, target):
        self.ghid = ghid
        self.author = author
        self.target = target
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self.target == other.target
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.binder,
            target = golix_obj.target,
        ) class _GobdLite(_BaseLite):
    ''' Lightweight description of a GOBD.
    '''
    __slots__ = [
        'author',
        'counter',
        'target_vector',
        'frame_ghid'
    ]
    
    def __init__(self, ghid, author, counter, target_vector, frame_ghid):
        self.ghid = ghid
        self.author = author
        self.counter = counter
        self.target_vector = target_vector
        self.frame_ghid = frame_ghid
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self.target == other.target and
                self.frame_ghid == other.frame_ghid
                # Skip history, because it could potentially vary
                # self.history == other.history
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid_dynamic,
            author = golix_obj.binder,
            counter = golix_obj.counter,
            target_vector = golix_obj.target_vector,
            frame_ghid = golix_obj.ghid
        )
        
    @property
    def target(self):
        ''' Access the current target.
        '''
        return self.target_vector[0] class _GdxxLite(_BaseLite):
    ''' Lightweight description of a GDXX.
    '''
    __slots__ = [
        'author',
        'target',
        '_debinding',
    ]
    
    def __init__(self, ghid, author, target):
        self.ghid = ghid
        self.author = author
        self.target = target
        self._debinding = True
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self._debinding == other._debinding
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.debinder,
            target = golix_obj.target,
        ) class _GarqLite(_BaseLite):
    ''' Lightweight description of a GARQ.
    '''
    __slots__ = [
        'recipient',
    ]
    
    def __init__(self, ghid, recipient):
        self.ghid = ghid
        self.recipient = recipient
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.recipient == other.recipient
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            recipient = golix_obj.recipient,
        )
import unittest
import threading
import pathlib
import logging
import tempfile
import shutil
import random
from loopa.utils import await_coroutine_threadsafe
from loopa import TaskCommander
from loopa import NoopLoop
from hypergolix.remotes import RemotePersistenceProtocol
from hypergolix.remotes import Salmonator
from hypergolix.service import RemotePersistenceServer
from hypergolix.comms import BasicServer
from hypergolix.comms import WSConnection
from hypergolix.comms import ConnectionManager
from hypergolix.comms import _ConnectionBase
from hypergolix.utils import Aengel
from hypergolix.persistence import PersistenceCore
from hypergolix.librarian import LibrarianCore
from hypergolix.postal import PostalCore
from hypergolix.postal import PostOffice
from hypergolix.core import GolixCore
from hypergolix.exceptions import RemoteNak
from hypergolix.exceptions import StillBoundWarning
from golix import Ghid
from golix import ThirdParty
from golix import SecondParty
from golix._getlow import GIDC
from hypergolix.persistence import _GidcLite
from hypergolix.persistence import _GeocLite
from hypergolix.persistence import _GobsLite
from hypergolix.persistence import _GobdLite
from hypergolix.persistence import _GdxxLite
from hypergolix.persistence import _GarqLite
from _fixtures.identities import TEST_AGENT1
from _fixtures.identities import TEST_AGENT2
from _fixtures.identities import TEST_AGENT3
from _fixtures.identities import TEST_READER1
from _fixtures.identities import TEST_READER2
from _fixtures.identities import TEST_READER3
from _fixtures.remote_exchanges import gidc1
from _fixtures.remote_exchanges import gidc2
from _fixtures.remote_exchanges import cont1_1  # Known author
from _fixtures.remote_exchanges import cont1_2  # Known author
from _fixtures.remote_exchanges import cont2_1  # Known author
from _fixtures.remote_exchanges import cont3_1  # Unknown author
from _fixtures.remote_exchanges import bind1_1  # Known author
from _fixtures.remote_exchanges import bind3_1  # Unknown author
from _fixtures.remote_exchanges import dyn1_1a  # Known author frame 1
from _fixtures.remote_exchanges import dyn1_1b  # Known author frame 2
from _fixtures.remote_exchanges import dyn3_1a  # Unknown author frame 1
from _fixtures.remote_exchanges import dyn3_1b  # Unknown author frame 2
from _fixtures.remote_exchanges import dynF_a   # Inconsistent author frame 1
from _fixtures.remote_exchanges import dynF_b   # Inconsistent author frame 2
from _fixtures.remote_exchanges import dynF_c   # Inconsistent, unk author frm2
from _fixtures.remote_exchanges import debind1_1        # Consistent author
from _fixtures.remote_exchanges import debind1_F        # Inconsistent author
from _fixtures.remote_exchanges import debind3_1        # Unknown author
from _fixtures.remote_exchanges import debindR_1
from _fixtures.remote_exchanges import debindR_F
from _fixtures.remote_exchanges import debind3_TF
from _fixtures.remote_exchanges import dyndebind1_1
from _fixtures.remote_exchanges import dyndebind1_F
from _fixtures.remote_exchanges import dedebind1_1
from _fixtures.remote_exchanges import dededebind1_1
from _fixtures.remote_exchanges import handshake1_1     # Known recipient
from _fixtures.remote_exchanges import handshake3_1     # Unknown recipient
    from hypergolix import logutils
'''
LICENSING
-------------------------------------------------

hypergolix: A python Golix client.
    Copyright (C) 2016 Muterra, Inc.
    
    Contributors
    ------------
    Nick Badger
        badg@muterra.io | badg@nickbadger.com | nickbadger.com

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.

    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the
    Free Software Foundation, Inc.,
    51 Franklin Street,
    Fifth Floor,
    Boston, MA  02110-1301 USA

------------------------------------------------------

'''

# Just used for fixture


# These are normal imports






# These are abnormal imports



# ###############################################
# Fixture imports
# ###############################################




# Containers
# Static bindings
# Dynamic bindings
# Debindings
# Requests


# Identities
gidclite1 = _GidcLite.from_golix(GIDC.unpack(gidc1))
gidclite2 = _GidcLite.from_golix(GIDC.unpack(gidc2))

# Containers
obj1 = _GeocLite.from_golix(cont1_1)
obj2 = _GeocLite.from_golix(cont1_2)
obj3 = _GeocLite.from_golix(cont3_1)

# Static bindings
sbind1 = _GobsLite.from_golix(bind1_1)

sbind3 = _GobsLite.from_golix(bind3_1)