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/asyncio/endpoint.py

@@ -163,7 +163,7 @@ async def run(self) -> AsyncIterator[RemoteEndpointAPI]:
             await self.stop()
 
     async def _start(self) -> None:
-        self._task = asyncio.ensure_future(self._run())
+        self._task = asyncio.ensure_future(self._process_incoming_messages())
         await self.wait_started()
 
     async def stop(self) -> None:

lahja/base.py

@@ -235,7 +235,7 @@ def is_ready(self) -> bool:
     def is_stopped(self) -> bool:
         return self._stopped.is_set()
 
-    async def _run(self) -> None:
+    async def _process_incoming_messages(self) -> None:
         self._running.set()
 
         # Send the hello message

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