CORD-1008 add required_models feature to modelaccessor
Change-Id: Iea4a0847c0255c4b8cd70a1e082b0b085b8a73d4
diff --git a/xos/coreapi/protos/utility.proto b/xos/coreapi/protos/utility.proto
index 4717581..6d5f18f 100644
--- a/xos/coreapi/protos/utility.proto
+++ b/xos/coreapi/protos/utility.proto
@@ -57,4 +57,11 @@
body: "*"
};
}
+
+ rpc NoOp(google.protobuf.Empty) returns (google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/xosapi/v1/utility/noop"
+ body: "*"
+ };
+ }
};
diff --git a/xos/coreapi/xos_utility_api.py b/xos/coreapi/xos_utility_api.py
index f595849..7377c91 100644
--- a/xos/coreapi/xos_utility_api.py
+++ b/xos/coreapi/xos_utility_api.py
@@ -104,3 +104,6 @@
return response
+ def NoOp(self, request, context):
+ return Empty()
+
diff --git a/xos/synchronizers/new_base/modelaccessor.py b/xos/synchronizers/new_base/modelaccessor.py
index 765b5d5..c715095 100644
--- a/xos/synchronizers/new_base/modelaccessor.py
+++ b/xos/synchronizers/new_base/modelaccessor.py
@@ -13,6 +13,9 @@
from xos.config import Config
from diag import update_diag
+from xos.logger import Logger, logging
+logger = Logger(level=logging.INFO)
+
class ModelAccessor(object):
def __init__(self):
self.all_model_classes = self.get_all_model_classes()
@@ -94,6 +97,27 @@
self.into=into
globals()["ModelLink"] = ModelLink
+def keep_trying(client, reactor):
+ # Keep checking the connection to wait for it to become unavailable.
+ # Then reconnect.
+
+ # logger.info("keep_trying") # message is unneccesarily verbose
+
+ from xosapi.xos_grpc_client import Empty
+
+ try:
+ client.utility.NoOp(Empty())
+ except:
+ # If we caught an exception, then the API has become unavailable.
+ # So reconnect.
+
+ logger.log_exc("exception in NoOp")
+ client.connected = False
+ client.connect()
+ return
+
+ reactor.callLater(1, functools.partial(keep_trying, client, reactor))
+
def grpcapi_reconnect(client, reactor):
global model_accessor
@@ -102,6 +126,33 @@
from apiaccessor import CoreApiModelAccessor
model_accessor = CoreApiModelAccessor(orm = client.xos_orm)
+
+ # If required_models is set, then check to make sure the required_models
+ # are present. If not, then the synchronizer needs to go to sleep until
+ # the models show up.
+
+ required_models = getattr(Config(), "observer_required_models", None)
+ if required_models:
+ required_models = required_models.split(",")
+ required_models = [x.strip() for x in required_models]
+
+ missing = []
+ found = []
+ for model in required_models:
+ if model_accessor.has_model_class(model):
+ found.append(model)
+ else:
+ missing.append(model)
+
+ logger.info("required_models: found %s" % ", ".join(found))
+ if missing:
+ logger.warning("required_models: missing %s" % ", ".join(missing))
+ # We're missing a required model. Give up and wait for the connection
+ # to reconnect, and hope our missing model has shown up.
+ reactor.callLater(1, functools.partial(keep_trying, client, reactor))
+ return
+
+ # import all models to global space
import_models_to_globals()
# Synchronizer framework isn't ready to embrace reactor yet...
diff --git a/xos/xos_client/xosapi/xos_grpc_client.py b/xos/xos_client/xosapi/xos_grpc_client.py
index 8213662..aaaa97f 100644
--- a/xos/xos_client/xosapi/xos_grpc_client.py
+++ b/xos/xos_client/xosapi/xos_grpc_client.py
@@ -57,7 +57,9 @@
try:
sys.path.append(self.work_dir)
m_protos = __import__(api + "_pb2")
+ reload(m_protos)
m_grpc = __import__(api + "_pb2_grpc")
+ reload(m_grpc)
finally:
sys.path = orig_sys_path