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 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 Oracle(metaclass=API):
    ''' Source for total internal truth and state tracking of objects.
    
    Maintains <ghid>: <obj> lookup. Used by dispatchers to track obj
    state. Might eventually be used by AgentBase. Just a quick way to
    store and retrieve any objects based on an associated ghid.
    '''
    # These are actually used by the oracle itself
    _salmonator = weak_property('__salmonator')
    
    # These are only here to pass along to GAOs
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    
    def __init__(self, *args, **kwargs):
        ''' Sets up internal tracking.
        '''
        super().__init__(*args, **kwargs)
        self._lookup = {}
        
    @fixture_api
    def RESET(self):
        ''' Simply re-call init.
        '''
        self._lookup.clear()
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        
    @fixture_api
    def add_object(self, ghid, obj):
        ''' Add an object to the fixture.
        '''
        self._lookup[ghid] = obj
            
    @public_api
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Get an object.
        '''
        if ghid in self._lookup:
            # TODO: this is bad, because we're suppresing any potential argsig
            # problems. We should fix the abstraction so that it doesn't break
            # assumptions like that.
            obj = self._lookup[ghid]
            logger.debug(
                'GAO ' + str(ghid) + ' already exists in Oracle memory as: ' +
                str(type(obj))
            )
            
            if not isinstance(obj, gaoclass):
                raise TypeError(
                    'Object has already been resolved, and is not the '
                    'correct GAO class.'
                )
            else:
                logger.debug(
                    'GAO ' + str(ghid) + ' acceptable for ' + str(gaoclass)
                )
                
            await obj._ctx.wait()
            if ghid not in self._lookup:
                raise RuntimeError('Contentious delete while getting object.')
        
        else:
            logger.info(
                'GAO ' + str(ghid) +
                ' not currently in Oracle memory. Attempting load as: ' +
                str(gaoclass)
            )
            
            # First create the actual GAO. We do not need to have the ghid
            # downloaded to do this -- object creation is just making a Python
            # object locally.
            obj = gaoclass(
                ghid,
                None,   # dynamic
                None,   # author
                7,      # legroom (will be overwritten by pull)
                *args,
                golcore = self._golcore,
                ghidproxy = self._ghidproxy,
                privateer = self._privateer,
                percore = self._percore,
                librarian = self._librarian,
                **kwargs
            )
            
            # TODO: fix leaky abstraction
            obj._ctx = asyncio.Event()
            
            # Always do this first to make sure we have the most recent version
            # in all subsequent calls, without a race condition.
            self._lookup[ghid] = obj
            
            try:
                # Now immediately subscribe to the object upstream, so that there
                # is no race condition getting updates
                await self._salmonator.register(ghid)
                # Add deregister as a finalizer, but don't call it atexit. TODO:
                # fix leaky abstraction
                finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                             ghid)
                finalizer.atexit = False
                # Explicitly pull the object from salmonator to ensure we have
                # the newest version, and that it is available locally in
                # librarian if also available anywhere else. Note that
                # salmonator handles modal switching for dynamic/static. It
                # will also (by default, with skip_refresh=False) pull in any
                # updates that have accumulated in the meantime.
                await self._salmonator.attempt_pull(ghid, quiet=True)
                
                # Now actually fetch the object. This may KeyError if the
                # ghid is still unknown.
                await obj._pull()
            
            # Got an exception? Revert the lookup and reraise
            except Exception:
                del self._lookup[ghid]
                raise
                
            # We have to release other waiters regardless
            finally:
                # Finally, let future callers unblock
                obj._ctx.set()
            
        return obj
        
    @get_object.fixture
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Do the easy thing and just pull it out of lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            ghid,
            None,   # dynamic
            None,   # author
            7,      # legroom (will be overwritten by pull)
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
            
        return self._lookup[ghid]
        
    @public_api
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Creates a new object and returns it. Passes all *kwargs to
        the declared gao_class. Requires a zeroth state, and calls push
        internally.
        '''
        obj = gaoclass(
            None,                   # ghid
            dynamic,
            self._golcore.whoami,   # author
            legroom,
            *args,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            **kwargs
        )
        await obj._push()
        
        # This is used strictly by oracle. TODO: fix leaky abstraction
        obj._ctx = asyncio.Event()
        obj._ctx.set()
        # Do this before registering with salmonator, in case the latter errors
        self._lookup[obj.ghid] = obj
        
        # Finally, register to receive any concurrent updates from other
        # simultaneous sessions, and then return the object
        await self._salmonator.register(obj.ghid)
        # Add deregister as a finalizer, but don't call it atexit. TODO: fix
        # leaky abstraction
        finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                     obj.ghid)
        finalizer.atexit = False
        
        return obj
            
    @new_object.fixture
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Relies upon add_object, but otherwise just pops something
        from the lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            None,               # ghid
            dynamic,
            None,               # author (normally assigned through whoami)
            legroom,
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
        
        ghid, obj = self._lookup.popitem()
        self._lookup[ghid] = obj
        return obj
        
    def forget(self, ghid):
        ''' Removes the object from the cache. Next time an application
        wants it, it will need to be acquired from persisters.
        
        Indempotent; will not raise KeyError if called more than once.
        '''
        try:
            del self._lookup[ghid]
        except KeyError:
            logger.debug(str(ghid) + ' unknown to oracle.')
            
    def __contains__(self, ghid):
        ''' Checks for the ghid in cache (but does not check for global
        availability; that would require checking the persister for its
        existence and the privateer for access).
        '''
        return ghid in self._lookup class _Dispatchable(GAOCore, metaclass=API):
    ''' A dispatchable object. Note that privacy is store within the
    dispatcher itself, and not within the individual dispatchables.
    '''
    _dispatch = weak_property('__dispatch')
    _account = weak_property('__account')
    _ipc_protocol = weak_property('__ipc_protocol')
    api_id = immutable_property('_api_id')
    
    def __init__(self, *args, api_id, state, dispatch, ipc_protocol, account,
                 **kwargs):
        super().__init__(*args, **kwargs)
        
        self._dispatch = dispatch
        self._ipc_protocol = ipc_protocol
        self._account = account
        
        # These may be explicitly set to None to allow pull() to apply them.
        if state is not None:
            self.state = state
            
        if api_id is not None:
            self.api_id = api_id
            
    @public_api
    async def _push(self):
        ''' Extend GAOCore._push() to flush the account upon completion.
        '''
        await super()._push()
        await self._account.flush()
        
    @_push.fixture
    async def _push(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._push()
    
    @public_api
    async def _pull(self):
        ''' Extend GAOCore._pull() to notify IPC about updates.
        '''
        await super()._pull()
        make_background_future(
            self._dispatch.distribute_update(self.ghid)
        )
        
    @_pull.fixture
    async def _pull(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._pull()
        
    async def pack_gao(self):
        ''' Packs state into a bytes object. May be overwritten in subs
        to pack more complex objects. Should always be a staticmethod or
        classmethod.
        '''
        version = b'\x00'
        return b'hgxd' + version + bytes(self.api_id) + self.state
        
    async def unpack_gao(self, packed):
        ''' Unpacks state from a bytes object. May be overwritten in
        subs to unpack more complex objects. Should always be a
        staticmethod or classmethod.
        '''
        magic = packed[0:4]
        version = packed[4:5]
        api_id = ApiID.from_bytes(packed[5:70])
        
        if magic != b'hgxd':
            raise DispatchError('Object does not appear to be dispatchable.')
        elif version != b'\x00':
            raise DispatchError('Incompatible dispatchable version number.')
        elif getattr(self, 'api_id', None) is None:
            self.api_id = api_id
        else:
            # YES, this does need to be _api_id (with underscore!) because
            # we may need to override something existing.
            self._api_id = api_id
        # Checking for this from other people isn't really that useful. We'll
        # already error out if you try to change it locally, and if someone
        # goes to the effort of changing it elsewhere, well, more power to them
        # I guess.
        # elif api_id != self.api_id:
        #     raise DispatchError('Cannot change API ID.')
            
        self.state = packed[70:]
    
    @fixture_noop
    @public_api
    async def apply_delete(self, debinding):
        ''' Extend GAOCore.apply_delete to notify consumers of deletion.
        '''
        await super().apply_delete(debinding)
        make_background_future(
            self._dispatch.distribute_update(self.ghid, deleted=True)
        )
        
    def __eq__(self, other):
        ''' Ensure the other object is the same gao with the same state.
        Intended pretty much exclusively for use in testing.
        '''
        equal = True
        
        try:
            equal &= (self.dynamic == other.dynamic)
            equal &= (self.ghid == other.ghid)
            equal &= (self.author == other.author)
            equal &= (self.state == other.state)
            equal &= (self.api_id == other.api_id)
            
        except AttributeError:
            equal = False
        
        return equal
        
    def __hash__(self):
        ''' Dispatchable hashes should mix the hashes of the api_id, the
        author, and the ghid together, but only if all of the above are
        defined.
        '''
        defined = self.ghid is not None
        defined &= self.author is not None
        defined &= self.api_id is not None
        
        if not defined:
            raise TypeError(
                'Dispatchable objects may only be hashed after they are ' +
                'fully defined (with a ghid, an author, and an api_id.'
            )
        
        return hash(self.ghid) ^ hash(self.author) ^ hash(self.api_id) class Dispatcher(metaclass=API):
    ''' The Dispatcher decides which objects should be delivered where.
    This is decided through either:
    
    1. An API identifier (schema) dictating general compatibility as a
        dispatchable
    2. Declaring an object to be private, at which point the Dispatcher
        will internally (though distributedly) maintain that object to
        be exclusively available to an app token.
        
    Objects, once declared as non-private, cannot be retroactively
    privatized. That cat has officially left the bag/building/rodeo.
    However, a private object can be made non-private at a later time,
    provided it has defined an API ID.
    
    Private objects do not track their owning app tokens internally.
    Instead, this is managed through a Dispatcher-internal GAO. As such,
    even a bug resulting in a leaked private object will not result in a
    leaked app token.
    
    Ideally, the dispatcher will eventually also enforce whatever user
    restrictions on sharing are desired BETWEEN INSTALLED APPLICATIONS.
    Sharing restrictions between external parties are within the purview
    of the Rolodex.
    
    TODO: support notification mechanism to push new objects to other
    concurrent hypergolix instances. See note in ipc.ipccore.send_object
    
    TODO: move most of the misc stuff into dedicated new_object and
    get_object methods within dispatch, which (for the most part) just
    pass on straight to the oracle, but add all of the extra tracking
    stuff (and so on) internally.
    '''
    _account = weak_property('__account')
    _ipc_protocol = weak_property('__ipc_protocol')
    _oracle = weak_property('__oracle')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Yup yup yup yup yup yup yup
        '''
        super().__init__(*args, **kwargs)
        
        # Temporarily set distributed state to None.
        # Lookup for all known tokens: set(<tokens>)
        self._all_known_tokens = None
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token = None
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = None
        # Distributed lock for adding app tokens
        self._token_lock = None
        
        # Set of incoming shared ghids that had no conn
        # set(<ghid, sender tuples>)
        self._orphan_incoming_shares = None
        # Setmap-like lookup for share acks that had no conn
        # <app token>: set(<ghids>)
        self._orphan_share_acks = None
        # Setmap-like lookup for share naks that had no conn
        # <app token>: set(<ghids>)
        self._orphan_share_naks = None
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token = weakref.WeakValueDictionary()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn = weakref.WeakKeyDictionary()
        
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api = WeakSetMap()
        
        # This lookup directly tracks who has a copy of the object
        # Lookup <object ghid>: set(<connection/session/conn>)
        self._update_listeners = WeakSetMap()
        # This lookup preserves a strong reference to the dispatchable **as the
        # key**, allowing the object to be GC'd from local memory only when no
        # connections remain that have it
        self._obj_binding = WeakSetMap()
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Create a dispatch fixture.
        '''
        super(Dispatcher.__fixture__, self).__init__(*args, **kwargs)
        
        self._all_known_tokens = set()
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token = {}
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = {}
        # Distributed lock for adding app tokens
        self._token_lock = NoContext()
        
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api = WeakSetMap()
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token = weakref.WeakValueDictionary()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn = weakref.WeakKeyDictionary()
        
    @fixture_api
    def RESET(self):
        ''' Reset the fixture to a pristine state.
        '''
        self._all_known_tokens.clear()
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token.clear()
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid.clear()
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api.clear_all()
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token.clear()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn.clear()
        
    def assemble(self, oracle, ipc_protocol):
        self._oracle = oracle
        self._ipc_protocol = ipc_protocol
        
    @public_api
    def bootstrap(self, account):
        ''' Initialize distributed state.
        '''
        self._account = account
        
        # Now init distributed state.
            
        # Lookup for all known tokens: set(<tokens>)
        self._all_known_tokens = account.dispatch_tokens
        # Lookup (set-map-like) for <app token>: set(<startup ghids>)
        self._startup_by_token = account.dispatch_startup
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = account.dispatch_private
        
        # These need to be distributed but aren't yet. TODO!
        # Distributed lock for adding app tokens
        self._token_lock = NoContext()
        
        # Set of incoming shared ghids that had no conn
        # set(<ghid, sender tuples>)
        self._orphan_incoming_shares = account.dispatch_incoming
        # Setmap-like lookup for share acks that had no conn
        # <app token>: set(<ghid, recipient tuples>)
        self._orphan_share_acks = account.dispatch_orphan_acks
        # Setmap-like lookup for share naks that had no conn
        # <app token>: set(<ghid, recipient tuples>)
        self._orphan_share_naks = account.dispatch_orphan_naks
        
        # All known tokens must already contain a key for the null token, which
        # is used by dispatch to indicate a lack of a token in an exchange. Do
        # this as LBYL to avoid false positive _mutated flags in the GAO.
        if AppToken.null() not in self._all_known_tokens:
            self._all_known_tokens.add(AppToken.null())
            
    @public_api
    def token_lookup(self, connection):
        ''' Return the token associated with the connection, or None if
        there is no currently defined token.
        '''
        try:
            return self._token_from_conn[connection]
            
        except KeyError as exc:
            return None
            
    @public_api
    def connection_lookup(self, token):
        ''' Returns the current connection associated with the token, or
        None if there is no currently available connection.
        '''
        try:
            return self._conn_from_token[token]
            
        except KeyError as exc:
            return None
    
    @public_api
    def private_parent_lookup(self, ghid):
        ''' Returns the app_token parent for the passed ghid, if (and
        only if) it's private. Otherwise, returns None.
        '''
        try:
            return self._private_by_ghid[ghid]
        except KeyError:
            return None
    
    @public_api
    async def make_public(self, ghid):
        ''' Makes a private object public.
        '''
        try:
            del self._private_by_ghid[ghid]
            
        except KeyError as exc:
            raise ValueError(
                'Obj w/ passed ghid is unknown or already public: ' + str(ghid)
            ) from exc
            
        else:
            await self._account.flush()
            
    @make_public.fixture
    async def make_public(self, ghid):
        # Bypass all validation
        del self._private_by_ghid[ghid]
        
    @public_api
    async def add_api(self, connection, api_id):
        ''' Register the connection as currently tracking the api_id.
        '''
        self._conns_from_api.add(api_id, connection)
        await self._account.flush()
        
    @add_api.fixture
    async def add_api(self, connection, api_id):
        # Bypass account flushing
        self._conns_from_api.add(api_id, connection)
        
    @public_api
    async def remove_api(self, connection, api_id):
        ''' Remove a connection's registration for the api_id. Happens
        automatically when connections are GC'd.
        '''
        self._conns_from_api.discard(api_id, connection)
        await self._account.flush()
        
    @remove_api.fixture
    async def remove_api(self, connection, api_id):
        # Bypass account flushing
        self._conns_from_api.discard(api_id, connection)
        
    def _make_new_token(self):
        token = None
        # Birthday paradox be damned; we can actually *enforce* uniqueness
        while token is None or token in self._all_known_tokens:
            token = AppToken(os.urandom(AppToken.TOKEN_LEN))
        return token
        
    @public_api
    async def start_application(self, connection, token=None):
        ''' Ensures that an application is known to the dispatcher.
        
        Currently just checks within all known tokens. In the future,
        this be responsible for starting the application (unless we move
        that responsibility elsewhere), and then sending all of the
        startup objects to the application.
        '''
        if connection in self._token_from_conn:
            raise ExistantAppError('Token already registered for connection.')
            
        elif token in self._conn_from_token:
            raise ExistantAppError(
                'Token already started at different connection.'
            )
            
        elif token is None:
            # Token lock is currently a NoContext, but eventually will need to
            # be a distributed lock -- or, the token lookup needs to be made
            # into a no-contention system. Which I'm not sure it possibly can
            # be, since we're best off actually ENFORCING the "no birthday
            # collisions" for tokens.
            with self._token_lock:
                token = self._make_new_token()
                # Do this right away to prevent race condition
                self._all_known_tokens.add(token)
                await self._account.flush()
            
        elif token not in self._all_known_tokens:
            raise UnknownToken('App token unknown to dispatcher.')
            
        # TODO: should these be enclosed within an operations lock?
        self._conn_from_token[token] = connection
        self._token_from_conn[connection] = token
        
        return token
        
    @start_application.fixture
    async def start_application(self, connection, token=None):
        if token is None:
            # ONLY FOR THE FIXTURE! Do we use a pseudorandom token.
            token = AppToken.pseudorandom()
        
        # Don't emulate normal dispatcher behavior here; let everything start,
        # regardless of status re: "exists in tokens"
        self._all_known_tokens.add(token)
        self._conn_from_token[token] = connection
        self._token_from_conn[connection] = token
        
        return token
    
    @fixture_noop
    @public_api
    async def track_object(self, connection, ghid):
        ''' Registers a connection as tracking a ghid.
        
        This is necessary so that upstream updates can be properly
        dispatched to any applications with copies of the object.
        '''
        logger.debug(
            'CONN ' + str(connection) + ' tracking ' + str(ghid) + '...'
        )
        self._update_listeners.add(ghid, connection)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        # This keeps the object in memory, allowing it to receive subscriptions
        # When all apps untrack the object, it gets unsubbed.
        self._obj_binding.add(obj, connection)
    
    @fixture_noop
    @public_api
    async def untrack_object(self, connection, ghid):
        ''' Remove a connection as tracking a ghid.
        
        This indicates an application no longer has a copy of the
        object, therefore silencing any updates it would otherwise have
        received.
        '''
        logger.debug(
            'CONN ' + str(connection) + ' untracking ' + str(ghid) + '...'
        )
        self._update_listeners.discard(ghid, connection)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        # This keeps the object in memory, allowing it to receive subscriptions
        # When all apps untrack the object, it gets unsubbed.
        self._obj_binding.discard(obj, connection)
        
    @public_api
    async def register_object(self, connection, ghid, private):
        ''' Call this every time a new object is created to register it
        with the dispatcher, recording it as private or distributing it
        to other applications as needed.
        '''
        # If the object is private, register it as such.
        if private:
            try:
                token = self._token_from_conn[connection]
            
            except KeyError as exc:
                raise UnknownToken(
                    'Must register app token before creating private objects.'
                ) from exc
                
            else:
                logger.debug(
                    'Creating private object for ' + str(connection) +
                    '; bypassing distribution.'
                )
                self._private_by_ghid[ghid] = token
            
        # Otherwise, make sure to notify any other interested parties.
        else:
            make_background_future(
                self.distribute_share(
                    ghid,
                    origin = None,
                    skip_conn = connection
                )
            )
            
    @register_object.fixture
    async def register_object(self, connection, ghid, private):
        ''' Just don't distribute the object.
        '''
        if private:
            try:
                token = self._token_from_conn[connection]
            
            except KeyError as exc:
                raise UnknownToken(
                    'Must register app token before creating private objects.'
                ) from exc
                
            else:
                logger.debug(
                    'Creating private object for ' + str(connection) +
                    '; bypassing distribution.'
                )
                self._private_by_ghid[ghid] = token
            
    @public_api
    async def register_startup(self, connection, ghid):
        ''' Registers a ghid to be used as a startup object for token.
        '''
        try:
            token = self._token_from_conn[connection]
            
        except KeyError as exc:
            raise UnknownToken(
                'Must register app token before registering startup objects.'
            ) from exc
            
        else:
            if token in self._startup_by_token:
                raise ValueError(
                    'Startup object already defined for that application. '
                    'Deregister it before registering a new startup object.'
                )
            else:
                self._startup_by_token[token] = ghid
                await self._account.flush()
                
    @public_api
    async def deregister_startup(self, connection):
        ''' Deregisters a ghid to be used as a startup object for token.
        '''
        try:
            token = self._token_from_conn[connection]
            
        except KeyError as exc:
            raise UnknownToken(
                'Must register app token before deregistering startup objects.'
            ) from exc
            
        else:
            if token not in self._startup_by_token:
                raise ValueError(
                    'Startup object has not been defined for that application.'
                )
            else:
                del self._startup_by_token[token]
                await self._account.flush()
    
    @public_api
    async def get_startup_obj(self, token):
        ''' Returns the ghid of the declared startup object for that
        token, or None if none has been declared.
        '''
        if token not in self._all_known_tokens:
            raise UnknownToken()
        
        elif token not in self._startup_by_token:
            return None
        
        else:
            return self._startup_by_token[token]
    
    @fixture_noop
    @public_api
    async def distribute_share(self, ghid, origin, skip_conn=None):
        ''' Perform an actual share distribution.
        
        Pass an explicit None as the origin to indicate self as origin.
        '''
        callsheet = set()
        
        # Note that this is called in exactly two situations: when creating a
        # new object locally, through register_object, or when receiving a new
        # share, through rolodex. As such, this will never be called when the
        # object is private. So, don't check for a private owner.
        
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        
        callsheet.update(
            # Get any connections that have registered the api_id
            self._conns_from_api.get_any(obj.api_id)
        )
        callsheet.update(
            # Get any connections that have an instance of the object
            self._update_listeners.get_any(obj.ghid)
        )
            
        # Now, check to see if we have anything in the callsheet. Do it before
        # discarding, so that we know if it's actually an orphan share.
        if callsheet:
            # Discard the skipped connection, if one is defined
            callsheet.discard(skip_conn)
            str_origin = origin or 'self'
            logger.debug(
                'Distributing ' + str(ghid) + ' shared object from ' +
                str(str_origin) + ' to ' + repr(callsheet)
            )
            
            # If we still have a callsheet, distribute it.
            if callsheet:
                await self._distribute(
                    self._ipc_protocol.share_obj,    # distr_coro
                    callsheet,
                    ghid,
                    origin,
                    obj.api_id
                )
        
        # Only make an orphan_shares entry if we didn't create the object.
        elif origin is not None:
            sharelog = _ShareLog(ghid, origin)
            self._orphan_incoming_shares.add(sharelog)
            await self._account.flush()
    
    @fixture_noop
    @public_api
    async def distribute_update(self, ghid, deleted=False, skip_conn=None):
        ''' Perform an actual update distribution.
        '''
        # Get any connections that have an instance of the object
        callsheet = set()
        # Note that this call returns a frozenset, hence the copy.
        callsheet.update(self._update_listeners.get_any(ghid))
        # Skip the connection if one is passed.
        callsheet.discard(skip_conn)
        
        logger.debug(
            'Distributing {!s} update to {!r}.'.format(ghid, callsheet)
        )
        
        if deleted:
            await self._distribute(
                self._ipc_protocol.delete_obj,   # distr_coro
                callsheet,
                ghid
            )
            
        else:
            await self._distribute(
                self._ipc_protocol.update_obj,   # distr_coro
                callsheet,
                ghid
            )
            
    @fixture_noop
    @public_api
    async def distribute_share_success(self, ghid, recipient, tokens):
        ''' Notify any available connections for all passed <tokens> of
        a share failure.
        '''
        callsheet = set()
        for token in tokens:
            # Escape any keys that have gone missing during the rat race
            try:
                callsheet.add(self._conn_from_token[token])
            except KeyError:
                logger.info('No connection for token ' + str(token))
        
        # Distribute the share failure to all apps that requested its delivery
        await self._distribute(
            self._ipc_protocol.notify_share_success,   # distr_coro
            callsheet,
            ghid,
            recipient
        )
    
    @fixture_noop
    @public_api
    async def distribute_share_failure(self, ghid, recipient, tokens):
        ''' Notify any available connections for all passed <tokens> of
        a share failure.
        '''
        callsheet = set()
        for token in tokens:
            # Escape any keys that have gone missing during the rat race
            try:
                callsheet.add(self._conn_from_token[token])
            except KeyError:
                logger.info('No connection for token ' + str(token))
        
        # Distribute the share failure to all apps that requested its delivery
        await self._distribute(
            self._ipc_protocol.notify_share_failure,   # distr_coro
            callsheet,
            ghid,
            recipient
        )
                
    async def _distribute(self, distr_coro, callsheet, *args, **kwargs):
        ''' Call distr_coro with *args and **kwargs at each of the
        connections in the callsheet. Log (but don't raise) any errors.
        '''
        logger.info(
            'Dispatching ' + str(len(callsheet)) + ' calls to ' +
            str(distr_coro)
        )
        distributions = set()
        for connection in callsheet:
            # For each connection...
            distributions.add(
                # ...in parallel, schedule a single execution of the
                # distribution coroutine.
                make_background_future(distr_coro(connection, *args, **kwargs))
            )
            
        # And gather the results, logging (but not raising) any exceptions.
        # BUT, only do it if we have some to do.
        if distributions:
            try:
                await asyncio.wait(
                    fs = distributions,
                    return_when = asyncio.ALL_COMPLETED
                )
            
            except asyncio.CancelledError:
                logger.debug('Cancelling pending distributions.')
                for distr in distributions:
                    distr.cancel()
                raise 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 Rolodex(metaclass=API):
    ''' Handles sharing, requests, etc.
    
    In the future, will maintain a contacts list to approve/reject
    incoming requests. In the further future, will maintain sharing
    pipelines, negotiated through handshakes, to perform sharing
    symmetrically between contacts.
    '''
    _account = weak_property('__account')
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    _dispatch = weak_property('__dispatch')
    
    @public_api
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = None
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = None
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Init the fixture.
        '''
        super(Rolodex.__fixture__, self).__init__(*args, **kwargs)
        
        self.shared = {}
        self._pending_requests = {}
        self._outstanding_shares = {}
        
    @fixture_api
    def RESET(self):
        ''' Super simple reset alias to __init__.
        '''
        self.shared.clear()
        self._pending_requests.clear()
        self._outstanding_shares.clear()
        
    @fixture_noop
    @public_api
    def bootstrap(self, account):
        ''' Initialize distributed state.
        '''
        self._account = account
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = account.rolodex_pending
        
        # These need to be distributed but aren't yet. TODO!
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = account.rolodex_outstanding
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator, dispatch):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        self._dispatch = dispatch
        
    @public_api
    async def share_object(self, target, recipient, requesting_token):
        ''' Share a target ghid with the recipient.
        '''
        if not isinstance(target, Ghid):
            raise TypeError(
                'target must be Ghid or similar.'
            )
        if not isinstance(recipient, Ghid):
            raise TypeError(
                'recipient must be Ghid or similar.'
            )

        sharepair = _SharePair(target, recipient)
            
        # For now, this is just doing a handshake with some typechecking.
        await self._hand_object(*sharepair)
        
        if requesting_token is not None:
            self._outstanding_shares.add(sharepair, requesting_token)
            await self._account.flush()
            
    @share_object.fixture
    async def share_object(self, target, recipient, requesting_token):
        ''' Pretty basic.
        '''
        # Ignore requesting token, because we won't generally set it when
        # testing this.
        self.shared[target] = recipient
        
    async def _hand_object(self, target, recipient):
        ''' Initiates a handshake request with the recipient.
        '''
        if not (await self._librarian.contains(recipient)):
            await self._salmonator.attempt_pull(recipient, quiet=True)
        
        recipient_identity = \
            (await self._librarian.summarize(recipient)).identity
        
        # This is guaranteed to resolve the container fully.
        container_ghid = await self._ghidproxy.resolve(target)
        
        # TODO: fix Golix library so this isn't such a shitshow re:
        # breaking abstraction barriers.
        handshake = self._golcore._identity.make_handshake(
            target = target,
            secret = self._privateer.get(container_ghid)
        )
        
        request = self._golcore._identity.make_request(
            recipient = recipient_identity,
            request = handshake
        )
        
        # Note that this must be called before publishing to the persister, or
        # there's a race condition between them.
        self._pending_requests[request.ghid] = target, recipient
        
        # Note the potential race condition here. Should catch errors with the
        # persister in case we need to resolve pending requests that didn't
        # successfully post.
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(request),
            packed = request.packed,
            remotable = True
        )
    
    @fixture_noop
    @public_api
    async def notification_handler(self, subscription, notification):
        ''' Callback to handle any requests.
        '''
        if not (await self._librarian.contains(notification)):
            await self._salmonator.attempt_pull(notification, quiet=True)
        
        # Note that the notification could also be a GDXX.
        request_or_debind = await self._librarian.summarize(notification)
        
        if isinstance(request_or_debind, _GarqLite):
            await self._handle_request(notification)
            
        elif isinstance(request_or_debind, _GdxxLite):
            # This case should only be relevant if we have multiple agents
            # logged in at separate locations at the same time, processing the
            # same GARQs.
            await self._handle_debinding(request_or_debind)
            
        else:
            raise RuntimeError(
                'Unexpected Golix primitive while listening for requests.'
            )
            
    async def _handle_request(self, notification):
        ''' The notification is an asymmetric request. Deal with it.
        '''
        try:
            packed = await self._librarian.retrieve(notification)
            unpacked = await self._golcore.unpack_request(packed)
        
            # TODO: have this filter based on contacts.
            if not (await self._librarian.contains(unpacked.author)):
                await self._salmonator.attempt_pull(
                    unpacked.author,
                    quiet = True
                )
        
            payload = await self._golcore.open_request(unpacked)
            await self._dispatch_payload(payload, notification)

        # Don't forget to (always) debind.
        finally:
            debinding = await self._golcore.make_debinding(notification)
            await self._percore.direct_ingest(
                obj = _GdxxLite.from_golix(debinding),
                packed = debinding.packed,
                remotable = True
            )
        
    async def _handle_debinding(self, debinding):
        ''' The notification is a debinding. Deal with it.
        '''
        # For now we just need to remove any pending requests for the
        # debinding's target.
        try:
            del self._pending_requests[debinding.target]
            await self._account.flush()
        except KeyError:
            logger.debug(
                'Request at ' + str(debinding.ghid) + ' target missing ' +
                'from pending requests: ' + str(debinding.target)
            )
        
    async def _dispatch_payload(self, payload, source_ghid):
        ''' Appropriately handles a request payload.
        '''
        if isinstance(payload, AsymHandshake):
            logger.debug(
                'Share handshake received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_handshake(payload, source_ghid)
            
        elif isinstance(payload, AsymAck):
            logger.debug(
                'Handshake ACK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_ack(payload)
            
        elif isinstance(payload, AsymNak):
            logger.debug(
                'Handshake NAK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_nak(payload)
            
        else:
            raise RuntimeError('Encountered an unknown request type.')
            
    async def _handle_handshake(self, request, source_ghid):
        ''' Handles a handshake request after reception.
        '''
        try:
            # First, we need to figure out what the actual container object's
            # address is, and then stage the secret for it.
            # ALWAYS attempt to pull it first, or we may have a stale frame
            # locally, and attempt to stage a new secret for a long-expired
            # frame
            await self._salmonator.attempt_pull(
                request.target,
                quiet = True
            )
            
            binding_or_obj = await self._librarian.summarize(request.target)
            if isinstance(binding_or_obj, _GobdLite):
                container_ghid = binding_or_obj.target
                self._privateer.quarantine(
                    ghid = container_ghid,
                    secret = request.secret
                )
                if not (await self._librarian.contains(container_ghid)):
                    await self._salmonator.attempt_pull(
                        container_ghid,
                        quiet = True
                    )
            elif isinstance(binding_or_obj, _GeocLite):
                self._privateer.quarantine(
                    ghid = binding_or_obj.ghid,
                    secret = request.secret
                )
            else:
                raise TypeError(
                    'Invalid handshake target: ' + str(type(binding_or_obj))
                )
            
            # Note that unless we raise a HandshakeError RIGHT NOW, we'll be
            # sending an ack to the handshake, just to indicate successful
            # receipt of the share. If the originating app wants to check for
            # availability, well, that's currently on them. In the future, add
            # handle for that in SHARE instead of HANDSHAKE?
            
        except asyncio.CancelledError:
            logger.debug('Handshake handler cancelled.')
            raise
            
        except Exception as exc:
            logger.error(
                'Exception encountered while handling a handshake. Returned a '
                'NAK.\n' + ''.join(traceback.format_exc())
            )
            # Erfolglos. Send a nak to whomever sent the handshake
            response_obj = self._golcore._identity.make_nak(
                target = source_ghid
            )
            
        else:
            # Success. Send an ack to whomever sent the handshake
            response_obj = self._golcore._identity.make_ack(
                target = source_ghid
            )
        
        response = await self._golcore.make_request(
            request.author,
            response_obj
        )
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(response),
            packed = response.packed,
            remotable = True
        )
            
        await self.share_handler(request.target, request.author)
            
    async def _handle_ack(self, request):
        ''' Handles a handshake ack after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received an ACK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_ack_handler(target, recipient)
            
    async def _handle_nak(self, request):
        ''' Handles a handshake nak after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received a NAK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_nak_handler(target, recipient)
    
    @fixture_noop
    @public_api
    async def share_handler(self, target, sender):
        ''' Incoming share targets (well, their ghids anyways) are
        forwarded to the _ipccore.
        
        Only separate from _handle_handshake right now because in the
        future, object sharing will be at least partly handled within
        its own dedicated rolodex pipeline.
        '''
        # Make sure we actually have the target
        if not (await self._librarian.contains(target)):
            await self._salmonator.attempt_pull(target, quiet=True)
        
        # Distribute the share in the background
        make_background_future(
            self._dispatch.distribute_share(target, sender)
        )
    
    @fixture_noop
    @public_api
    async def receipt_ack_handler(self, target, recipient):
        ''' Receives a share ack from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share success in the background
        make_background_future(
            self._dispatch.distribute_share_success(
                target,
                recipient,
                tokens
            )
        )
    
    @fixture_noop
    @public_api
    async def receipt_nak_handler(self, target, recipient):
        ''' Receives a share nak from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share failure in the background
        make_background_future(
            self._dispatch.distribute_share_failure(
                target,
                recipient,
                tokens
            )
        ) class Account(metaclass=API):
    ''' Accounts settle all sorts of stuff.
    
    TODO: move GolixCore into account. That way, everything to do with
    the private keys stays within the account, and is never directly
    accessed outside of it.
    '''
    _identity = immortal_property('__identity')
    _user_id = immortal_property('__user_id')
    
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _oracle = weak_property('__oracle')
    _rolodex = weak_property('__rolodex')
    _dispatch = weak_property('__dispatch')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    
    @public_api
    def __init__(self, user_id, root_secret, *args, hgxcore, **kwargs):
        ''' Gets everything ready for account bootstrapping.
        
        +   user_id explicitly passed with None means create a new
            Account.
        +   identity explicitly passed with None means load an existing
            account.
        +   user_id XOR identity must be passed.
        '''
        super().__init__(*args, **kwargs)
        
        if isinstance(user_id, Ghid):
            logger.info('Logging in with existing user ID: ' + str(user_id))
            self._identity = None
            self._user_id = user_id
        
        # This is used exclusively for testing.
        elif isinstance(user_id, FirstParty):
            logger.info('Creating a new account.')
            self._identity = user_id
            self._user_id = None
        
        else:
            raise TypeError('user_id must be an instance of Ghid or None.')
            
        self._golcore = hgxcore.golcore
        self._ghidproxy = hgxcore.ghidproxy
        self._privateer = hgxcore.privateer
        self._oracle = hgxcore.oracle
        self._rolodex = hgxcore.rolodex
        self._dispatch = hgxcore.dispatch
        self._percore = hgxcore.percore
        self._librarian = hgxcore.librarian
        self._salmonator = hgxcore.salmonator
        
        self._root_secret = root_secret
        
    @__init__.fixture
    def __init__(self, identity, *args, **kwargs):
        ''' Lulz just ignore errytang and skip calling super!
        '''
        self._identity = identity
        
        self.privateer_persistent = {}
        self.privateer_quarantine = {}
        
        self.rolodex_pending = {}
        self.rolodex_outstanding = SetMap()
        
        self.dispatch_tokens = set()
        self.dispatch_startup = {}
        self.dispatch_private = {}
        self.dispatch_incoming = set()
        self.dispatch_orphan_acks = SetMap()
        self.dispatch_orphan_naks = SetMap()
        
    @property
    def _fingerprint(self):
        ''' Get our fingerprint.
        '''
        return self._identity.ghid
        
    async def _inject_gao(self, gao):
        ''' Bypass the normal oracle get_object, new_object process and
        create the object directly.
        '''
        await self._salmonator.register(gao.ghid)
        await self._salmonator.attempt_pull(gao.ghid, quiet=True)
        gao._ctx = asyncio.Event()
        gao._ctx.set()
        self._oracle._lookup[gao.ghid] = gao
            
    async def bootstrap(self):
        ''' Used for account creation, to initialize the root node with
        its resource directory.
        '''
        # We need to pre-allocate the privateer stuff so we can bootstrap it
        # before pulling the root node when reloading.
        self.privateer_persistent = GAODict(
            ghid = None,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian
        )
        self.privateer_quarantine = GAODict(
            ghid = None,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian
        )
        
        # Privateer can be bootstrapped with or without pulling. Even though it
        # won't work *right* before pulling, it won't work AT ALL until it's
        # bootstrapped, so bootstrap first and pull later.
        logger.info('Bootstrapping privateer.')
        self._privateer.bootstrap(self)
        
        if self._user_id is not None:
            logger.info('Loading the root node.')
            root_node = GAO(
                ghid = self._user_id,
                dynamic = True,
                author = None,
                legroom = 7,
                state = b'you pass butter',
                golcore = self._golcore,
                ghidproxy = self._ghidproxy,
                privateer = self._privateer,
                percore = self._percore,
                librarian = self._librarian,
                master_secret = self._root_secret
            )
            # And now, remove the root secret from the parent namespace. This
            # will make the root_node GAO the only live reference to the root
            # secret from within the Account.
            del self._root_secret
            await root_node._pull()
                
            # TODO: convert all of this into a smartyparser (after rewriting
            # smartyparse, that is)
            
            # The last 64 bytes of the state is a SHA512 checksum for the whole
            # rest of the state
            password_validator = root_node.state[-64:]
            password_comparator = root_node.state[:-64]
            
            # This comparison is timing-insensitive: if attempting to brute
            # force a password, incorrect password attempts cannot extract any
            # information from the timing of the comparison.
            if sha512(password_comparator).digest() != password_validator:
                logger.critical('Incorrect password.')
                raise ValueError('Incorrect password.')
            
            # Identity
            identity_ghid = Ghid.from_bytes(
                root_node.state[0: 65])     # 65 bytes
            identity_master = Secret.from_bytes(
                root_node.state[65: 118])   # 53 bytes
            
            # Privateer
            privateer_persistent_ghid = Ghid.from_bytes(
                root_node.state[118: 183])
            privateer_persistent_master = Secret.from_bytes(
                root_node.state[183: 236])
            
            privateer_quarantine_ghid = Ghid.from_bytes(
                root_node.state[236: 301])
            privateer_quarantine_master = Secret.from_bytes(
                root_node.state[301: 354])
            
            # Rolodex stuff
            rolodex_pending_ghid = Ghid.from_bytes(
                root_node.state[354: 419])
            rolodex_pending_master = Secret.from_bytes(
                root_node.state[419: 472])
            
            rolodex_outstanding_ghid = Ghid.from_bytes(
                root_node.state[472: 537])
            rolodex_outstanding_master = Secret.from_bytes(
                root_node.state[537: 590])
            
            # Dispatch stuff
            dispatch_tokens_ghid = Ghid.from_bytes(
                root_node.state[590: 655])
            dispatch_tokens_master = Secret.from_bytes(
                root_node.state[655: 708])
            
            dispatch_startup_ghid = Ghid.from_bytes(
                root_node.state[708: 773])
            dispatch_startup_master = Secret.from_bytes(
                root_node.state[773: 826])
            
            dispatch_private_ghid = Ghid.from_bytes(
                root_node.state[826: 891])
            dispatch_private_master = Secret.from_bytes(
                root_node.state[891: 944])
            
            dispatch_incoming_ghid = Ghid.from_bytes(
                root_node.state[944: 1009])
            dispatch_incoming_master = Secret.from_bytes(
                root_node.state[1009: 1062])
            
            dispatch_orphan_acks_ghid = Ghid.from_bytes(
                root_node.state[1062: 1127])
            dispatch_orphan_acks_master = Secret.from_bytes(
                root_node.state[1127: 1180])
            
            dispatch_orphan_naks_ghid = Ghid.from_bytes(
                root_node.state[1180: 1245])
            dispatch_orphan_naks_master = Secret.from_bytes(
                root_node.state[1245: 1298])
        
        else:
            # We need an identity at to golcore before we can do anything
            logger.info('Bootstrapping golcore.')
            self._golcore.bootstrap(self)
            
            # Note that we first need to push our identity secondparty.
            packed = self._identity.second_party.packed
            gidc = _GidcLite.from_golix(GIDC.unpack(packed))
            logger.info('Publishing public keys.')
            # We haven't bootstrapped salmonator yet, so don't remote the keys
            await self._percore.direct_ingest(gidc, packed, remotable=False)
            
            # Now we need to declare the root note. It needs some kind of
            # useless, garbage data as well, lest it error out because it can't
            # encrypt nothingness.
            root_node = GAO(
                ghid = None,
                dynamic = True,
                author = None,
                legroom = 7,
                state = b'you pass butter',
                golcore = self._golcore,
                ghidproxy = self._ghidproxy,
                privateer = self._privateer,
                percore = self._percore,
                librarian = self._librarian,
                master_secret = self._root_secret
            )
            # And now, remove the root secret from the parent namespace. This
            # will make the root_node GAO the only live reference to the root
            # secret from within the Account.
            del self._root_secret
            
            identity_ghid = None
            identity_master = self._identity.new_secret()
            
            privateer_persistent_ghid = None
            privateer_persistent_master = self._identity.new_secret()
            
            privateer_quarantine_ghid = None
            privateer_quarantine_master = self._identity.new_secret()
            
            # Rolodex stuff
            rolodex_pending_ghid = None
            rolodex_pending_master = self._identity.new_secret()
            
            rolodex_outstanding_ghid = None
            rolodex_outstanding_master = self._identity.new_secret()
            
            # Dispatch stuff
            dispatch_tokens_ghid = None
            dispatch_tokens_master = self._identity.new_secret()
            
            dispatch_startup_ghid = None
            dispatch_startup_master = self._identity.new_secret()
            
            dispatch_private_ghid = None
            dispatch_private_master = self._identity.new_secret()
            
            dispatch_incoming_ghid = None
            dispatch_incoming_master = self._identity.new_secret()
            
            dispatch_orphan_acks_ghid = None
            dispatch_orphan_acks_master = self._identity.new_secret()
            
            dispatch_orphan_naks_ghid = None
            dispatch_orphan_naks_master = self._identity.new_secret()
        
        # Allocate the identity container
        #######################################################################
        # This stores the private keys
        identity_container = GAODict(
            ghid = identity_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = identity_master
        )
        
        # Jiggery-pokery for the privateer restoration
        #######################################################################
        # This stores persistent secrets
        self.privateer_persistent.ghid = privateer_persistent_ghid
        self.privateer_persistent._inject_msec(privateer_persistent_master)
        
        self.privateer_quarantine.ghid = privateer_quarantine_ghid
        self.privateer_quarantine._inject_msec(privateer_quarantine_master)
        
        # Save/load the identity container and bootstrap golcore, privateer
        #######################################################################
        
        # Load existing account
        if self._user_id is not None:
            logger.info('Loading identity.')
            await identity_container._pull()
            self._identity = FirstParty._from_serialized(
                identity_container.state
            )
            logger.info('Bootstrapping golcore.')
            self._golcore.bootstrap(self)
            logger.info('Bootstrapping salmonator.')
            await self._salmonator.bootstrap(self)
            
            logger.info('Loading persistent keystore.')
            await self.privateer_persistent._pull()

            logger.info('Loading quarantined keystore.')
            await self.privateer_quarantine._pull()
            
        # Save new account
        else:
            # First, bootstrap salmonator, so that we don't get errors as an
            # unknown author
            await self._salmonator.bootstrap(self)
            # Initializing is needed to prevent losing the first frame while
            # the secret ratchet initializes
            logger.info('Allocating the root node.')
            await root_node._push()
            # Don't forget to do this before creating the actual identity
            # frame, or we'll lose the first frame (and therefore, everything
            # we care about)
            logger.info('Allocating identity container.')
            await identity_container._push(force=True)
            logger.info('Saving identity.')
            identity_container.update(self._identity._serialize())
            await identity_container._push(force=True)
            
            # Because these use a master secret, they need to be initialized,
            # or the first frame will be unrecoverable.
            logger.info('Allocating persistent keystore.')
            await self.privateer_persistent._push(force=True)
            await self.privateer_persistent._push(force=True)
            
            logger.info('Allocating quarantined keystore.')
            await self.privateer_quarantine._push(force=True)
            await self.privateer_quarantine._push(force=True)
            
        # Establish the rest of the above at the various tracking agencies
        logger.info('Reticulating keystores.')
        await self._inject_gao(self.privateer_persistent)
        await self._inject_gao(self.privateer_quarantine)
        # We don't need to do this with the secondary manifest (unless we're
        # planning on adding things to it while already running, which would
        # imply an ad-hoc, on-the-fly upgrade process)
        
        # Rolodex gaos:
        self.rolodex_pending = GAODict(
            ghid = rolodex_pending_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = rolodex_pending_master
        )
        self.rolodex_outstanding = GAOSetMap(
            ghid = rolodex_outstanding_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = rolodex_outstanding_master
        )
        
        # Dispatch gaos:
        self.dispatch_tokens = GAOSet(
            ghid = dispatch_tokens_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_tokens_master
        )
        self.dispatch_startup = GAODict(
            ghid = dispatch_startup_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_startup_master
        )
        self.dispatch_private = GAODict(
            ghid = dispatch_private_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_private_master
        )
        self.dispatch_incoming = GAOSet(
            ghid = dispatch_incoming_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_incoming_master
        )
        self.dispatch_orphan_acks = GAOSetMap(
            ghid = dispatch_orphan_acks_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_orphan_acks_master
        )
        self.dispatch_orphan_naks = GAOSetMap(
            ghid = dispatch_orphan_naks_ghid,
            dynamic = True,
            author = None,
            legroom = 7,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            master_secret = dispatch_orphan_naks_master
        )
        
        # These need not have the actual objects pulled yet
        self._rolodex.bootstrap(self)
        self._dispatch.bootstrap(self)
        
        if self._user_id is not None:
            logger.info('Restoring sharing subsystem.')
            await self.rolodex_pending._pull()
            await self.rolodex_outstanding._pull()
            
            logger.info('Restoring object dispatch.')
            await self.dispatch_tokens._pull()
            await self.dispatch_startup._pull()
            await self.dispatch_private._pull()
            await self.dispatch_incoming._pull()
            await self.dispatch_orphan_acks._pull()
            await self.dispatch_orphan_naks._pull()
        
        else:
            logger.info('Building sharing subsystem.')
            await self.rolodex_pending._push(force=True)
            await self.rolodex_pending._push(force=True)
            await self.rolodex_outstanding._push(force=True)
            await self.rolodex_outstanding._push(force=True)
            
            logger.info('Building object dispatch.')
            await self.dispatch_tokens._push(force=True)
            await self.dispatch_tokens._push(force=True)
            await self.dispatch_startup._push(force=True)
            await self.dispatch_startup._push(force=True)
            await self.dispatch_private._push(force=True)
            await self.dispatch_private._push(force=True)
            await self.dispatch_incoming._push(force=True)
            await self.dispatch_incoming._push(force=True)
            await self.dispatch_orphan_acks._push(force=True)
            await self.dispatch_orphan_acks._push(force=True)
            await self.dispatch_orphan_naks._push(force=True)
            await self.dispatch_orphan_naks._push(force=True)
            
            self._user_id = root_node.ghid
        
            logger.info('Building root node...')
            # Generate secure-random-length, pseudorandom-content padding
            logger.info('    Generating noisy padding.')
            # Note that we don't actually need CSRNG for the padding, just the
            # padding length, since the whole thing is encrypted. We could just
            # as easily fill it with zeros, but by filling it with pseudorandom
            # noise, we can remove a recognizable pattern and therefore slighly
            # hinder brute force attacks against the password.
            # While we COULD use CSRNG despite all that, entropy is a limited
            # resource, and I'd rather conserve it as much as possible.
            padding_seed = int.from_bytes(os.urandom(2), byteorder='big')
            padding_min_size = 1024
            padding_clip_mask = 0b0001111111111111
            # Clip the seed to an upper range of 13 bits, of 8191, for a
            # maximum padding length of 8191 + 1024 = 9215 bytes
            padding_len = padding_min_size + (padding_seed & padding_clip_mask)
            padding_int = random.getrandbits(padding_len * 8)
            padding = padding_int.to_bytes(length=padding_len, byteorder='big')
            
            logger.info('    Serializing primary manifest.')
            root_node.state = (bytes(identity_container.ghid) +
                               bytes(identity_master) +
                               bytes(self.privateer_persistent.ghid) +
                               bytes(privateer_persistent_master) +
                               bytes(self.privateer_quarantine.ghid) +
                               bytes(privateer_quarantine_master) +
                               bytes(self.rolodex_pending.ghid) +
                               bytes(rolodex_pending_master) +
                               bytes(self.rolodex_outstanding.ghid) +
                               bytes(rolodex_outstanding_master) +
                               bytes(self.dispatch_tokens.ghid) +
                               bytes(dispatch_tokens_master) +
                               bytes(self.dispatch_startup.ghid) +
                               bytes(dispatch_startup_master) +
                               bytes(self.dispatch_private.ghid) +
                               bytes(dispatch_private_master) +
                               bytes(self.dispatch_incoming.ghid) +
                               bytes(dispatch_incoming_master) +
                               bytes(self.dispatch_orphan_acks.ghid) +
                               bytes(dispatch_orphan_acks_master) +
                               bytes(self.dispatch_orphan_naks.ghid) +
                               bytes(dispatch_orphan_naks_master) +
                               padding)
            
            # We'll use this upon future logins to verify password correctness
            logger.info('    Generating validator and comparator.')
            # Just append a SHA512 checksum to the whole thing
            password_validator = sha512(root_node.state).digest()
            root_node.state += password_validator
            
            logger.info('Saving root node.')
            await root_node._push()
        
        #######################################################################
        #######################################################################
        # ROOT NODE CREATION (PRIMARY BOOTSTRAP) COMPLETE!
        #######################################################################
        #######################################################################
        
        logger.info('Reticulating sharing subsystem.')
        await self._inject_gao(self.rolodex_pending)
        await self._inject_gao(self.rolodex_outstanding)
        
        logger.info('Reticulating object dispatch.')
        await self._inject_gao(self.dispatch_tokens)
        await self._inject_gao(self.dispatch_startup)
        await self._inject_gao(self.dispatch_private)
        await self._inject_gao(self.dispatch_incoming)
        await self._inject_gao(self.dispatch_orphan_acks)
        await self._inject_gao(self.dispatch_orphan_naks)
        
        logger.info('Account login successful.')
    
    @fixture_noop
    @public_api
    async def flush(self):
        ''' Push changes to any modified account components.
        '''
        tasks = {
            make_background_future(self.privateer_persistent.push()),
            make_background_future(self.privateer_quarantine.push()),
            make_background_future(self.rolodex_pending.push()),
            make_background_future(self.rolodex_outstanding.push()),
            make_background_future(self.dispatch_tokens.push()),
            make_background_future(self.dispatch_startup.push()),
            make_background_future(self.dispatch_private.push()),
            make_background_future(self.dispatch_incoming.push()),
            make_background_future(self.dispatch_orphan_acks.push()),
            make_background_future(self.dispatch_orphan_naks.push())
        }
        await asyncio.wait(fs=tasks, return_when=asyncio.ALL_COMPLETED) 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 SetMap:
    ''' Combines a mapping with a set. Threadsafe.
    '''
    
    def __init__(self):
        ''' Create a lookup!
        
        Currently does not support pre-population during __init__().
        '''
        self._mapping = {}
        self._lock = threading.RLock()
    
    def __getitem__(self, key):
        ''' Pass-through to the core lookup. Will return a frozenset.
        Raises keyerror if missing.
        '''
        with self._lock:
            return frozenset(self._mapping[key])
        
    def get_any(self, key):
        ''' Pass-through to the core lookup. Will return a frozenset.
        Will never raise a keyerror; if key not in self, returns empty
        frozenset.
        '''
        with self._lock:
            try:
                return frozenset(self._mapping[key])
            except KeyError:
                return frozenset()
                
    def pop_any(self, key):
        ''' Unlike other methods, pop_any returns the actual set,
        instead of a frozenset copy.
        '''
        with self._lock:
            try:
                return self._mapping.pop(key)
            except KeyError:
                return set()
        
    def __contains__(self, key):
        ''' Check to see if the key exists.
        '''
        with self._lock:
            return key in self._mapping
        
    def contains_within(self, key, value):
        ''' Check to see if the key exists, AND the value exists at key.
        '''
        with self._lock:
            try:
                return value in self._mapping[key]
            except KeyError:
                return False
        
    def add(self, key, value):
        ''' Adds the value to the set at key. Creates a new set there if
        none already exists.
        '''
        with self._lock:
            try:
                self._mapping[key].add(value)
            except KeyError:
                self._mapping[key] = {value}
                
    def update(self, key, value):
        ''' Updates the key with the value. Value must support being
        passed to set.update(), and the set constructor.
        '''
        with self._lock:
            try:
                self._mapping[key].update(value)
            except KeyError:
                self._mapping[key] = set(value)
                
    def update_all(self, other):
        ''' Updates all keys in other to self.
        '''
        with self._lock:
            for key in other:
                try:
                    self._mapping[key].update(other[key])
                except KeyError:
                    self._mapping[key] = set(other[key])
            
    def _remove_if_empty(self, key):
        ''' Removes a key entirely if it no longer has any values. Will
        suppress KeyError if the key is not found.
        '''
        try:
            if len(self._mapping[key]) == 0:
                del self._mapping[key]
        except KeyError:
            pass
        
    def remove(self, key, value):
        ''' Removes the value from the set at key. Will raise KeyError
        if either the key is missing, or the value is not contained at
        the key.
        '''
        with self._lock:
            try:
                self._mapping[key].remove(value)
            finally:
                self._remove_if_empty(key)
        
    def discard(self, key, value):
        ''' Same as remove, but will never raise KeyError.
        '''
        with self._lock:
            try:
                self._mapping[key].discard(value)
            except KeyError:
                pass
            finally:
                self._remove_if_empty(key)
        
    def clear(self, key):
        ''' Clears the specified key. Raises KeyError if key is not
        found.
        '''
        with self._lock:
            del self._mapping[key]
            
    def clear_any(self, key):
        ''' Clears the specified key, if it exists. If not, suppresses
        KeyError.
        '''
        with self._lock:
            try:
                del self._mapping[key]
            except KeyError:
                pass
        
    def clear_all(self):
        ''' Clears the entire mapping.
        '''
        with self._lock:
            self._mapping.clear()
            
    def __len__(self):
        ''' Returns the length of the mapping only.
        '''
        return len(self._mapping)
            
    def __iter__(self):
        ''' Send this through to the dict, bypassing the usual route,
        because otherwise we'll have a deadlock.
        '''
        with self._lock:
            for key in self._mapping:
                yield key
            
    def __eq__(self, other):
        ''' Expand comparison to search insides.
        '''
        with self._lock, other._lock:
            # First make sure both have same mapping keys.
            if set(self._mapping) != set(other._mapping):
                return False
            
            # Okay, now check each key has identical sets
            else:
                for this_key, this_set in self._mapping.items():
                    if other._mapping[this_key] != this_set:
                        return False
                        
        # If we successfully got through all of that, they are identical.
        return True
        
    def combine(self, other):
        ''' Returns a new SetMap with the union of both.
        '''
        new = type(self)()
        # Note that the iterator will take the lock, so we don't need to.
        for key in self:
            new.update(key, self._mapping[key])
        for key in other:
            new.update(key, other._mapping[key])
        return new

    def __getstate__(self):
        ''' Ignore self._lock to support pickling. Basically boilerplate
        copy from the pickle reference docs.
        '''
        # Copy the object's state from self.__dict__ which contains
        # all our instance attributes. Always use the dict.copy()
        # method to avoid modifying the original state.
        state = self.__dict__.copy()
        # Remove the unpicklable entries.
        del state['_lock']
        return state

    def __setstate__(self, state):
        ''' Ignore self._lock to support pickling. Basically boilerplate
        copy from the pickle reference docs.
        '''
        # Restore instance attributes (i.e., filename and lineno).
        self.__dict__.update(state)
        # Restore the lock.
        self._lock = threading.RLock()
        
    def __bool__(self):
        # Pass bool straight to mapping.
        return bool(self._mapping) class WeakSetMap(SetMap):
    ''' SetMap that uses WeakerSets internally.
    '''
    
    def add(self, key, value):
        ''' Adds the value to the set at key. Creates a new set there if
        none already exists.
        '''
        with self._lock:
            try:
                self._mapping[key].add(value)
            except KeyError:
                self._mapping[key] = _KeyedWeakSet(
                    data = (value,),
                    parent = self,
                    key = key,
                )
                
    def update(self, key, value):
        ''' Updates the key with the value. Value must support being
        passed to set.update(), and the set constructor.
        '''
        with self._lock:
            try:
                self._mapping[key].update(value)
            except KeyError:
                self._mapping[key] = _KeyedWeakSet(
                    data = (item for item in value),
                    parent = self,
                    key = key,
                )
                
    def update_all(self, other):
        ''' Updates all keys in other to self.
        '''
        with self._lock:
            for key in other:
                try:
                    self._mapping[key].update(other[key])
                except KeyError:
                    self._mapping[key] = _KeyedWeakSet(
                        data = other[key],
                        parent = self,
                        key = key
                    )
                
    def __getitem__(self, key):
        ''' Resolves all of our references.
        '''
        result = super().__getitem__(key)
        return frozenset(ref() for ref in result)
        
    def get_any(self, key):
        ''' Resolve references.
        '''
        result = super().get_any(key)
        return frozenset(ref() for ref in result) class ApiID(Ghid):
    ''' Subclass Ghid in a way that makes the API ID seem like a normal
    64-byte string.
    
    Remind me again why the hell I'm subclassing ghid for this when I'm
    ending up removing just about everything that makes it a ghid?
    '''
    
    def __init__(self, address, algo=None):
        ''' Wrap the normal Ghid creation with a forced algo.
        '''
        if len(address) != 64:
            raise ValueError('Improper API ID length.')
        
        elif algo is None:
            algo = 0
            
        elif algo != 0:
            raise ValueError('Improper API ID format.')
        
        super().__init__(algo, address)
        
    def __repr__(self):
        ''' Hide that this is a ghid.
        '''
        return type(self).__name__ + '(' + repr(self.address) + ')'
        
    @property
    def algo(self):
        return self._algo
        
    @algo.setter
    def algo(self, value):
        # Currently, ensure algo is b'\x00'
        if value == 0:
            self._algo = value
        else:
            raise ValueError('Invalid address algorithm.')
            
    @property
    def address(self):
        return self._address
            
    @address.setter
    def address(self, address):
        self._address = address
        
    @classmethod
    def pseudorandom(cls):
        return super().pseudorandom(0) class AppToken(bytes):
    ''' Extend bytes to be a fixed-length thing.
    '''
    # The size, in bytes, of an app token
    TOKEN_LEN = 16
    
    def __init__(self, *args, **kwargs):
        ''' For simplicity, enforce length AFTER performing init.
        '''
        if len(self) != self.TOKEN_LEN:
            raise ValueError('Improper token length.')
    
    @classmethod
    def null(cls):
        ''' Return the null token (used for passing "no token" in IPC,
        and not a valid token for app usage.)
        '''
        return cls(cls.TOKEN_LEN)
        
    @classmethod
    def pseudorandom(cls):
        ''' Return a PSEUDOrandom (**NOT** secure) app token, for use in
        testing.
        '''
        return cls([random.randint(0, 255) for __ in range(cls.TOKEN_LEN)])
        
    def __repr__(self):
        ''' Add in the class name to repr.
        '''
        return type(self).__name__ + '(' + super().__repr__() + ')'
        
    def __str__(self, *args, **kwargs):
        ''' Add in the class name to repr.
        '''
        return type(self).__name__ + '(' + \
            super().__str__(*args, **kwargs) + ')' class HGXLink(loopa.TaskCommander, metaclass=TriplicateAPI):
    ''' Amalgamate all of the necessary app functions into a single
    namespace. Also, and threadsafe and loopsafe bindings for stuff.
    '''
    
    @public_api
    def __init__(self, ipc_port=7772, autostart=True, *args, aengel=None,
                 threaded=True, ipc_fixture=None, **kwargs):
        ''' Args:
        ipc_port    is self-explanatory
        autostart   True -> immediately start the link
                    False -> app must explicitly start() the link
        debug       Sets debug mode for eg. asyncio
        aengel      Set up a watcher for main thread exits
        '''
        super().__init__(
            reusable_loop = False,
            threaded = threaded,
            *args,
            **kwargs
        )
        self._legroom = 7
        
        # Normally we'll need to define the protocol and the connection manager
        if ipc_fixture is None:
            ipc_protocol = IPCClientProtocol()
            ipc_manager = ConnectionManager(
                connection_cls = WSConnection,
                msg_handler = ipc_protocol,
                conn_init = self.conn_init,
                conn_close = self.conn_close
            )
            self.register_task(
                ipc_manager,
                host = 'localhost',
                port = ipc_port,
                tls = False
            )
            ipc_protocol.assemble(hgxlink=self)
            self._ipc_manager = ipc_manager
            self._ipc_protocol = ipc_protocol
        
        # Handle fixturing (for testing)
        else:
            self._ipc_manager = ipc_fixture
            self._ipc_protocol = None
            self.register_task(ipc_fixture)
        
        # All of the various object handlers
        # Lookup api_id: async awaitable share handler
        self._share_handlers = {}
        # Lookup api_id: object class
        self._share_typecast = {}
        
        # Lookup for ghid -> object
        self._objs_by_ghid = weakref.WeakValueDictionary()
        
        # Currently unused
        self._nonlocal_handlers = {}
        
        # These are intentionally None.
        self._token = None
        self._whoami = None
        # The startup object ghid gets stored here
        self._startup_obj = None
        
        # Create an executor for awaiting threadsafe callbacks and handlers
        self._executor = concurrent.futures.ThreadPoolExecutor()
        # And set up a flag so we know that we have whoami
        self._ctx = threading.Event()
        
        if autostart:
            self.start()
        
    @__init__.fixture
    def __init__(self, whoami=None, ipc_manager=None, *args, **kwargs):
        ''' Fixture all the things!
        '''
        super(HGXLink.__fixture__, self).__init__(
            threaded = True,
            debug = True,
            autostart = False,
            *args,
            **kwargs
        )
        
        self._ipc_manager = ipc_manager
        self._ipc_protocol = None
        
        self.state_lookup = {}
        self.share_lookup = {}
        self.api_lookup = {}
        self.deleted = set()
        self._whoami = whoami
        self.obj_lookup = {}
        
    @fixture_api
    def RESET(self):
        ''' Do this manually, because otherwise we create problems.
        '''
        self.state_lookup = {}
        self.share_lookup = {}
        self.api_lookup = {}
        self.deleted = set()
        self.obj_lookup = {}
            
    def start(self, *args, **kwargs):
        ''' Await a connection if we're running threaded-ly.
        '''
        # Explicit is needed because of the fixture
        super().start(*args, **kwargs)
        
        if self._ipc_protocol is not None and self.threaded:
            self._ipc_manager.await_connection_threadsafe()
            self._ctx.wait()
            
    async def conn_init(self, ipc_manager, connection):
        ''' Figure out whoami, set up tokens, set self._ctx, etc.
        '''
        self.whoami = await self._ipc_protocol.get_whoami(connection)
        
        if self._token is not None:
            await self._ipc_protocol.set_token(self._token)
            
        self._ctx.set()
        
    async def conn_close(self, ipc_manager, connection):
        ''' Clear whoami. Keep token, though, in case we restore the
        connection successfully. The application isn't changing, but our
        agent might.
        
        Also clear self._ctx, not that it makes much difference.
        '''
        self._whoami = None
        self._ctx.clear()
            
    @fixture_api
    def prep_obj(self, obj):
        ''' Stages an object for a get() call.
        '''
        self.obj_lookup[obj._hgx_ghid] = obj
        
    @property
    def whoami(self):
        ''' Read-only access to self._whoami with a raising wrapper if
        it is undefined.
        '''
        if self._whoami is None:
            raise HGXLinkError(
                'Whoami has not been defined. Most likely, no IPC client is ' +
                'currently available.'
            )
        else:
            return self._whoami
            
    @whoami.setter
    def whoami(self, ghid):
        ''' Set whoami, if (and only if) it has yet to be set.
        '''
        if self._whoami is None:
            self._whoami = ghid
        else:
            raise HGXLinkError(
                'Whoami has already been defined. It must be cleared before ' +
                'being re-set.'
            )
    
    @property
    def token(self):
        ''' Read-only access to the current app token.
        '''
        if self._token is None:
            raise HGXLinkError('No token available.')
        else:
            return self._token
            
    @token.setter
    def token(self, value):
        ''' Set the app token, if (and only if) it has yet to be set.
        '''
        if self._token is None:
            if isinstance(value, AppToken):
                self._token = value
            else:
                raise TypeError('Tokens must be of type ' +
                                'hypergolix.utils.AppToken')
        else:
            raise HGXLinkError(
                'Token already set. It must be cleared before being re-set.'
            )
        
    @triplicated
    async def register_token(self, token=None):
        ''' Registers the application as using a particular token, OR
        gets a new token. Returns the ghid for the startup object (if
        one has already been defined), or None. Tokens are be available
        in self.token.
        '''
        # The result of this will be the actual token.
        self.token = await self._ipc_manager.set_token(token)
        
        if token is not None:
            self._startup_obj = await self._ipc_manager.get_startup_obj()
        
        # Let applications manually request the startup object, so that they
        # can deal with casting it appropriately.
        return self._startup_obj
        
    @triplicated
    async def register_startup(self, obj):
        ''' Registers the object as the startup object.
        '''
        await self._ipc_manager.register_startup_obj(obj._hgx_ghid)
        
    @triplicated
    async def deregister_startup(self):
        ''' Inverse of the above.
        '''
        await self._ipc_manager.deregister_startup_obj()
        
    @public_api
    @triplicated
    async def get(self, cls, ghid, obj_def=None):
        ''' Pass to connection manager. Also, turn the object into the
        specified class. If obj is not None, re-cast it as such.
        '''
        # Obj_def is passed in when recasting.
        if obj_def is not None:
            (address,
             author,
             state,
             is_link,
             api_id,
             private,
             dynamic,
             _legroom) = obj_def
            
        # This is bypassed during recasting, but otherwise pulls an existing
        # object from our local cache
        elif ghid in self._objs_by_ghid:
            obj = self._objs_by_ghid[ghid]
            if type(obj) != cls:
                raise HGXLinkError(
                    'Cannot attempt to get a new copy of an object using a ' +
                    'new class. Use obj.recast instead.'
                )
            else:
                return obj
        
        # We don't have the object in our cache, nor are we recasting, so we
        # must get it directly from the hypergolix service
        else:
            (address,
             author,
             state,
             is_link,
             api_id,
             private,
             dynamic,
             _legroom) = await self._ipc_manager.get_ghid(ghid)
            
        if is_link:
            # First discard the object, since we can't support it.
            await self._ipc_manager.discard_ghid(ghid)
            
            # Now raise.
            raise NotImplementedError(
                'Hypergolix does not yet support nested links to other '
                'dynamic objects.'
            )
            # link = Ghid.from_bytes(state)
            # state = await self._get(link)
            
        if _legroom is None:
            _legroom = self._legroom
        
        state = await cls.hgx_unpack(state)
        obj = cls(
            hgxlink = self,
            ipc_manager = self._ipc_manager,
            state = state,
            api_id = api_id,
            dynamic = dynamic,
            private = private,
            ghid = address,
            binder = author,
            _legroom = _legroom,
        )
            
        # Don't forget to add it to local lookup so we can apply updates.
        self._objs_by_ghid[obj._hgx_ghid] = obj
        
        return obj
        
    @get.fixture
    async def get(self, cls, ghid, obj_def=None):
        ''' Fixture get behavior.
        '''
        if obj_def is None:
            old_obj = self.obj_lookup[ghid]
            # ehhhh just ignore the class on this one
            return old_obj
            
        else:
            (address,
             author,
             state,
             is_link,
             api_id,
             private,
             dynamic,
             _legroom) = obj_def
            state = await cls.hgx_unpack(state)
            new_obj = cls(
                hgxlink = self,
                ipc_manager = self._ipc_manager,
                state = state,
                api_id = api_id,
                dynamic = dynamic,
                private = private,
                ghid = address,
                binder = author,
                _legroom = _legroom,
            )
            return new_obj
        
    @triplicated
    async def new(self, cls, state, api_id=None, dynamic=True, private=False,
                  _legroom=None, *args, **kwargs):
        ''' Create a new object w/ class cls.
        '''
        if _legroom is None:
            _legroom = self._legroom
            
        if api_id is None:
            api_id = cls._hgx_DEFAULT_API
            
        obj = cls(
            hgxlink = self,
            ipc_manager = self._ipc_manager,
            _legroom = _legroom,
            state = state,
            api_id = api_id,
            dynamic = dynamic,
            private = private,
            binder = self.whoami,
            *args, **kwargs
        )
        
        packed_state = await obj.hgx_pack(state)
        
        address = await self._ipc_manager.new_ghid(
            packed_state,
            api_id,
            dynamic,
            private,
            _legroom
        )
        
        obj._hgx_ghid = address
        # Don't forget to add it to local lookup so we can apply updates.
        self._objs_by_ghid[obj._hgx_ghid] = obj
        return obj
    
    @triplicated
    async def register_nonlocal_handler(self, api_id, handler):
        ''' Call this to register a handler for any private objects
        created by the same hypergolix identity and the same hypergolix
        application, but at a separate, concurrent session.
        
        This HANDLER will be called from within the IPC embed's internal
        event loop.
        
        This METHOD must be called from within the IPC embed's internal
        event loop.
        '''
        raise NotImplementedError()
        api_id = self._normalize_api_id(api_id)
        
        # self._nonlocal_handlers = {}
    
    @triplicated
    async def register_share_handler(self, api_id, handler):
        ''' Call this to register a handler for an object shared by a
        different hypergolix identity, or the same hypergolix identity
        but a different application. Any api_id can have at most one
        share handler, across ALL forms of callback (internal,
        threadsafe, loopsafe).
        
        The handler will be called as:
            await handler(ghid, origin, api_id)
        
        This HANDLER will be called from within the IPC embed's internal
        event loop.
        
        This METHOD must be called from within the IPC embed's internal
        event loop.
        '''
        if not isinstance(api_id, ApiID):
            raise TypeError('api_id must be ApiID.')
        # Add this check to help shield against accidentally-incomplete
        # loopsafe decorators, and attempts to directly set functions.
        elif not inspect.iscoroutinefunction(handler):
            raise TypeError('Handler must be defined with "async def".')
        
        await self._ipc_manager.register_api(api_id)
        
        # Any handlers passed to us this way can already be called natively
        # from withinour own event loop, so they just need to be wrapped such
        # that they never raise.
        async def wrap_handler(*args, handler=handler, **kwargs):
            try:
                await handler(*args, **kwargs)
                
            except asyncio.CancelledError:
                logger.debug('Share handling cancelled.')
                raise
                
            except Exception:
                logger.error(
                    'Error while running share handler. Traceback:\n' +
                    ''.join(traceback.format_exc())
                )
        
        # Hey, look at this! Because we're running a single-threaded event loop
        # and not ceding flow control to the loop, we don't need to worry about
        # synchro primitives here!
        self._share_handlers[api_id] = wrap_handler
        
    @triplicated
    async def deregister_share_handler(self, api_id):
        ''' Removes a share handler.
        '''
        await self._ipc_manager.deregister_api(api_id)
        try:
            del self._share_handlers[api_id]
        except KeyError:
            logger.warning('No existing share handler for ' + str(api_id))
    
    def wrap_threadsafe(self, callback):
        ''' Call this to register a handler for an object shared by a
        different hypergolix identity, or the same hypergolix identity
        but a different application. Any api_id can have at most one
        share handler, across ALL forms of callback (internal,
        threadsafe, loopsafe).
        
        typecast determines what kind of ObjProxy class the object will
        be cast into before being passed to the handler.
        
        This HANDLER will be called from within a single-use, dedicated
        thread.
        
        This METHOD must be called from a different thread than the IPC
        embed's internal event loop.
        '''
        # For simplicity, wrap the handler, so that any shares can be called
        # normally from our own event loop.
        @functools.wraps(callback)
        async def wrapped_handler(*args, self_weakref=weakref.ref(self),
                                  func=callback):
            ''' Wrap the handler in run_in_executor.
            '''
            self = self_weakref()
            if self is not None:
                await self._loop.run_in_executor(
                    self._executor,
                    func,
                    *args
                )
        
        return wrapped_handler
    
    def wrap_loopsafe(self, callback=None, *, target_loop):
        ''' Call this to register a handler for an object shared by a
        different hypergolix identity, or the same hypergolix identity
        but a different application. Any api_id can have at most one
        share handler, across ALL forms of callback (internal,
        threadsafe, loopsafe).
        
        typecast determines what kind of ObjProxy class the object will
        be cast into before being passed to the handler.
        
        This HANDLER will be called within the specified event loop,
        also implying the specified event loop context (ie thread).
        
        This METHOD must be called from a different event loop than the
        IPC embed's internal event loop. It is internally loopsafe, and
        need not be wrapped by run_coroutine_loopsafe.
        '''
        # This can be used as a decorator, or directly as a function. If used
        # as a decorator, it will be called with a single kwarg -- the
        # target_loop -- which must then be used to construct a decorator that
        # can then be invoked to make the actual wrapped function.
        
        # No callback, so this is a decorator generation call.
        if callback is None:
            def decorator_closure(func, self_weakref=weakref.ref(self),
                                  target_loop=target_loop):
                ''' Returns a decorator that can be used to wrap things
                in loop safety.
                '''
                self = self_weakref()
                if self is None:
                    raise RuntimeError(
                        'HGXLink garbage collected before properly wrapping ' +
                        'a loopsafe callback.'
                    )
                else:
                    # Note that target_loop must be passed as an explicit kwarg
                    return self.wrap_loopsafe(func, target_loop=target_loop)
                    
            return decorator_closure
        
        # Target loop defined, so this is actually generating a wrapped handler
        else:
            # For simplicity, wrap the handler, so that any shares can be
            # called normally from our own event loop.
            @functools.wraps(callback)
            async def wrapped_handler(*args, target_loop=target_loop,
                                      coro=callback):
                ''' Wrap the handler in await_coroutine_loopsafe.
                '''
                await await_coroutine_loopsafe(
                    coro = coro(*args),
                    loop = target_loop
                )
            
            return wrapped_handler
        
    @public_api
    async def _pull_state(self, ghid, state):
        ''' Applies an incoming state update.
        '''
        if isinstance(state, Ghid):
            raise NotImplementedError('Linked objects not yet supported.')
        
        try:
            obj = self._objs_by_ghid[ghid]
            
        except KeyError:
            # Just discard the object, since we don't actually have a copy of
            # it locally.
            logger.warning(
                'Received an object update, but the object was no longer '
                'contained in memory. Discarding its subscription: ' +
                str(ghid) + '.'
            )
            await self._ipc_manager.discard_ghid(ghid)
            
        else:
            logger.debug(
                'Received update for ' + str(ghid) + '; forcing pull.'
            )
            await obj._hgx_force_pull(state)
            
    @_pull_state.fixture
    async def _pull_state(self, ghid, state):
        ''' Fixture for applying incoming state update.
        '''
        self.state_lookup[ghid] = state
            
    @public_api
    async def handle_share(self, ghid, origin, api_id):
        ''' Handles an incoming shared object.
        '''
        # This is async, which is single-threaded, so there's no race condition
        try:
            handler = self._share_handlers[api_id]
            
        except KeyError:
            logger.warning(
                'Received a share for an API_ID that was lacking a handler or '
                'typecast. Deregistering the API_ID.'
            )
            await self._ipc_manager.deregister_api(api_id)
            
        else:
            # Run the share handler concurrently, so that we can release the
            # req/res session
            share_task = asyncio.ensure_future(handler(ghid, origin, api_id))
            share_task.add_done_callback(_reap_wrapped_task)
            
    @handle_share.fixture
    async def handle_share(self, ghid, origin, api_id):
        ''' Fixture handling an incoming share object.
        '''
        self.share_lookup[ghid] = origin
        self.api_lookup[ghid] = api_id
            
    @public_api
    async def handle_delete(self, ghid):
        ''' Applies an incoming delete.
        '''
        try:
            obj = self._objs_by_ghid[ghid]
        
        except KeyError:
            logger.debug(str(ghid) + ' not known to IPCEmbed.')
        
        else:
            await obj._hgx_force_delete()
            del self._objs_by_ghid[ghid]
            
    @handle_delete.fixture
    async def handle_delete(self, ghid):
        ''' Fixtures handling an incoming delete.
        '''
        self.deleted.add(ghid) class HypergolixException(Exception):
    ''' This is suclassed for all exceptions and warnings, so that code
    using hypergolix as an import can successfully catch all hypergolix
    exceptions with a single except.
    '''
    pass class IPCError(HypergolixException, RuntimeError):
    ''' Raised when something goes wrong with IPC or embed (bad
    commands, etc).
    '''
    pass class IPCServerProtocol(_IPCSerializer, metaclass=RequestResponseAPI,
                        error_codes=ERROR_CODES, default_version=b'\x00\x00'):
    ''' Defines the protocol for IPC, with handlers specific to servers.
    '''
    _dispatch = weak_property('__dispatch')
    _oracle = weak_property('__oracle')
    _golcore = weak_property('__golcore')
    _rolodex = weak_property('__rolodex')
    _salmonator = weak_property('__salmonator')
        
    @fixture_api
    def __init__(self, whoami, *args, **kwargs):
        super(IPCServerProtocol.__fixture__, self).__init__(*args, **kwargs)
        self._whoami = whoami
        self.shares = collections.deque()
        self.updates = collections.deque()
        self.deletes = collections.deque()
        self.share_successes = collections.deque()
        self.share_failures = collections.deque()
        
    def assemble(self, golcore, oracle, dispatch, rolodex, salmonator):
        # Chicken, egg, etc.
        self._golcore = golcore
        self._oracle = oracle
        self._dispatch = dispatch
        self._rolodex = rolodex
        self._salmonator = salmonator
        
    @request(b'+T')
    async def set_token(self, connection, token=None):
        ''' Register an existing token or get a new token, or notify an
        app of its existing token.
        '''
        # On the server side, this will only be implemented once actual
        # application launching is available.
        raise NotImplementedError()
        
    @set_token.request_handler
    async def set_token(self, connection, body):
        ''' Handles token-setting requests.
        '''
        token = AppToken(body)
        
        # Getting a new token.
        if token == AppToken.null():
            token = await self._dispatch.start_application(connection)
            logger.info(''.join((
                'CONN ', str(connection), ' generating new token: ', str(token)
            )))
        
        # Setting an existing token, but the connection already exists.
        elif self._dispatch.token_lookup(connection):
            raise IPCError(
                'Attempt to reregister a new concurrent token for an ' +
                'existing connection. Each app may use only one token.'
            )
            
        # Setting an existing token, but the token already exists.
        elif self._dispatch.connection_lookup(token) is not None:
            raise IPCError(
                'Attempt to reregister a new concurrent connection for ' +
                'an existing token. Each app may only use one connection.'
            )
        
        # Setting an existing token, with valid state.
        else:
            logger.info(''.join((
                'CONN ', str(connection), ' registering existing token: ',
                str(token)
            )))
            await self._dispatch.start_application(connection, token)
        
        return token
        
    @request(b'+A')
    async def register_api(self, connection):
        ''' Registers the application as supporting an API. Client only.
        '''
        raise NotImplementedError()
        
    @register_api.request_handler
    async def register_api(self, connection, body):
        ''' Handles API registration requests. Server only.
        '''
        api_id = ApiID.from_bytes(body)
        await self._dispatch.add_api(connection, api_id)
        
        return b'\x01'
        
    @request(b'-A')
    async def deregister_api(self, connection):
        ''' Removes any existing registration for the app supporting an
        API. Client only.
        '''
        raise NotImplementedError()
        
    @deregister_api.request_handler
    async def deregister_api(self, connection, body):
        ''' Handles API deregistration requests. Server only.
        '''
        api_id = ApiID.from_bytes(body)
        await self._dispatch.remove_api(connection, api_id)
        
        return b'\x01'
        
    @public_api
    @request(b'?I')
    async def whoami(self, connection):
        ''' Get the current hypergolix fingerprint, or notify an app of
        the current hypergolix fingerprint.
        '''
        # On the server side, this will only be implemented once actual
        # application launching is available.
        raise NotImplementedError()
        
    @whoami.request_handler
    async def whoami(self, connection, body):
        ''' Handles whoami requests.
        '''
        ghid = self._golcore.whoami
        return bytes(ghid)
        
    @request(b'>$')
    async def get_startup_obj(self, connection):
        ''' Request a startup object, or notify an app of its declared
        startup object.
        '''
        token = self._dispatch.token_lookup(connection)
        ghid = await self._dispatch.get_startup_obj(token)
        
        if ghid is not None:
            return bytes(ghid)
        else:
            return b''
        
    @get_startup_obj.request_handler
    async def get_startup_obj(self, connection, body):
        ''' Handles requests for startup objects.
        '''
        token = self._dispatch.token_lookup(connection)
        ghid = await self._dispatch.get_startup_obj(token)
        
        if ghid is not None:
            return bytes(ghid)
        else:
            return b''
        
    @request(b'+$')
    async def register_startup_obj(self, connection, ghid):
        ''' Register a startup object. Client only.
        '''
        raise NotImplementedError()
        
    @register_startup_obj.request_handler
    async def register_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._dispatch.register_startup(connection, ghid)
        return b'\x01'
        
    @request(b'-$')
    async def deregister_startup_obj(self, connection):
        ''' Register a startup object. Client only.
        '''
        raise NotImplementedError()
        
    @deregister_startup_obj.request_handler
    async def deregister_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        await self._dispatch.deregister_startup(connection)
        return b'\x01'
        
    @request(b'>O')
    async def get_obj(self, connection):
        ''' Get an object with the specified address. Client only.
        '''
        raise NotImplementedError()
        
    @get_obj.request_handler
    async def get_obj(self, connection, body):
        ''' Handles requests for an object. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        
        await self._dispatch.track_object(connection, obj.ghid)
            
        if isinstance(obj.state, Ghid):
            is_link = True
            state = bytes(obj.state)
        else:
            is_link = False
            state = obj.state
            
        # For now, anyways.
        # Note: need to add some kind of handling for legroom.
        _legroom = None
        
        # Not a big fan of how this works, seems inelegant to me
        private = bool(self._dispatch.private_parent_lookup(obj.ghid))
        
        return self._pack_object_def(
            obj.ghid,
            obj.author,
            state,
            is_link,
            obj.api_id,
            private,
            obj.dynamic,
            _legroom
        )
        
    @request(b'+O')
    async def new_obj(self, connection):
        ''' Create a new object, or notify an app of a new object
        created by a concurrent instance of the app on a different
        hypergolix session.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @new_obj.request_handler
    async def new_obj(self, connection, body):
        ''' Handles requests for new objects.
        '''
        (address,    # Unused and set to None.
         author,     # Unused and set to None.
         state,
         is_link,
         api_id,
         private,
         dynamic,
         legroom) = self._unpack_object_def(body)
        
        if is_link:
            raise NotImplementedError('Linked objects are not yet supported.')
            state = Ghid.from_bytes(state)
        
        obj = await self._oracle.new_object(
            gaoclass = _Dispatchable,
            dispatch = self._dispatch,
            ipc_protocol = self,
            state = state,
            dynamic = dynamic,
            legroom = legroom,
            api_id = api_id,
            account = self._dispatch._account
        )
            
        # Add the endpoint as a listener.
        await self._dispatch.register_object(connection, obj.ghid, private)
        await self._dispatch.track_object(connection, obj.ghid)
        
        return bytes(obj.ghid)
    
    @public_api
    @request(b'!O')
    async def update_obj(self, connection, ghid):
        ''' Update an object or notify an app of an incoming update.
        '''
        try:
            obj = await self._oracle.get_object(
                gaoclass = _Dispatchable,
                ghid = ghid,
                api_id = None,  # Let _pull() apply this.
                state = None,   # Let _pull() apply this.
                dispatch = self._dispatch,
                ipc_protocol = self,
                account = self._dispatch._account
            )
            
        # No CancelledError catch necessary because we're re-raising any exc
            
        except Exception:
            # At some point we'll need some kind of proper handling for this.
            logger.error(
                'Failed to retrieve object at ' + str(ghid) + '\n' +
                ''.join(traceback.format_exc())
            )
            raise
            
        else:
            return self._pack_object_def(
                obj.ghid,
                obj.author,
                obj.state,
                False,      # is_link (currently unsupported)
                obj.api_id,
                None,       # private
                obj.dynamic,
                None        # legroom
            )
    
    @update_obj.fixture
    async def update_obj(self, connection, ghid):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.updates.append((connection, ghid))
    
    @update_obj.request_handler
    async def update_obj(self, connection, body):
        ''' Handles update object requests.
        '''
        logger.debug('Handling update request from ' + str(connection))
        (ghid,
         author,    # Unused and set to None.
         state,
         is_link,
         api_id,    # Unused and set to None.
         private,   # TODO: use this.
         dynamic,   # Unused and set to None.
         legroom   # TODO: use this.
         ) = self._unpack_object_def(body)
        
        if is_link:
            raise NotImplementedError('Linked objects are not yet supported.')
            state = Ghid.from_bytes(state)
            
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        obj.state = state
        
        # Converting a private object to a public one
        if self._dispatch.private_parent_lookup(ghid):
            if not private:
                await self._dispatch.make_public(ghid)
        
        else:
            if private:
                raise CatOuttaBagError(
                    'Once made public, objects cannot be made private. They ' +
                    'have already been distributed to other applications.'
                )
            
        await obj.push()
        
        # Schedule an update in the background.
        make_background_future(
            self._dispatch.distribute_update(
                obj.ghid,
                skip_conn = connection
            )
        )
        
        return b'\x01'
        
    @request(b'~O')
    async def sync_obj(self, connection):
        ''' Manually force Hypergolix to check an object for updates.
        Client only.
        '''
        raise NotImplementedError()
        
    @sync_obj.request_handler
    async def sync_obj(self, connection, body):
        ''' Handles manual syncing requests. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._salmonator.attempt_pull(ghid)
        return b'\x01'
    
    @public_api
    @request(b'@O')
    async def share_obj(self, connection, ghid, origin, api_id):
        ''' Request an object share or notify an app of an incoming
        share.
        '''
        if origin is None:
            origin = self._golcore.whoami
        
        return bytes(ghid) + bytes(origin) + bytes(api_id)
        
    @share_obj.fixture
    async def share_obj(self, connection, ghid, origin, api_id):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.shares.append(
            _ShareLog(connection, ghid, origin, api_id)
        )
        
    @share_obj.request_handler
    async def share_obj(self, connection, body):
        ''' Handles object share requests.
        '''
        ghid = Ghid.from_bytes(body[0:65])
        recipient = Ghid.from_bytes(body[65:130])
        
        # Instead of forbidding unregistered apps from sharing objects,
        # go for it, but document that you will never be notified of a
        # share success or failure without an app token.
        requesting_token = self._dispatch.token_lookup(connection)
        if requesting_token is None:
            logger.info(
                'CONN ' + str(connection) + ' is sharing ' + str(ghid) +
                ' with ' + str(recipient) + ' without defining an app ' +
                'token, and therefore cannot be notified of share success ' +
                'or failure.'
            )
            
        await self._rolodex.share_object(ghid, recipient, requesting_token)
        return b'\x01'
    
    @public_api
    @request(b'^S')
    async def notify_share_success(self, connection, ghid, recipient):
        ''' Notify app of successful share. Server only.
        '''
        return bytes(ghid) + bytes(recipient)
    
    @notify_share_success.fixture
    async def notify_share_success(self, connection, ghid, recipient):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.share_successes.append((connection, ghid, recipient))
    
    @notify_share_success.request_handler
    async def notify_share_success(self, connection, body):
        ''' Handles app notifications for successful shares. Client
        only.
        '''
        raise NotImplementedError()
    
    @public_api
    @request(b'^F')
    async def notify_share_failure(self, connection, ghid, recipient):
        ''' Notify app of unsuccessful share. Server only.
        '''
        return bytes(ghid) + bytes(recipient)
        
    @notify_share_failure.fixture
    async def notify_share_failure(self, connection, ghid, recipient):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.share_failures.append((connection, ghid, recipient))
    
    @notify_share_failure.request_handler
    async def notify_share_failure(self, connection, body):
        ''' Handles app notifications for unsuccessful shares. Client
        only.
        '''
        raise NotImplementedError()
        
    @request(b'*O')
    async def freeze_obj(self, connection):
        ''' Creates a new static copy of the object, or notifies an app
        of a frozen copy of an existing object created by a concurrent
        instance of the app.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @freeze_obj.request_handler
    async def freeze_obj(self, connection, body):
        ''' Handles object freezing requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        frozen_address = await obj.freeze()
        
        return bytes(frozen_address)
        
    @request(b'#O')
    async def hold_obj(self, connection):
        ''' Creates a new static binding for the object, or notifies an
        app of a static binding created by a concurrent instance of the
        app.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @hold_obj.request_handler
    async def hold_obj(self, connection, body):
        ''' Handles object holding requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        await obj.hold()
        return b'\x01'
        
    @request(b'-O')
    async def discard_obj(self, connection):
        ''' Stop listening to object updates. Client only.
        '''
        raise NotImplementedError()
        
    @discard_obj.request_handler
    async def discard_obj(self, connection, body):
        ''' Handles object discarding requests. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._dispatch.untrack_object(connection, ghid)
        return b'\x01'
    
    @public_api
    @request(b'XO')
    async def delete_obj(self, connection, ghid):
        ''' Request an object deletion or notify an app of an incoming
        deletion.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('ghid must be type Ghid or similar.')
        
        return bytes(ghid)
    
    @delete_obj.fixture
    async def delete_obj(self, connection, ghid):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.deletes.append((connection, ghid))
    
    @delete_obj.request_handler
    async def delete_obj(self, connection, body):
        ''' Handles object deletion requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        await self._dispatch.untrack_object(connection, ghid)
        # TODO: shift to a dispatch.delete_object method that can ignore this
        # connection when the intevitable "deleted that object!" warning comes
        # back
        await obj.delete()
        return b'\x01' class IPCClientProtocol(_IPCSerializer, metaclass=RequestResponseAPI,
                        error_codes=ERROR_CODES, default_version=b'\x00\x00'):
    ''' Defines the protocol for IPC, with handlers specific to clients.
    '''
    _hgxlink = weak_property('__hgxlink')
        
    @fixture_api
    def __init__(self, whoami, *args, **kwargs):
        ''' Create the fixture internals.
        '''
        # This is necessary because of the fixture_bases bit.
        super(IPCClientProtocol.__fixture__, self).__init__(*args, **kwargs)
        self.whoami = whoami
        self.apis = set()
        self.token = None
        self.startup = None
        self.pending_obj = None
        self.pending_ghid = None
        self.discarded = set()
        self.updates = []
        self.syncs = []
        self.shares = SetMap()
        self.frozen = set()
        self.held = set()
        self.deleted = set()
        
    @fixture_api
    def RESET(self):
        ''' Nothing beyond just re-running __init__, reusing whoami.
        '''
        self.apis = set()
        self.token = None
        self.startup = None
        self.pending_obj = None
        self.pending_ghid = None
        self.discarded = set()
        self.updates = []
        self.syncs = []
        self.shares = SetMap()
        self.frozen = set()
        self.held = set()
        self.deleted = set()
        
    @fixture_api
    def prep_obj(self, obj):
        ''' Define the next object to be returned in any obj-based
        operations.
        '''
        self.pending_obj = (
            obj._hgx_ghid,
            obj._hgx_binder,
            obj._hgx_state,
            obj._hgx_linked,
            obj._hgx_api_id,
            obj._hgx_private,
            obj._hgx_dynamic,
            obj._hgx_legroom
        )
        
    def assemble(self, hgxlink):
        # Chicken, egg, etc.
        self._hgxlink = hgxlink
        
    @public_api
    @request(b'+T')
    async def set_token(self, connection, token):
        ''' Register an existing token or get a new token, or notify an
        app of its existing token.
        '''
        if token is None:
            return AppToken.null()
        else:
            return token
        
    @set_token.request_handler
    async def set_token(self, connection, body):
        ''' Handles token-setting requests.
        '''
        self._hgxlink.token = AppToken(body)
        
    @set_token.fixture
    async def set_token(self, token):
        ''' Fixture for setting a token (or getting a new one).
        '''
        if token is None:
            # ONLY use pseudorandom for the fixture
            token = AppToken.pseudorandom()
            
        self.token = token
        return token
        
    @set_token.response_handler
    async def set_token(self, connection, response, exc):
        ''' Converts the response into an AppToken.
        '''
        if exc is not None:
            raise exc
        else:
            return AppToken(response)
    
    @public_api
    @request(b'+A')
    async def register_api(self, connection, api_id):
        ''' Registers the application as supporting an API. Client only.
        '''
        if not isinstance(api_id, ApiID):
            raise TypeError('api_id must be ApiID.')
            
        return bytes(api_id)
        
    @register_api.request_handler
    async def register_api(self, connection, body):
        ''' Handles API registration requests. Server only.
        '''
        raise NotImplementedError()
        
    @register_api.response_handler
    async def register_api(self, connection, response, exc):
        ''' Handles responses to API registration requests. Client only.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while registering API.')
            
    @register_api.fixture
    async def register_api(self, api_id):
        ''' Fixture for registering an api.
        '''
        self.apis.add(api_id)
        
    @public_api
    @request(b'-A')
    async def deregister_api(self, connection, api_id):
        ''' Removes any existing registration for the app supporting an
        API. Client only.
        '''
        if not isinstance(api_id, ApiID):
            raise TypeError('api_id must be ApiID.')
            
        return bytes(api_id)
        
    @deregister_api.request_handler
    async def deregister_api(self, connection, body):
        ''' Handles API deregistration requests. Server only.
        '''
        raise NotImplementedError()
        
    @deregister_api.response_handler
    async def deregister_api(self, connection, response, exc):
        ''' Handles responses to API deregistration requests. Client
        only.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while deregistering API.')
            
    @deregister_api.fixture
    async def deregister_api(self, api_id):
        ''' Fixture for api removal.
        '''
        self.apis.discard(api_id)
        
    @public_api
    @request(b'?I')
    async def get_whoami(self, connection):
        ''' Get the current hypergolix fingerprint, or notify an app of
        the current hypergolix fingerprint.
        '''
        return b''
        
    @get_whoami.request_handler
    async def get_whoami(self, connection, body):
        ''' Handles whoami requests.
        '''
        ghid = Ghid.from_bytes(body)
        self._hgxlink.whoami = ghid
        return b''
        
    @get_whoami.response_handler
    async def get_whoami(self, connection, response, exc):
        ''' Handles responses to whoami requests.
        '''
        if exc is not None:
            raise exc
        else:
            ghid = Ghid.from_bytes(response)
            return ghid
            
    @get_whoami.fixture
    async def get_whoami(self, connection):
        ''' Fixture for get_whoami.
        '''
        return self.whoami
        
    @public_api
    @request(b'>$')
    async def get_startup_obj(self, connection):
        ''' Request a startup object, or notify an app of its declared
        startup object.
        '''
        return b''
        
    @get_startup_obj.request_handler
    async def get_startup_obj(self, connection, body):
        ''' Handles requests for startup objects.
        '''
        ghid = Ghid.from_bytes(body)
        self._hgxlink._startup_obj = ghid
        return b'\x01'
        
    @get_startup_obj.response_handler
    async def get_startup_obj(self, connection, response, exc):
        ''' Handle the response to retrieving startup obj ghids.
        '''
        if exc is not None:
            raise exc
        elif response == b'':
            return None
        else:
            ghid = Ghid.from_bytes(response)
            return ghid
            
    @get_startup_obj.fixture
    async def get_startup_obj(self):
        ''' Yep, go ahead and fixture that, too.
        '''
        return self.startup
        
    @public_api
    @request(b'+$')
    async def register_startup_obj(self, connection, ghid):
        ''' Register a startup object. Client only.
        '''
        return bytes(ghid)
        
    @register_startup_obj.request_handler
    async def register_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        raise NotImplementedError()
        
    @register_startup_obj.fixture
    async def register_startup_obj(self, ghid):
        ''' Fixture startup obj registration and stuff.
        '''
        self.startup = ghid
        
    @public_api
    @request(b'-$')
    async def deregister_startup_obj(self, connection):
        ''' Register a startup object. Client only.
        '''
        return b''
        
    @deregister_startup_obj.request_handler
    async def deregister_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        raise NotImplementedError()
        
    @deregister_startup_obj.fixture
    async def deregister_startup_obj(self):
        ''' Still more fixtures.
        '''
        self.startup = None
        
    @public_api
    @request(b'>O')
    async def get_ghid(self, connection, ghid):
        ''' Get an object with the specified address. Client only.
        '''
        return bytes(ghid)
        
    @get_ghid.request_handler
    async def get_ghid(self, connection, body):
        ''' Handles requests for an object. Server only.
        '''
        raise NotImplementedError()
        
    @get_ghid.response_handler
    async def get_ghid(self, connection, response, exc):
        ''' Handles responses to get object requests. Client only.
        '''
        if exc is not None:
            raise exc
            
        return self._unpack_object_def(response)
        
    @get_ghid.fixture
    async def get_ghid(self, ghid):
        ''' Interact with pending_obj.
        '''
        return self.pending_obj
        
    @public_api
    @request(b'+O')
    async def new_ghid(self, connection, state, api_id, dynamic, private,
                       _legroom):
        ''' Create a new object, or notify an app of a new object
        created by a concurrent instance of the app on a different
        hypergolix session.
        '''
        # If state is Ghid, it's a link.
        if isinstance(state, Ghid):
            is_link = True
        else:
            is_link = False
        
        return self._pack_object_def(
            None,               # address
            None,               # author
            state,              # state
            is_link,            # is_link
            bytes(api_id),      # api_id
            private,            # private
            dynamic,            # dynamic
            _legroom            # legroom
        )
        
    @new_ghid.request_handler
    async def new_ghid(self, connection, body):
        ''' Handles requests for new objects.
        '''
        raise NotImplementedError()
        
    @new_ghid.response_handler
    async def new_ghid(self, connection, response, exc):
        ''' Handles responses to requests for new objects.
        '''
        if exc is not None:
            raise exc
        
        else:
            return Ghid.from_bytes(response)
            
    @new_ghid.fixture
    async def new_ghid(self, state, api_id, dynamic, private, _legroom):
        ''' We just need an address.
        '''
        self.pending_obj = (
            self.pending_ghid,
            self.whoami,
            state,
            False,  # is_link
            api_id,
            private,
            _legroom
        )
        
        return self.pending_ghid
        
    @public_api
    @request(b'!O')
    async def update_ghid(self, connection, ghid, state, private, _legroom):
        ''' Update an object or notify an app of an incoming update.
        '''
        # If state is Ghid, it's a link.
        if isinstance(state, Ghid):
            is_link = True
        else:
            is_link = False
            
        return self._pack_object_def(
            ghid,       # ghid
            None,       # Author
            state,      # state
            is_link,    # is_link
            None,       # api_id
            private,    # private
            None,       # dynamic
            _legroom    # legroom
        )
        
    @update_ghid.request_handler
    async def update_ghid(self, connection, body):
        ''' Handles update object requests.
        '''
        (address,
         author,    # Will be unused and set to None
         state,
         is_link,
         api_id,    # Will be unused and set to None
         private,   # Will be unused and set to None
         dynamic,   # Will be unused and set to None
         _legroom   # Will be unused and set to None
         ) = self._unpack_object_def(body)
        
        if is_link:
            state = Ghid.from_bytes(state)
            
        await self._hgxlink._pull_state(address, state)
            
        return b'\x01'
        
    @update_ghid.response_handler
    async def update_ghid(self, connection, response, exc):
        ''' Handles responses to update object requests.
        '''
        if exc is not None:
            raise exc
            
        elif response != b'\x01':
            raise HGXLinkError('Unknown error while updating object.')
            
        else:
            return True
            
    @update_ghid.fixture
    async def update_ghid(self, ghid, state, private, _legroom):
        ''' Yarp, fixture that.
        '''
        self.updates.append(
            {ghid: (state, private, _legroom)}
        )
        
    @public_api
    @request(b'~O')
    async def sync_ghid(self, connection, ghid):
        ''' Manually force Hypergolix to check an object for updates.
        Client only.
        '''
        return bytes(ghid)
        
    @sync_ghid.request_handler
    async def sync_ghid(self, connection, body):
        ''' Handles manual syncing requests. Server only.
        '''
        raise NotImplementedError()
        
    @sync_ghid.response_handler
    async def sync_ghid(self, connection, response, exc):
        ''' Handles responses to manual syncing requests. Client only.
        '''
        if exc is not None:
            raise exc
        elif response != b'\x01':
            raise IPCError('Unknown error while updating object.')
        else:
            return True
            
    @sync_ghid.fixture
    async def sync_ghid(self, ghid):
        # Moar fixturing.
        self.syncs.append(ghid)
        
    @public_api
    @request(b'@O')
    async def share_ghid(self, connection, ghid, recipient):
        ''' Request an object share or notify an app of an incoming
        share.
        '''
        return bytes(ghid) + bytes(recipient)
        
    @share_ghid.request_handler
    async def share_ghid(self, connection, body):
        ''' Handles object share requests.
        '''
        ghid = Ghid.from_bytes(body[0:65])
        origin = Ghid.from_bytes(body[65:130])
        api_id = ApiID.from_bytes(body[130:195])
        
        await self._hgxlink.handle_share(ghid, origin, api_id)
        return b'\x01'
        
    @share_ghid.response_handler
    async def share_ghid(self, connection, response, exc):
        ''' Handles responses to object share requests.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while sharing object.')
            
    @share_ghid.fixture
    async def share_ghid(self, ghid, recipient):
        # Blahblah
        self.shares.add(ghid, recipient)
        
    @request(b'^S')
    async def share_success(self, connection):
        ''' Notify app of successful share. Server only.
        '''
        raise NotImplementedError()
        
    @share_success.request_handler
    async def share_success(self, connection, body):
        ''' Handles app notifications for successful shares. Client
        only.
        '''
        # Don't raise NotImplementedError, just because this will be called for
        # every share, and we're perfectly well aware that it isn't implemented
        # TODO: implement this.
        return b''
        
    @request(b'^F')
    async def share_failure(self, connection):
        ''' Notify app of unsuccessful share. Server only.
        '''
        raise NotImplementedError()
        
    @share_failure.request_handler
    async def share_failure(self, connection, body):
        ''' Handles app notifications for unsuccessful shares. Client
        only.
        '''
        # Don't raise NotImplementedError, just because this will be called for
        # every share, and we're perfectly well aware that it isn't implemented
        # TODO: implement this.
        return b''
        
    @public_api
    @request(b'*O')
    async def freeze_ghid(self, connection, ghid):
        ''' Creates a new static copy of the object, or notifies an app
        of a frozen copy of an existing object created by a concurrent
        instance of the app.
        '''
        return bytes(ghid)
        
    @freeze_ghid.request_handler
    async def freeze_ghid(self, connection, body):
        ''' Handles object freezing requests.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @freeze_ghid.response_handler
    async def freeze_ghid(self, connection, response, exc):
        ''' Handles responses to object freezing requests.
        '''
        if exc is not None:
            raise exc
        
        else:
            return Ghid.from_bytes(response)
            
    @freeze_ghid.fixture
    async def freeze_ghid(self, ghid):
        # Moar fixture.
        self.frozen.add(ghid)
        return ghid
        
    @public_api
    @request(b'#O')
    async def hold_ghid(self, connection, ghid):
        ''' Creates a new static binding for the object, or notifies an
        app of a static binding created by a concurrent instance of the
        app.
        '''
        return bytes(ghid)
        
    @hold_ghid.request_handler
    async def hold_ghid(self, connection, body):
        ''' Handles object holding requests.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @hold_ghid.response_handler
    async def hold_ghid(self, connection, response, exc):
        ''' Handles responses to object holding requests.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while holding object.')
            
    @hold_ghid.fixture
    async def hold_ghid(self, ghid):
        # Yep yep yep
        self.held.add(ghid)
        
    @public_api
    @request(b'-O')
    async def discard_ghid(self, connection, ghid):
        ''' Stop listening to object updates. Client only.
        '''
        return bytes(ghid)
        
    @discard_ghid.request_handler
    async def discard_ghid(self, connection, body):
        ''' Handles object discarding requests. Server only.
        '''
        raise NotImplementedError()
        
    @discard_ghid.response_handler
    async def discard_ghid(self, connection, response, exc):
        ''' Handles responses to object discarding requests. Client
        only.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while discarding object.')
            
    @discard_ghid.fixture
    async def discard_ghid(self, ghid):
        # Nothing special.
        self.discarded.add(ghid)
        
    @public_api
    @request(b'XO')
    async def delete_ghid(self, connection, ghid):
        ''' Request an object deletion or notify an app of an incoming
        deletion.
        '''
        return bytes(ghid)
        
    @delete_ghid.request_handler
    async def delete_ghid(self, connection, body):
        ''' Handles object deletion requests.
        '''
        ghid = Ghid.from_bytes(body)
        await self._hgxlink.handle_delete(ghid)
        return b'\x01'
        
    @delete_ghid.response_handler
    async def delete_ghid(self, connection, response, exc):
        ''' Handles responses to object deletion requests.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x01':
            return True
        else:
            raise IPCError('Unknown error while deleting object.')
            
    @delete_ghid.fixture
    async def delete_ghid(self, ghid):
        # mmmmhmmm
        self.deleted.add(ghid)
import unittest
import random
import logging
from loopa import TaskCommander
from loopa.utils import await_coroutine_threadsafe
from hypergolix.comms import BasicServer
from hypergolix.comms import WSConnection
from hypergolix.comms import ConnectionManager
from hypergolix.core import GolixCore
from hypergolix.core import Oracle
from hypergolix.dispatch import _Dispatchable
from hypergolix.dispatch import Dispatcher
from hypergolix.remotes import Salmonator
from hypergolix.rolodex import Rolodex
from hypergolix.accounting import Account
from hypergolix.utils import Aengel
from hypergolix.utils import SetMap
from hypergolix.utils import WeakSetMap
from hypergolix.utils import ApiID
from hypergolix.utils import AppToken
from hypergolix.embed import HGXLink
from golix import Ghid
from hypergolix.exceptions import HypergolixException
from hypergolix.exceptions import IPCError
from hypergolix.ipc import IPCServerProtocol
from hypergolix.ipc import IPCClientProtocol
from _fixtures.ghidutils import make_random_ghid
from _fixtures.identities import TEST_AGENT1
from _fixtures.identities import TEST_AGENT2
    from hypergolix import logutils
        self.assertIn(apiid_2, self.dispatch._conns_from_api)
        
    def test_whoami(self):
        whoami = await_coroutine_threadsafe(
            coro = self.client1.get_whoami(timeout=1),
            loop = self.client1_commander._loop
        )
        
        self.assertEqual(whoami, self.golcore.whoami)
        
    def test_startup_obj(self):
        ''' Test getting and setting startup objects.
        '''
        self.dispatch.RESET()
        
        # We need a token to do anything with startup objects, but we don't
        # need to do anything WITH the token.
        await_coroutine_threadsafe(
            coro = self.client1.set_token(None, timeout=1),
            loop = self.client1_commander._loop
        )
        
        ghid = await_coroutine_threadsafe(
            coro = self.client1.get_startup_obj(timeout=1),
            loop = self.client1_commander._loop
        )
        self.assertIsNone(ghid)
        
        obj = make_random_ghid()
        await_coroutine_threadsafe(
            coro = self.client1.register_startup_obj(obj, timeout=1),
            loop = self.client1_commander._loop
        )
        ghid = await_coroutine_threadsafe(
            coro = self.client1.get_startup_obj(timeout=1),
            loop = self.client1_commander._loop
        )
        self.assertEqual(obj, ghid)
        
        await_coroutine_threadsafe(
            coro = self.client1.deregister_startup_obj(timeout=1),
            loop = self.client1_commander._loop
        )
        
    def test_obj_get(self):
        ''' Test getting and making objects.
        '''
        # Test setup
        self.oracle.RESET()
        self.dispatch.RESET()
        seed_state = bytes([random.randint(0, 255) for i in range(0, 20)])
        obj = _Dispatchable.__fixture__(
            ghid = make_random_ghid(),
            dynamic = True,
            author = self.golcore.whoami,
            legroom = 7,
            api_id = ApiID(bytes(64)),
            state = seed_state,
            dispatch = self.dispatch,
            ipc_protocol = self.server_protocol,
            golcore = self.golcore,
            ghidproxy = self,   # Well, we can't use None because weakref...
            privateer = self,   # Ditto...
            percore = self,     # Ditto...
            librarian = self,   # Ditto...
            account = self.account
        )
        self.oracle.add_object(obj.ghid, obj)
        
        # Test getting that object
        ghid, author, state, is_link, api_id, private, dynamic, _legroom =\
            await_coroutine_threadsafe(
                coro = self.client1.get_ghid(obj.ghid),
                loop = self.client1_commander._loop
            )
        self.assertEqual(ghid, obj.ghid)
        self.assertEqual(state, seed_state)
        self.assertEqual(author, self.golcore.whoami)
        
    def test_obj_new(self):
        # Test setup
        self.oracle.RESET()
        self.dispatch.RESET()
        seed_state = bytes([random.randint(0, 255) for i in range(0, 20)])
        obj = _Dispatchable.__fixture__(
            ghid = make_random_ghid(),
            dynamic = True,
            author = self.golcore.whoami,
            legroom = 7,
            api_id = ApiID(bytes(64)),
            state = seed_state,
            dispatch = self.dispatch,
            ipc_protocol = self.server_protocol,
            golcore = self.golcore,
            ghidproxy = self,   # Well, we can't use None because weakref...
            privateer = self,   # Ditto...
            percore = self,     # Ditto...
            librarian = self,   # Ditto...
            account = self.account
        )
        self.oracle.add_object(obj.ghid, obj)
        
        # Test "creating" a "new" object
        ghid = await_coroutine_threadsafe(
            coro = self.client1.new_ghid(
                b'hello world',
                ApiID(bytes(64)),
                True,
                False,
                7
            ),
            loop = self.client1_commander._loop
        )
        self.assertEqual(ghid, obj.ghid)
        
    def test_obj_update(self):
        # Test setup
        self.oracle.RESET()
        self.dispatch.RESET()
        seed_state = bytes([random.randint(0, 255) for i in range(0, 20)])

        obj = _Dispatchable.__fixture__(