Add trio based Endpoint implementation

.circleci/config.yml

@@ -69,12 +69,24 @@ jobs:
       - image: circleci/python:3.6
         environment:
           TOXENV: py36-core-asyncio
+  py36-core-trio:
+    <<: *common
+    docker:
+      - image: circleci/python:3.6
+        environment:
+          TOXENV: py36-core-trio
   py37-core-asyncio:
     <<: *common
     docker:
       - image: circleci/python:3.7
         environment:
           TOXENV: py37-core-asyncio
+  py37-core-trio:
+    <<: *common
+    docker:
+      - image: circleci/python:3.7
+        environment:
+          TOXENV: py37-core-trio
   py36-examples:
     <<: *common
     docker:
@@ -93,12 +105,24 @@ jobs:
       - image: circleci/python:3.6
         environment:
           TOXENV: py36-snappy-core-asyncio
+  py36-snappy-core-trio:
+    <<: *common
+    docker:
+      - image: circleci/python:3.6
+        environment:
+          TOXENV: py36-snappy-core-trio
   py37-snappy-core-asyncio:
     <<: *common
     docker:
       - image: circleci/python:3.7
         environment:
           TOXENV: py37-snappy-core-asyncio
+  py37-snappy-core-trio:
+    <<: *common
+    docker:
+      - image: circleci/python:3.7
+        environment:
+          TOXENV: py37-snappy-core-trio
 workflows:
   version: 2
   test:
@@ -108,8 +132,12 @@ workflows:
       - doctest
       - lint
       - py36-core-asyncio
-      - py37-core-asyncio
+      - py36-core-trio
       - py36-examples
       - py37-examples
+      - py37-core-asyncio
+      - py37-core-trio
       - py36-snappy-core-asyncio
+      - py36-snappy-core-trio
       - py37-snappy-core-asyncio
+      - py37-snappy-core-trio

lahja/base.py

@@ -241,11 +241,15 @@ def is_stopped(self) -> bool:
     async def _process_incoming_messages(self) -> None:
         self._running.set()
 
-        # Send the hello message
-        await self.send_message(Hello(self._local_name))
+        try:
+            # Send the hello message
+            await self.send_message(Hello(self._local_name))
+            # Wait for the other endpoint to identify itself.
+            hello = await self.conn.read_message()
+        except RemoteDisconnected:
+            self._stopped.set()
+            return
 
-        # Wait for the other endpoint to identify itself.
-        hello = await self.conn.read_message()
         if isinstance(hello, Hello):
             self._name = hello.name
             self.logger.debug(

lahja/exceptions.py

@@ -12,6 +12,16 @@ class BindError(LahjaError):
     """
 
 
+class LifecycleError(LahjaError):
+    """
+    Raised when attempting to violate the lifecycle of an endpoint such as
+    starting an already started endpoint or starting an endpoint that has
+    already stopped.
+    """
+
+    pass
+
+
 class ConnectionAttemptRejected(LahjaError):
     """
     Raised when an attempt was made to connect to an endpoint that is already connected.

lahja/tools/benchmark/backends.py

@@ -3,6 +3,7 @@
 
 from lahja.asyncio import AsyncioEndpoint
 from lahja.base import BaseEndpoint
+from lahja.trio import TrioEndpoint
 
 
 class BaseBackend(ABC):
@@ -40,3 +41,23 @@ async def sleep(seconds: float) -> None:
         import asyncio
 
         await asyncio.sleep(seconds)
+
+
+class TrioBackend(BaseBackend):
+    name = "trio"
+    Endpoint = TrioEndpoint
+
+    @staticmethod
+    def run(coro: Any, *args: Any) -> None:
+        # UNCOMMENT FOR DEBUGGING
+        # logger = multiprocessing.log_to_stderr()
+        # logger.setLevel(logging.INFO)
+        import trio
+
+        trio.run(coro, *args)
+
+    @staticmethod
+    async def sleep(seconds: float) -> None:
+        import trio
+
+        await trio.sleep(seconds)

lahja/trio/__init__.py

@@ -0,0 +1 @@
+from .endpoint import TrioEndpoint  # noqa: F401