async log event

Author: Jack Kordas

client/globus_cw_client/client.py

@@ -2,6 +2,7 @@
 Python client API for cwlogs daemon
 """
 
+import asyncio
 import json
 import socket
 import time
@@ -41,6 +42,25 @@ def log_event(message, retries=10, wait=0.1):
     req["timestamp"] = int(time.time() * 1000)
     return _request(req, retries, wait)
 
+async def log_event_async(message, retries=10, wait=0.1):
+    if isinstance(message, bytes):
+        message = message.decode("utf-8")
+    _checktype(message, str, "message type must be bytes or unicode")
+
+    _checktype(retries, int, "retries must be an int")
+    if retries < 0:
+        raise ValueError("retries must be non-negative")
+
+    _checktype(wait, (int, float), "wait must be an int or float")
+    if wait < 0:
+        raise ValueError("wait must be non-negative")
+
+    req = {}
+    req["message"] = message
+    req["timestamp"] = int(time.time() * 1000)
+    return await _request_async(req, retries, wait)
+    
+
 
 def _connect(retries, wait):
     """
@@ -63,6 +83,21 @@ def _connect(retries, wait):
     raise CWLoggerConnectionError("couldn't connect to cw", error)
 
 
+async def _connect_async(retries, wait):
+    addr = "\0org.globus.cwlogs"
+    for _ in range(retries + 1):
+        try:
+            reader, writer = asyncio.open_unix_connection(path=addr)
+        except Exception as err:
+            if writer:
+                writer.close()
+            error = err
+        else:
+            return reader, writer
+        time.sleep(wait):
+
+
+
 def _request(req, retries, wait):
     buf = json.dumps(req, indent=None) + "\n"
     # dumps returns unicode with python3, but sock requires bytes
@@ -92,6 +127,32 @@ def _request(req, retries, wait):
         raise CWLoggerDaemonError("unknown response type", d)
 
 
+async def _request_async(req, retries, wait):
+    buf = json.dumps(req, indent=None) + "\n"
+    # dumps returns unicode with python3, but sock requires bytes
+    if isinstance(buf, str):
+        buf = buf.encode("utf-8")
+
+    reader, writer = await _connect_async(retries, wait)
+    writer.write(buf)
+    await writer.drain()
+
+    resp = await reader.readline()
+    if not resp.endswith(b"\n"):
+        raise Exception("no data")
+    resp = resp.decode("utf-8")
+
+    d = json.loads(resp[:-1])
+    if isinstance(d, dict):
+        status = d["status"]
+        if status == "ok":
+            return d
+        else:
+            raise CWLoggerDaemonError("forwarded error", d["message"])
+    else:
+        raise CWLoggerDaemonError("unknown response type", d)
+
+
 """
 Ignore (swallow) these exceptions at your own risk.
 CWLoggerDaemonError can be caused by many things, including but not limited to: