SEBA-405 Convert synchronizer framework to library
Change-Id: If8562f23dc15c7d18d7a8b040b33756708b3c5ec
diff --git a/lib/xos-synchronizer/xossynchronizer/backend.py b/lib/xos-synchronizer/xossynchronizer/backend.py
new file mode 100644
index 0000000..b404864
--- /dev/null
+++ b/lib/xos-synchronizer/xossynchronizer/backend.py
@@ -0,0 +1,165 @@
+# Copyright 2017-present Open Networking Foundation
+#
+# 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 __future__ import print_function
+import os
+import inspect
+import imp
+import sys
+import threading
+import time
+from xossynchronizer.steps.syncstep import SyncStep
+from xossynchronizer.event_loop import XOSObserver
+from xossynchronizer.model_policy_loop import XOSPolicyEngine
+from xossynchronizer.event_engine import XOSEventEngine
+from xossynchronizer.pull_step_engine import XOSPullStepEngine
+from xossynchronizer.modelaccessor import *
+
+from xosconfig import Config
+from multistructlog import create_logger
+
+log = create_logger(Config().get("logging"))
+
+
+class Backend:
+ def __init__(self, log=log):
+ self.log = log
+
+ def load_sync_step_modules(self, step_dir):
+ sync_steps = []
+
+ self.log.info("Loading sync steps", step_dir=step_dir)
+
+ for fn in os.listdir(step_dir):
+ pathname = os.path.join(step_dir, fn)
+ if (
+ os.path.isfile(pathname)
+ and fn.endswith(".py")
+ and (fn != "__init__.py")
+ and (not fn.startswith("test"))
+ ):
+
+ # we need to extend the path to load modules in the step_dir
+ sys_path_save = sys.path
+ sys.path.append(step_dir)
+ module = imp.load_source(fn[:-3], pathname)
+
+ self.log.debug("Loaded file: %s", pathname)
+
+ # reset the original path
+ sys.path = sys_path_save
+
+ for classname in dir(module):
+ c = getattr(module, classname, None)
+
+ # if classname.startswith("Sync"):
+ # print classname, c, inspect.isclass(c), issubclass(c, SyncStep), hasattr(c,"provides")
+
+ # make sure 'c' is a descendent of SyncStep and has a
+ # provides field (this eliminates the abstract base classes
+ # since they don't have a provides)
+
+ if inspect.isclass(c):
+ bases = inspect.getmro(c)
+ base_names = [b.__name__ for b in bases]
+ if (
+ ("SyncStep" in base_names)
+ and (hasattr(c, "provides") or hasattr(c, "observes"))
+ and (c not in sync_steps)
+ ):
+ sync_steps.append(c)
+
+ self.log.info("Loaded sync steps", steps=sync_steps)
+
+ return sync_steps
+
+ def run(self):
+ observer_thread = None
+ model_policy_thread = None
+ event_engine = None
+
+ steps_dir = Config.get("steps_dir")
+ if steps_dir:
+ sync_steps = []
+
+ # load sync_steps
+ if steps_dir:
+ sync_steps = self.load_sync_step_modules(steps_dir)
+
+ # if we have at least one sync_step
+ if len(sync_steps) > 0:
+ # start the observer
+ self.log.info("Starting XOSObserver", sync_steps=sync_steps)
+ observer = XOSObserver(sync_steps, self.log)
+ observer_thread = threading.Thread(
+ target=observer.run, name="synchronizer"
+ )
+ observer_thread.start()
+
+ else:
+ self.log.info("Skipping observer thread due to no steps dir.")
+
+ pull_steps_dir = Config.get("pull_steps_dir")
+ if pull_steps_dir:
+ self.log.info("Starting XOSPullStepEngine", pull_steps_dir=pull_steps_dir)
+ pull_steps_engine = XOSPullStepEngine()
+ pull_steps_engine.load_pull_step_modules(pull_steps_dir)
+ pull_steps_thread = threading.Thread(
+ target=pull_steps_engine.start, name="pull_step_engine"
+ )
+ pull_steps_thread.start()
+ else:
+ self.log.info("Skipping pull step engine due to no pull_steps_dir dir.")
+
+ event_steps_dir = Config.get("event_steps_dir")
+ if event_steps_dir:
+ self.log.info("Starting XOSEventEngine", event_steps_dir=event_steps_dir)
+ event_engine = XOSEventEngine(self.log)
+ event_engine.load_event_step_modules(event_steps_dir)
+ event_engine.start()
+ else:
+ self.log.info("Skipping event engine due to no event_steps dir.")
+
+ # start model policies thread
+ policies_dir = Config.get("model_policies_dir")
+ if policies_dir:
+ policy_engine = XOSPolicyEngine(policies_dir=policies_dir, log=self.log)
+ model_policy_thread = threading.Thread(
+ target=policy_engine.run, name="policy_engine"
+ )
+ model_policy_thread.is_policy_thread = True
+ model_policy_thread.start()
+ else:
+ self.log.info(
+ "Skipping model policies thread due to no model_policies dir."
+ )
+
+ if (not observer_thread) and (not model_policy_thread) and (not event_engine):
+ self.log.info(
+ "No sync steps, no policies, and no event steps. Synchronizer exiting."
+ )
+ # the caller will exit with status 0
+ return
+
+ while True:
+ try:
+ time.sleep(1000)
+ except KeyboardInterrupt:
+ print("exiting due to keyboard interrupt")
+ # TODO: See about setting the threads as daemons
+ if observer_thread:
+ observer_thread._Thread__stop()
+ if model_policy_thread:
+ model_policy_thread._Thread__stop()
+ sys.exit(1)