Add kci event commands for unicast list push and pop

kernelci/api/__init__.py

@@ -194,6 +194,14 @@ def send_event(self, channel: str, data):
     def receive_event(self, sub_id: int) -> CloudEvent:
         """Listen and receive an event from a given subscription id"""
 
+    @abc.abstractmethod
+    def push_event(self, list_name: str, data):
+        """Push an event to a given Redis List"""
+
+    @abc.abstractmethod
+    def pop_event(self, list_name: str) -> CloudEvent:
+        """Listen and pop an event from a given List"""
+
     # -----
     # Nodes
     # -----

kernelci/api/helper.py

@@ -43,6 +43,10 @@ def receive_event_data(self, sub_id):
         """Receive CloudEvent from Pub/Sub and return its data payload"""
         return self.api.receive_event(sub_id).data
 
+    def pop_event_data(self, list_name):
+        """Receive CloudEvent from Redis list and return its data payload"""
+        return self.api.pop_event(list_name).data
+
     def get_node_from_event(self, event_data):
         """Listen for an event and get the matching node object from it"""
         return self.api.get_node(event_data['id'])

kernelci/api/latest.py

@@ -6,6 +6,7 @@
 """KernelCI API bindings for the latest version"""
 
 import enum
+import json
 from typing import Optional, Sequence
 
 from cloudevents.http import from_json
@@ -72,6 +73,17 @@ def receive_event(self, sub_id: int):
                 continue
             return event
 
+    def push_event(self, list_name: str, data):
+        self._post('/'.join(['push', list_name]), data)
+
+    def pop_event(self, list_name: str):
+        path = '/'.join(['pop', str(list_name)])
+        while True:
+            resp = self._get(path)
+            data = json.dumps(resp.json())
+            event = from_json(data)
+            return event
+
     def get_node(self, node_id: str) -> dict:
         return self._get(f'node/{node_id}').json()
 

kernelci/cli/event.py

@@ -24,8 +24,8 @@
 
 @kci_event.command(secrets=True)
 @click.argument('channel')
 @Args.config
 @Args.api
 def subscribe(config, api, channel, secrets):
     """Subscribe to a Pub/Sub channel"""
     configs = kernelci.config.load(config)
@@ -37,8 +37,8 @@
 
 @kci_event.command(secrets=True)
 @click.argument('sub_id')
 @Args.config
 @Args.api
 def unsubscribe(config, api, sub_id, secrets):
     """Unsubscribe from a Pub/Sub channel"""
     configs = kernelci.config.load(config)
@@ -49,8 +49,8 @@
 
 @kci_event.command(secrets=True)
 @click.option('--is-json', help="Parse input data as JSON", is_flag=True)
 @Args.config
 @Args.api
 @click.argument('channel')
 def send(config, api, is_json, channel, secrets):
     """Read some data on stdin and send it as an event on a channel"""
@@ -65,9 +65,9 @@
 
 @kci_event.command(secrets=True)
 @click.argument('sub_id')
 @Args.config
 @Args.api
 @Args.indent
 def receive(config, api, indent, sub_id, secrets):
     """Wait and receive an event from a subscription and print on stdout"""
     configs = kernelci.config.load(config)
@@ -81,3 +81,39 @@
         click.echo(json.dumps(event, indent=indent))
     else:
         click.echo(event)
+
+
+@kci_event.command(secrets=True)
+@click.option('--is-json', help="Parse input data as JSON", is_flag=True)
+@Args.config
+@Args.api
+@click.argument('list_name')
+def push(config, api, is_json, list_name, secrets):
+    """Read some data on stdin and push it as an event on a list"""
+    configs = kernelci.config.load(config)
+    api_config = configs['api'][api]
+    api = kernelci.api.get_api(api_config, secrets.api.token)
+    data = sys.stdin.read()
+    if is_json:
+        data = json.loads(data)
+    api.push_event(list_name, {'data': data})
+
+
+@kci_event.command(secrets=True)
+@click.argument('list_name')
+@Args.config
+@Args.api
+@Args.indent
+def pop(config, api, indent, list_name, secrets):
+    """Wait and pop an event from a List when received print on stdout"""
+    configs = kernelci.config.load(config)
+    api_config = configs['api'][api]
+    api = kernelci.api.get_api(api_config, secrets.api.token)
+    helper = kernelci.api.helper.APIHelper(api)
+    event = helper.pop_event_data(list_name)
+    if isinstance(event, str):
+        click.echo(event.strip())
+    elif isinstance(event, dict):
+        click.echo(json.dumps(event, indent=indent))
+    else:
+        click.echo(event)