Merge branch 'stream-prototype'

ceiclient/client.py

@@ -332,6 +332,12 @@ class PDClient(DashiCeiClient):
     def set_system_boot(self, system_boot):
         self.connection.call(self.dashi_name, 'set_system_boot', system_boot=system_boot)
 
+    def add_engine(self, engine_id, slots, **kwargs):
+        definition = dict(engine_id=engine_id, slots=slots)
+        for key, val in kwargs.iteritems():
+            definition[key] = val
+        return self.connection.call(self.dashi_name, 'add_engine', definition=definition)
+
     def create_process_definition(self, process_definition=None, process_definition_id=None):
         if process_definition is None:
             raise CeiClientError("You must provide a process defintion")

ceiclient/commands.py

@@ -610,7 +610,8 @@ def _validate_process_definition(definition):
 
     module = executable.get('module')
     cls = executable.get('class')
-    if not module or not cls:
+    execu = executable.get('exec')
+    if not execu and (not module or not cls):
         raise ValueError("definition has invalid executable")
 
 
@@ -732,6 +733,19 @@ def __init__(self, subparsers):
     def execute(client, opts):
         return client.describe_process_definition(opts.process_definition_id)
 
+class PDNodeState(CeiCommand):
+
+    name = 'nodestate'
+
+    def __init__(self, subparsers):
+        parser = subparsers.add_parser(self.name)
+        parser.add_argument('node_id')
+        parser.add_argument('domain_id')
+        parser.add_argument('state')
+
+    @staticmethod
+    def execute(client, opts):
+        return client.node_state(opts.node_id, opts.domain_id, opts.state)
 
 class PDUpdateProcessDefinition(CeiCommand):
 
@@ -2011,7 +2025,7 @@ class Process(CeiService):
 
     commands = {}
     for command in [PDScheduleProcess, PDDescribeProcess, PDDescribeProcesses,
-            PDTerminateProcess, PDDump, PDRestartProcess, PDWaitProcess]:
+            PDTerminateProcess, PDDump, PDRestartProcess, PDWaitProcess, PDNodeState]:
         commands[command.name] = command
 
     @staticmethod