* rather than subclass instantiation, xapps now use initialization and registration functions to register handlers
* rmr xapps can now register handlers for specific message types (and they must prodive a default callback); if the user does this then "message to function routing" is now handled by the framework itself
* RMRXapp now runs the polling loop in a thread, and returns execution back to the caller. The user is then free to loop, or do nothing, and call stop() when they want.
* Raises tox coverage minimum to 70 from 50 (currently at 86)
Issue-ID: RIC-228
Change-Id: I15bfb708dbd14a46dc1207296e77383642d22b29
Signed-off-by: Tommy Carpenter <tc677g@att.com>
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2019-2020 Nokia
+# Copyright (c) 2018-2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+FROM python:3.7-alpine
+
+# sdl uses hiredis which needs gcc
+RUN apk update && apk add gcc musl-dev
+
+# copy rmr .sos from the builder image
+COPY --from=nexus3.o-ran-sc.org:10004/bldr-alpine3-go:1-rmr1.13.1 /usr/local/lib64/libnng.so /usr/local/lib64/libnng.so
+COPY --from=nexus3.o-ran-sc.org:10004/bldr-alpine3-go:1-rmr1.13.1 /usr/local/lib64/librmr_nng.so /usr/local/lib64/librmr_nng.so
+
+# Upgrade pip, install tox
+RUN pip install --upgrade pip && pip install tox
+
+# copies
+COPY ricxappframe/ /tmp/ricxappframe
+COPY tests/ /tmp/tests
+COPY setup.py tox.ini LICENSE.txt /tmp/
+WORKDIR /tmp
+
+# Run the unit tests
+RUN tox -e code,flake8
---------
This class of Xapps are purely reactive to rmr; data is always "pushed" to it via rmr.
That is, every time the Xapp receives an rmr message, they do something, then wait for the next message to arrive, end never need to execute functionality at another time (if they do, use the next class).
-This is represented by a very simple callback `consume` that is invoked every time an rmr message arrives (note, this is subject to change, with more callbacks for specific messages like `A1_POLICY_REQUEST`).
+This is represented by a series of callbacks that get registered to receive rmr message types.
+Every time an rmr message arrives, the user callback for that message type is invoked, or if the user has not registered a callback for that type, their default callback (mandatory) is invoked.
An analogy of this is AWS Lambda: "execute this code every time an event comes in" (the code to execute can depend on the type of event).
General Xapps
This is important because rmr is *not* a persistent message bus, if any rmr client does not read "fast enough", messages can be lost.
So in this framework the client code is not in the same thread as the rmr reads, so that long running client code can never lead to lost messages.
+In the case of RMR Xapps, there are currently 3 total threads; the thread that reads from rmr directly, the thread that reads from the queue and invokes the client callback, and the user thread. Running the xapp returns to the user and runs until the user calls `stop`.
+
Examples
--------
There are two examples in the `examples` directory; `ping` which is a general Xapp, and `pong` which is an RMR Xapp.
::
* a logger has to be provided to the xapp
- * the ability to specify more callacks per message type?
-
:depth: 3
:local:
+
+[0.3.0] - 3/10/2020
+-------------------
+::
+
+ * Large change to the "feel" of this framework: rather than subclass instantiation, xapps now use initialization and registration functions to register handlers
+ * rmr xapps can now register handlers for specific message types (and they must prodive a default callback); if the user does this then "message to function routing" is now handled by the framework itself
+ * RMRXapp now runs the polling loop in a thread, and returns execution back to the caller. The user is then free to loop, or do nothing, and call stop() when they want.
+ * Raises tox coverage minimum to 70 from 50 (currently at 86)
+
[0.2.0] - 3/3/2020
-------------------
::
- * now allows for RMR Xapps to call code before entering the infinite loop
+ * now allows for RMRXapps to call code before entering the infinite loop
* stop is now called before throwing NotImplemented in the case where the client fails to provide a must have callback; this ensures there is no dangling rmr thread
* stop now calls rmr_close to correctly free up any port(s)
* (breaking) renames `loop` to `entrypoint` since the function does not have to contain a loop (though it most likely does)
import time
import json
-from threading import Thread
-from rmr import rmr
from ricxappframe.xapp_frame import Xapp
-# Now we use the framework to echo back the acks
-class MyXapp(Xapp):
- def entrypoint(self):
- my_ns = "myxapp"
- number = 0
- while True:
- # test healthcheck
- print("Healthy? {}".format(xapp.healthcheck()))
- # rmr send
- val = json.dumps({"test_send": number}).encode()
- self.rmr_send(val, 60000)
- number += 1
+def entry(self):
+ my_ns = "myxapp"
+ number = 0
+ while True:
+ # test healthcheck
+ print("Healthy? {}".format(xapp.healthcheck()))
- # store it in SDL and read it back; delete and read
- self.sdl_set(my_ns, "numba", number)
- print((self.sdl_get(my_ns, "numba"), self.sdl_find_and_get(my_ns, "num")))
- self.sdl_delete(my_ns, "numba")
- print(self.sdl_get(my_ns, "numba"))
+ # rmr send to default handler
+ self.rmr_send(json.dumps({"sup": number}).encode(), 6660666)
- # rmr receive
- for (summary, sbuf) in self.rmr_get_messages():
- print(summary)
- self.rmr_free(sbuf)
+ # rmr send 60000, should trigger registered callback
+ val = json.dumps({"test_send": number}).encode()
+ self.rmr_send(val, 60000)
+ number += 1
- time.sleep(1)
+ # store it in SDL and read it back; delete and read
+ self.sdl_set(my_ns, "numba", number)
+ print((self.sdl_get(my_ns, "numba"), self.sdl_find_and_get(my_ns, "num")))
+ self.sdl_delete(my_ns, "numba")
+ print(self.sdl_get(my_ns, "numba"))
+ # rmr receive
+ for (summary, sbuf) in self.rmr_get_messages():
+ print(summary)
+ self.rmr_free(sbuf)
-xapp = MyXapp(4564, use_fake_sdl=True)
+ time.sleep(2)
+
+
+xapp = Xapp(entrypoint=entry, rmr_port=4564, use_fake_sdl=True)
xapp.run()
from ricxappframe.xapp_frame import RMRXapp
-# Note, this is an OOP pattern for this that I find slightly more natural
-# The problem is we want the client xapp to be able to call methods defined in the RMRXapp
-# Another exactly equivelent way would have been to use Closures like
-# def consume(summary, sbuf):
-# xapp.rts()
-# xapp = RMRXapp(consume)
-# However, the subclass looks slightly more natural. Open to the alternative.
+def post_init(_self):
+ """post init"""
+ print("ping xapp could do some useful stuff here!")
-class MyXapp(RMRXapp):
- def post_init(self):
- print("ping xapp could do some useful stuff here!")
+def sixtyh(self, summary, sbuf):
+ """callback for 60000"""
+ print("registered 60000 handler called!")
+ print(summary)
+ jpay = json.loads(summary["payload"])
+ self.rmr_rts(sbuf, new_payload=json.dumps({"ACK": jpay["test_send"]}).encode(), new_mtype=60001, retries=100)
+ self.rmr_free(sbuf)
- def consume(self, summary, sbuf):
- """callbnack called for each new message"""
- print(summary)
- jpay = json.loads(summary["payload"])
- self.rmr_rts(sbuf, new_payload=json.dumps({"ACK": jpay["test_send"]}).encode(), new_mtype=60001, retries=100)
- self.rmr_free(sbuf)
+def defh(self, summary, sbuf):
+ """default callback"""
+ print("default handler called!")
+ print(summary)
+ self.rmr_free(sbuf)
-xapp = MyXapp(use_fake_sdl=True)
+
+xapp = RMRXapp(default_handler=defh, post_init=post_init, use_fake_sdl=True)
+xapp.register_callback(sixtyh, 60000)
xapp.run()
newrt|start
+rte|6660666|localhost:4562
rte|60000|localhost:4562
newrt|end
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================
-
-
+from threading import Thread
from ricxappframe import xapp_rmr
from ricxappframe.xapp_sdl import SDLWrapper
from rmr import rmr
Base xapp; not for client use directly
"""
- def __init__(self, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False):
+ def __init__(self, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False, post_init=None):
"""
Init
use_fake_sdl: bool (optional)
if this is True, it uses dbaas' "fake dict backend" instead of Redis or other backends.
Set this to true when developing your xapp or during unit testing to completely avoid needing a dbaas running or any network at all
+
+ post_init: function (optional)
+ runs this user provided function after the base xapp is initialized
+ it's signature should be post_init(self)
"""
# Start rmr rcv thread
self._sdl = SDLWrapper(use_fake_sdl)
# run the optionally provided user post init
- self.post_init()
-
- # Public methods to be implemented by the client
- def post_init(self):
- """
- this method can optionally be implemented by the client to run code immediately after the xall initialized (but before the xapp starts it's processing loop)
- the base method here does nothing (ie nothing is executed prior to starting if the client does not implement this)
- """
- pass
+ if post_init:
+ post_init(self)
# Public rmr methods
def stop(self):
"""
- cleans up and stops the xapp.
- Currently this only stops the rmr thread
+ cleans up and stops the xapp rmr thread (currently)
This is critical for unit testing as pytest will never return if the thread is running.
TODO: can we register a ctrl-c handler so this gets called on ctrl-c? Because currently two ctrl-c are needed to stop
When run is called, the xapp framework waits for rmr messages, and calls the client provided consume callback on every one
"""
- def consume(self, summary, sbuf):
+ def __init__(self, default_handler, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False, post_init=None):
+ """
+ Parameters
+ ----------
+ default_handler: function
+ a function with the signature (summary, sbuf) to be called when a message of type message_type is received
+ summary: dict
+ the rmr message summary
+ sbuf: ctypes c_void_p
+ Pointer to an rmr message buffer. The user must call free on this when done.
+
+ post_init: function (optional)
+ optionally runs this function after the app initializes and before the run loop
+ it's signature should be post_init(self)
+
+ For the other parameters, see _BaseXapp
+ """
+ # init base
+ super().__init__(
+ rmr_port=rmr_port, rmr_wait_for_ready=rmr_wait_for_ready, use_fake_sdl=use_fake_sdl, post_init=post_init
+ )
+
+ # setup callbacks
+ self._default_handler = default_handler
+ self._dispatch = {}
+
+ # used for thread control
+ self._keep_going = True
+
+ def register_callback(self, handler, message_type):
"""
- This function is to be implemented by the client and is called whenever a new rmr message is received.
- It is expected to take two parameters (besides self):
+ registers this xapp to call handler(summary, buf) when an rmr message is received of type message_type
Parameters
----------
- summary: dict
- the rmr message summary
- sbuf: ctypes c_void_p
- Pointer to an rmr message buffer. The user must call free on this when done.
+ handler: function
+ a function with the signature (summary, sbuf) to be called when a message of type message_type is received
+ summary: dict
+ the rmr message summary
+ sbuf: ctypes c_void_p
+ Pointer to an rmr message buffer. The user must call free on this when done.
+
+ message:type: int
+ the message type to look for
+
+ Note if this method is called multiple times for a single message type, the "last one wins".
"""
- self.stop()
- raise NotImplementedError()
+ self._dispatch[message_type] = handler
def run(self):
"""
- This function should be called when the client xapp is ready to wait for consume to be called on received messages
+ This function should be called when the client xapp is ready to wait for their handlers to be called on received messages
+
+ execution is returned to caller
+ """
+
+ def loop():
+ while self._keep_going:
+ if not self._rmr_loop.rcv_queue.empty():
+ (summary, sbuf) = self._rmr_loop.rcv_queue.get()
+ # dispatch
+ func = self._dispatch.get(summary["message type"], None)
+ if not func:
+ func = self._default_handler
+ func(self, summary, sbuf)
- TODO: should we run this in a thread too? We can't currently call "stop" on rmr xapps at an arbitrary time because this doesn't return control
- Running the below in a thread probably makes the most sense.
+ Thread(target=loop).start()
+
+ def stop(self):
"""
- while True:
- if not self._rmr_loop.rcv_queue.empty():
- (summary, sbuf) = self._rmr_loop.rcv_queue.get()
- self.consume(summary, sbuf)
+ stops the rmr xapp completely.
+ """
+ super().stop()
+ mdc_logger.debug("Stopping queue reading thread..")
+ self._keep_going = False
class Xapp(_BaseXapp):
Represents an xapp where the client provides a generic function to call, which is mostly likely a loop-forever loop
"""
- def entrypoint(self):
+ def __init__(self, entrypoint, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False):
"""
- This function is to be implemented by the client and is called after post_init
+ Parameters
+ ----------
+ entrypoint: function
+ this function is called when the xapp runs; this is the user code
+ it's signature should be function(self)
+
+ For the other parameters, see _BaseXapp
"""
- self.stop()
- raise NotImplementedError()
+ # init base
+ super().__init__(rmr_port=rmr_port, rmr_wait_for_ready=rmr_wait_for_ready, use_fake_sdl=use_fake_sdl)
+ self._entrypoint = entrypoint
def run(self):
"""
This function should be called when the client xapp is ready to start their code
"""
- self.entrypoint()
+ self._entrypoint(self)
+
+ # there is no need for stop currently here (base has, and nothing special to do here)
def stop(self):
"""
sets a flag that will cleanly stop the thread
- note, this does not yet have a use yet for xapps to call, however this is very handy during unit testing.
"""
mdc_logger.debug("Stopping rmr thread. Waiting for last iteration to finish..")
self._keep_going = False
setup(
name="ricxappframe",
- version="0.2.0",
+ version="0.3.0",
packages=find_packages(exclude=["tests.*", "tests"]),
author="Tommy Carpenter",
description="Xapp framework for python",
--- /dev/null
+# do NOT use localhost, seems unresolved on jenkins VMs
+newrt|start
+mse| 60000 | -1 | 127.0.0.1:4564
+mse| 60001 | -1 | 127.0.0.1:4564
+newrt|end
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2020 Nokia
+# Copyright (c) 2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+import time
+import pytest
+from rmr.exceptions import InitFailed
+from ricxappframe.xapp_frame import Xapp, RMRXapp
+
+
+def test_bad_init():
+ """test that an xapp whose rmr fails to init blows up"""
+
+ def entry(self):
+ pass
+
+ with pytest.raises(InitFailed):
+ bad_xapp = Xapp(entrypoint=entry, rmr_port=-1)
+ bad_xapp.run() # we wont get here
+
+ def defh(self):
+ pass
+
+ with pytest.raises(InitFailed):
+ bad_xapp = RMRXapp(default_handler=defh, rmr_port=-1)
+ bad_xapp.run() # we wont get here
+
+
+def test_init_general_xapp():
+ def entry(self):
+ # normally we would have some kind of loop here
+ print("bye")
+
+ gen_xapp = Xapp(entrypoint=entry, rmr_wait_for_ready=False, use_fake_sdl=True)
+ gen_xapp.run()
+ time.sleep(1)
+ gen_xapp.stop() # pytest will never return without this.
+
+
+def test_init_rmr_xapp():
+ def post_init(self):
+ print("hey")
+
+ def foo(self, _summary, _sbuf):
+ pass
+
+ rmr_xapp = RMRXapp(foo, post_init=post_init, rmr_wait_for_ready=False, use_fake_sdl=True)
+ rmr_xapp.run()
+ time.sleep(1)
+ rmr_xapp.stop() # pytest will never return without this.
+++ /dev/null
-# ==================================================================================
-# Copyright (c) 2020 Nokia
-# Copyright (c) 2020 AT&T Intellectual Property.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# ==================================================================================
-import pytest
-from rmr.exceptions import InitFailed
-from ricxappframe.xapp_frame import Xapp
-
-gen_xapp = None
-rmr_xapp = None
-
-
-def test_bad_gen_xapp():
- """test that an xapp that does not implement entrypoint blows up"""
-
- class MyXapp(Xapp):
- def post_init(self):
- pass
-
- with pytest.raises(NotImplementedError):
- # missing entrypoint
- bad_xapp = MyXapp(rmr_wait_for_ready=False, use_fake_sdl=True)
- bad_xapp.run()
-
-
-def test_bad_init():
- """test that an xapp whose rmr fails to init blows up"""
-
- class MyXapp(Xapp):
- def entrypoint(self):
- pass
-
- with pytest.raises(InitFailed):
- bad_xapp = MyXapp(rmr_port=-1)
- bad_xapp.run() # we wont get here
-
-
-def test_init_general_xapp():
- class MyXapp(Xapp):
- def post_init(self):
- self.sdl_set("testns", "mykey", 6)
-
- def entrypoint(self):
- assert self.sdl_get("testns", "mykey") == 6
- assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6}
- assert self.healthcheck()
- # normally we would have some kind of loop here
- print("bye")
-
- global gen_xapp
- gen_xapp = MyXapp(rmr_wait_for_ready=False, use_fake_sdl=True)
- gen_xapp.run()
-
-
-def teardown_module():
- """
- module teardown
-
- pytest will never return without this.
- """
- gen_xapp.stop()
--- /dev/null
+# ==================================================================================
+# Copyright (c) 2020 Nokia
+# Copyright (c) 2020 AT&T Intellectual Property.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==================================================================================
+import json
+import time
+from contextlib import suppress
+from ricxappframe.xapp_frame import Xapp, RMRXapp
+
+rmr_xapp = None
+gen_xapp = None
+
+
+def test_flow():
+
+ # test variables
+ def_called = 0
+ sixty_called = 0
+
+ # create rmr app
+ global rmr_xapp
+
+ def post_init(self):
+ print("hey")
+
+ def default_handler(self, summary, sbuf):
+ nonlocal def_called
+ def_called += 1
+ assert json.loads(summary["payload"]) == {"test send 60001": 1}
+ self.rmr_free(sbuf)
+
+ rmr_xapp = RMRXapp(default_handler, post_init=post_init, rmr_port=4564, rmr_wait_for_ready=False, use_fake_sdl=True)
+
+ def sixtythou_handler(self, summary, sbuf):
+ nonlocal sixty_called
+ sixty_called += 1
+ assert json.loads(summary["payload"]) == {"test send 60000": 1}
+ self.rmr_free(sbuf)
+
+ rmr_xapp.register_callback(sixtythou_handler, 60000)
+ rmr_xapp.run()
+
+ time.sleep(1)
+
+ def entry(self):
+
+ self.sdl_set("testns", "mykey", 6)
+ assert self.sdl_get("testns", "mykey") == 6
+ assert self.sdl_find_and_get("testns", "myk") == {"mykey": 6}
+ assert self.healthcheck()
+
+ val = json.dumps({"test send 60000": 1}).encode()
+ self.rmr_send(val, 60000)
+
+ val = json.dumps({"test send 60001": 2}).encode()
+ self.rmr_send(val, 60001)
+
+ global gen_xapp
+ gen_xapp = Xapp(entrypoint=entry, rmr_wait_for_ready=False, use_fake_sdl=True)
+ gen_xapp.run()
+
+ time.sleep(1)
+
+ assert def_called == 1
+ assert sixty_called == 1
+
+
+def teardown_module():
+ """
+ this is like a "finally"; the name of this function is pytest magic
+ safer to put down here since certain failures above can lead to pytest never returning
+ for example if an exception gets raised before stop is called in any test function above, pytest will hang forever
+ """
+ with suppress(Exception):
+ gen_xapp.stop()
+ with suppress(Exception):
+ rmr_xapp.stop()
pytest-cov
setenv =
LD_LIBRARY_PATH = /usr/local/lib/:/usr/local/lib64
+ RMR_SEED_RT = tests/fixtures/test_local.rt
+ RMR_ASYNC_CONN = 0
+
commands =
- pytest --cov ricxappframe --cov-report xml --cov-report term-missing --cov-report html --cov-fail-under=50
+ pytest --cov ricxappframe --cov-report xml --cov-report term-missing --cov-report html --cov-fail-under=70
coverage xml -i
[testenv:flake8]
Code Review / ric-plt / xapp-frame-py.git / commitdiff