added register and foreach operations

Author: [email protected]

gearsclient/redisgears_builder.py

@@ -10,6 +10,14 @@ def AddToGB(self, gb, globalsDict):
         self.callback.__globals__.update(globalsDict)
         gb.map(self.callback)
 
+class GearsRemoteForeachStep():
+    def __init__(self, callback):
+        self.callback = callback
+
+    def AddToGB(self, gb, globalsDict):
+        self.callback.__globals__.update(globalsDict)
+        gb.foreach(self.callback)
+
 class GearsRemoteFlatMapStep():
     def __init__(self, callback):
         self.callback = callback
@@ -103,6 +111,13 @@ def __init__(self, arg, convertToStr, collect):
     def AddToGB(self, gb, globalsDict):
         gb.run(self.arg, self.convertToStr, self.collect)
 
+class GearsRemoteRegisterStep():
+    def __init__(self, arg):
+        self.arg = arg
+
+    def AddToGB(self, gb, globalsDict):
+        gb.register(self.arg) if self.arg else gb.register()
+
 
 class GearsPipe():
     def __init__(self, reader='KeysReader', defaultArg='*'):
@@ -114,6 +129,10 @@ def map(self, callback):
         self.steps.append(GearsRemoteMapStep(callback))
         return self
 
+    def foreach(self, callback):
+        self.steps.append(GearsRemoteForeachStep(callback))
+        return self
+
     def flatmap(self, callback):
         self.steps.append(GearsRemoteFlatMapStep(callback))
         return self
@@ -157,6 +176,9 @@ def limit(self, count, offset):
     def run(self, arg, convertToStr, collect):
         self.steps.append(GearsRemoteRunStep(arg, convertToStr, collect))
 
+    def register(self, arg):
+        self.steps.append(GearsRemoteRegisterStep(arg))
+
 
 class GearsRemoteBuilder():
     def __init__(self, reader='KeysReader', defaultArg='*', r=None):
@@ -169,6 +191,10 @@ def map(self, callback):
         self.pipe.map(callback)
         return self
 
+    def foreach(self, callback):
+        self.pipe.foreach(callback)
+        return self
+
     def flatmap(self, callback):
         self.pipe.flatmap(callback)
         return self
@@ -216,3 +242,9 @@ def run(self, arg=None, convertToStr=False, collect=True):
         res, errs = results
         res = [cloudpickle.loads(record) for record in res]
         return res, errs
+
+    def register(self, arg=None):
+        self.pipe.register(arg)
+        selfBytes = cloudpickle.dumps(self.pipe)
+        res = self.r.execute_command('RG.PYEXECUTEREMOTE', selfBytes)
+        return res