aicoap: a sample server for coap. scrapped however.

Signed-off-by: HeshamTB <hishaminv@gmail.com>

aiocoap/server.py

@@ -0,0 +1,121 @@
+import datetime
+import logging
+
+import asyncio
+
+import aiocoap.resource as resource
+import aiocoap
+
+
+class BlockResource(resource.Resource):
+    """Example resource which supports the GET and PUT methods. It sends large
+    responses, which trigger blockwise transfer."""
+
+    def __init__(self):
+        super().__init__()
+        self.set_content(b"This is the resource's default content. It is padded "
+                b"with numbers to be large enough to trigger blockwise "
+                b"transfer.\n")
+
+    def set_content(self, content):
+        self.content = content
+        while len(self.content) <= 1024:
+            self.content = self.content + b"0123456789\n"
+
+    async def render_get(self, request):
+        return aiocoap.Message(payload=self.content)
+
+    async def render_put(self, request):
+        print('PUT payload: %s' % request.payload)
+        self.set_content(request.payload)
+        return aiocoap.Message(code=aiocoap.CHANGED, payload=self.content)
+
+
+class SeparateLargeResource(resource.Resource):
+    """Example resource which supports the GET method. It uses asyncio.sleep to
+    simulate a long-running operation, and thus forces the protocol to send
+    empty ACK first. """
+
+    def get_link_description(self):
+        # Publish additional data in .well-known/core
+        return dict(**super().get_link_description(), title="A large resource")
+
+    async def render_get(self, request):
+        await asyncio.sleep(3)
+
+        payload = "Three rings for the elven kings under the sky, seven rings "\
+                "for dwarven lords in their halls of stone, nine rings for "\
+                "mortal men doomed to die, one ring for the dark lord on his "\
+                "dark throne.".encode('ascii')
+        return aiocoap.Message(payload=payload)
+
+class TimeResource(resource.ObservableResource):
+    """Example resource that can be observed. The `notify` method keeps
+    scheduling itself, and calles `update_state` to trigger sending
+    notifications."""
+
+    def __init__(self):
+        super().__init__()
+
+        self.handle = None
+
+    def notify(self):
+        self.updated_state()
+        self.reschedule()
+
+    def reschedule(self):
+        self.handle = asyncio.get_event_loop().call_later(5, self.notify)
+
+    def update_observation_count(self, count):
+        if count and self.handle is None:
+            print("Starting the clock")
+            self.reschedule()
+        if count == 0 and self.handle:
+            print("Stopping the clock")
+            self.handle.cancel()
+            self.handle = None
+
+    async def render_get(self, request):
+        payload = datetime.datetime.now().\
+                strftime("%Y-%m-%d %H:%M").encode('ascii')
+        return aiocoap.Message(payload=payload)
+
+class WhoAmI(resource.Resource):
+    async def render_get(self, request):
+        text = ["Used protocol: %s." % request.remote.scheme]
+
+        text.append("Request came from %s." % request.remote.hostinfo)
+        text.append("The server address used %s." % request.remote.hostinfo_local)
+
+        claims = list(request.remote.authenticated_claims)
+        if claims:
+            text.append("Authenticated claims of the client: %s." % ", ".join(repr(c) for c in claims))
+        else:
+            text.append("No claims authenticated.")
+
+        return aiocoap.Message(content_format=0,
+                payload="\n".join(text).encode('utf8'))
+
+# logging setup
+
+logging.basicConfig(level=logging.INFO)
+logging.getLogger("coap-server").setLevel(logging.DEBUG)
+
+async def main():
+    # Resource tree creation
+    root = resource.Site()
+
+    root.add_resource(['.well-known', 'core'],
+            resource.WKCResource(root.get_resources_as_linkheader))
+    root.add_resource(['time'], TimeResource())
+    root.add_resource(['other', 'block'], BlockResource())
+    root.add_resource(['other', 'separate'], SeparateLargeResource())
+    root.add_resource(['whoami'], WhoAmI())
+
+    await aiocoap.Context.create_server_context(root)
+
+    # Run forever
+    await asyncio.get_running_loop().create_future()
+
+if __name__ == "__main__":
+    asyncio.run(main())