update

Author: bierdok

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='talkops',
-    version='1.3.1',
+    version='1.3.2',
     author='PicoUX',
     description="TalkOps SDK",
     long_description=open('README.md',encoding='utf-8').read(),
@@ -16,7 +16,7 @@
         'Issues': 'https://github.com/talkops/sdk-python/issues',
         'Source': 'https://github.com/talkops/sdk-python',
     },
-    install_requires=['aiohttp-sse-client', 'jinja2', 'nest_asyncio'],
+    install_requires=['aiofiles', 'aiohttp-sse-client', 'jinja2', 'nest_asyncio'],
     keywords=['sdk'],
     classifiers=[
       "Programming Language :: Python :: 3",

talkops/extension.py

@@ -29,7 +29,6 @@ def __init__(self, token=None):
         self._publisher = None
         self._software_version = None
         self._started = False
-        self._subscriber = None
         self._token = token or os.environ.get('TALKOPS_TOKEN')
         self._website = None
 
@@ -77,7 +76,7 @@ async def _setup(self):
                     'functionSchemas': self._function_schemas,
                 }
             )
-            self._subscriber = Subscriber(
+            Subscriber(
                 lambda: {
                     'callbacks': self._callbacks,
                     'extension': self,
@@ -87,10 +86,6 @@ async def _setup(self):
                     'publisher': self._publisher,
                 }
             )
-            await asyncio.gather(
-                self._publisher.start(),
-                self._subscriber.start()
-            )
 
     def start(self):
         if self._started:

talkops/publisher.py

@@ -1,8 +1,8 @@
 from urllib.parse import urlencode
+import aiofiles.os
 import aiohttp
 import asyncio
 import json
-import sys
 import time
 
 class Publisher:
@@ -11,38 +11,20 @@ def __init__(self, use_config, use_state):
         self._use_state = use_state
         self._last_event_state = None
         self._last_ping_at = None
-
-    async def start(self):
-        self._original_stdout_write = sys.stdout.write
-        self._original_stderr_write = sys.stderr.write
-        def stdout_wrapper(chunk):
-            data = chunk.strip()
-            if(data != ""):
-                asyncio.create_task(self.publish_event({
-                    'type': 'stdout',
-                    'data': data,
-                    'time': time.time()
-                }))
-            return self._original_stdout_write(chunk)
-        def stderr_wrapper(chunk):
-            if "KeyboardInterrupt" not in chunk:
-                data = chunk.strip()
-                if(data != ""):
-                    asyncio.create_task(self.publish_event({
-                        'type': 'stderr',
-                        'data': data,
-                        'time': time.time()
-                    }))
-            return self._original_stderr_write(chunk)
-        sys.stdout.write = stdout_wrapper
-        sys.stderr.write = stderr_wrapper
-
         asyncio.create_task(self._publish_data({'type': 'init'}))
-        await asyncio.sleep(0.5)
-        asyncio.create_task(self._periodic_publish_state())
+        asyncio.create_task(self._publish_state_periodically())
+
+    async def _generate_event_state(self):
+        state = self._use_state()
+        try:
+            stat_result = await aiofiles.os.stat("error.log")
+            state["hasErrors"] = stat_result.st_size > 0
+        except FileNotFoundError:
+            state["hasErrors"] = False
+        return {"type": "state", "state": state}
 
     async def publish_state(self):
-        event = {'type': 'state', 'state': self._use_state()}
+        event = await self._generate_event_state()
         self._last_event_state = json.dumps(event)
         await self.publish_event(event)
 
@@ -76,11 +58,11 @@ async def _publish_data(self, data):
                     raise Exception(f"Failed to publish event: {response.status}")
                 return await response.text()
 
-    async def _periodic_publish_state(self):
+    async def _publish_state_periodically(self):
         while True:
-            event = {'type': 'state', 'state': self._use_state()}
+            await asyncio.sleep(0.5)
+            event = await self._generate_event_state()
             last_event_state = json.dumps(event)
             if self._last_event_state != last_event_state:
-                await self.publish_event(event)
                 self._last_event_state = last_event_state
-            await asyncio.sleep(1.0)
+                asyncio.create_task(self.publish_event(event))

talkops/subscriber.py

@@ -1,4 +1,5 @@
 from aiohttp_sse_client.client import EventSource
+from pathlib import Path
 from urllib.parse import quote
 import asyncio
 import json
@@ -7,8 +8,9 @@
 class Subscriber:
     def __init__(self, use_config):
         self._use_config = use_config
+        asyncio.create_task(self._subscribe())
 
-    async def start(self):
+    async def _subscribe(self):
         config = self._use_config()
         mercure = config['mercure']
         while True:
@@ -31,6 +33,14 @@ async def _on_event(self, event):
         if event['type'] == 'ping':
             asyncio.create_task(config['publisher'].on_ping())
             return
+        if event['type'] == 'debug':
+            file_path = 'error.log'
+            file = Path(file_path)
+            event['data'] = file.read_text(encoding='utf-8')
+            if event['data']:
+                asyncio.create_task(config['publisher'].publish_event(event))
+                file.write_text('', encoding='utf-8')
+            return
         if event['type'] == 'function_call':
             for function in config['functions']:
                 if function.__name__ != event['name']: