Improve JSON parsing in handle function

Incomplete JSONs coming from Stream transport are now stored and completed, then processed with the callback function.

Author: dariusstefan

README.md

@@ -47,7 +47,7 @@ Currently, the following packages are available:
     hdl = OpenSIPSEventHandler(mi_connector, 'datagram', ip='127.0.0.1', port=50012)
 
     def some_callback(message):
-        # do something with the message
+        # do something with the message (it is a JSON object)
         pass
     
     ev: OpenSIPSEvent = None

opensips/event/__main__.py

@@ -108,8 +108,7 @@ def main():
     def event_handler(message):
         """ Event handler callback """
         try:
-            message_json = json.loads(message.decode('utf-8'))
-            print(json.dumps(message_json, indent=4))
+            print(json.dumps(message, indent=4))
         except json.JSONDecodeError as e:
             print(f"Failed to decode JSON: {e}")
 

opensips/event/event.py

@@ -22,6 +22,7 @@
 
 from threading import Thread, Event
 from ..mi import OpenSIPSMIException
+from .json_helper import extract_json
 
 class OpenSIPSEventException(Exception):
     """ Exceptions generated by OpenSIPS Events """
@@ -37,6 +38,8 @@ def __init__(self, handler, name: str, callback, expire=None):
         self.thread = None
         self.thread_stop = Event()
         self.thread_stop.clear()
+        self.buf = b""
+        self.json_queue = []
 
         try:
             self.socket = self._handler.__new_socket__()
@@ -55,8 +58,14 @@ def handle(self, callback):
         """ Handles the event callbacks """
         while not self.thread_stop.is_set():
             data = self.socket.read()
-            if data:
-                callback(data)
+            if not data:
+                continue
+            
+            self.buf += data
+            self.json_queue, self.buf = extract_json(self.json_queue, self.buf)
+            while self.json_queue:
+                json_obj = self.json_queue.pop(0)
+                callback(json_obj)
 
     def unsubscribe(self):
         """ Unsubscribes the event """

opensips/event/json_helper.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+##
+## This file is part of the OpenSIPS Python Package
+## (see https://github.com/OpenSIPS/python-opensips).
+##
+## This program is free software: you can redistribute it and/or modify
+## it under the terms of the GNU General Public License as published by
+## the Free Software Foundation, either version 3 of the License, or
+## (at your option) any later version.
+##
+## This program is distributed in the hope that it will be useful,
+## but WITHOUT ANY WARRANTY; without even the implied warranty of
+## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+## GNU General Public License for more details.
+##
+## You should have received a copy of the GNU General Public License
+## along with this program. If not, see <http://www.gnu.org/licenses/>.
+##
+
+import json
+from collections import OrderedDict
+from typing import Tuple
+
+def extract_json(json_acc: list, data: bytes) -> Tuple[list, bytes]:
+
+    """ Extracts JSON data from a byte stream """
+
+    while data:
+        try:
+            json_obj, idx = json.JSONDecoder(object_pairs_hook=OrderedDict).raw_decode(data.decode("utf-8"))
+            json_acc.append(json_obj)
+            data = data[idx:]
+        except json.JSONDecodeError as e:
+            break
+    return json_acc, data