python: Add WatchImplementation class

Problem: The KVSWatchFuture and JobEventWatchFuture classes are
identical except for the get() and cancel() calls to get a value
or cancel the stream.

Solution: Create a new WatchImplementation class that can
act as an abstract class for both watch futures.  The futures both
need only implement their respective "get" and "cancel" implementations.

Update KVSWatchFuture and JobEventWatchFuture to be based on WatchImplementation.

Author: chu11

src/bindings/python/flux/job/event.py

@@ -11,8 +11,8 @@
 import json
 
 from _flux._core import ffi
-from flux.future import Future
 from flux.job._wrapper import _RAW as RAW
+from flux.kvs import WatchImplementation
 
 # Names of events that may appear in the main eventlog (i.e. ``eventlog="eventlog"``)
 # See Flux RFC 21 for documentation on each event.
@@ -74,23 +74,32 @@ def context_string(self):
         )
 
 
-class JobEventWatchFuture(Future):
+class JobEventWatchFuture(WatchImplementation):
     """
     A future returned from job.event_watch_async().
     Adds get_event() method to return an EventLogEntry event
     """
 
-    def __del__(self):
-        if self.needs_cancel is not False:
-            self.cancel()
-        try:
-            super().__del__()
-        except AttributeError:
-            pass
-
     def __init__(self, future_handle):
         super().__init__(future_handle)
-        self.needs_cancel = True
+
+    def watch_get(self, future):
+        """
+        Implementation of watch_get() for JobEventWatchFuture.
+
+        Will be called from WatchABC.get()
+        """
+        result = ffi.new("char *[1]")
+        RAW.event_watch_get(future, result)
+        return EventLogEvent(ffi.string(result[0]).decode("utf-8"))
+
+    def watch_cancel(self, future):
+        """
+        Implementation of watch_cancel() for JobEventWatchFuture.
+
+        Will be called from WatchABC.cancel()
+        """
+        RAW.event_watch_cancel(future)
 
     def get_event(self, autoreset=True):
         """
@@ -101,41 +110,7 @@ def get_event(self, autoreset=True):
         call to get_event() will try to fetch the next event and thus
         may block.
         """
-        result = ffi.new("char *[1]")
-        try:
-            #  Block until Future is ready:
-            self.wait_for()
-            RAW.event_watch_get(self.pimpl, result)
-        except OSError as exc:
-            if exc.errno == errno.ENODATA:
-                self.needs_cancel = False
-                return None
-            # raise handle exception if there is one
-            self.raise_if_handle_exception()
-            # re-raise all other exceptions
-            #
-            # Note: overwrite generic OSError strerror string with the
-            # EventWatch future error string to give the caller appropriate
-            # detail (e.g. instead of "No such file or directory" use
-            # "job <jobid> does not exist"
-            #
-            exc.strerror = self.error_string()
-            raise
-        event = EventLogEvent(ffi.string(result[0]).decode("utf-8"))
-        if autoreset is True:
-            self.reset()
-        return event
-
-    def cancel(self, stop=False):
-        """Cancel a streaming job.event_watch_async() future
-
-        If stop=True, then deactivate the multi-response future so no
-        further callbacks are called.
-        """
-        RAW.event_watch_cancel(self.pimpl)
-        self.needs_cancel = False
-        if stop:
-            self.stop()
+        return self.get(autoreset=autoreset)
 
 
 def event_watch_async(flux_handle, jobid, eventlog="eventlog"):

src/bindings/python/flux/kvs.py

@@ -12,6 +12,7 @@
 import errno
 import json
 import os
+from abc import ABC, abstractmethod
 from typing import Any, Mapping
 
 import flux.constants
@@ -781,9 +782,11 @@ def walk(directory, topdown=False, flux_handle=None, namespace=None):
     return _inner_walk(directory, "", topdown, namespace=namespace)
 
 
-class KVSWatchFuture(Future):
+class WatchImplementation(Future, ABC):
     """
-    A future returned from kvs_watch_async().
+    Interface for KVS based watchers
+
+    Users to implement watch_get() and watch_cancel() functions.
     """
 
     def __del__(self):
@@ -798,20 +801,26 @@ def __init__(self, future_handle):
         super().__init__(future_handle)
         self.needs_cancel = True
 
+    @abstractmethod
+    def watch_get(self, future):
+        pass
+
+    @abstractmethod
+    def watch_cancel(self, future):
+        pass
+
     def get(self, autoreset=True):
         """
-        Return the new value of the KVS key or None if the stream has
-        terminated.
+        Return the new value or None if the stream has terminated.
 
         The future is auto-reset unless autoreset=False, so a subsequent
         call to get() will try to fetch the next value and thus
         may block.
         """
-        valp = ffi.new("char *[1]")
         try:
             #  Block until Future is ready:
             self.wait_for()
-            RAW.flux_kvs_lookup_get(self.pimpl, valp)
+            ret = self.watch_get(self.pimpl)
         except OSError as exc:
             if exc.errno == errno.ENODATA:
                 self.needs_cancel = False
@@ -827,23 +836,49 @@ def get(self, autoreset=True):
             #
             exc.strerror = self.error_string()
             raise
-        val = _get_value(valp)
         if autoreset is True:
             self.reset()
-        return val
+        return ret
 
     def cancel(self, stop=False):
-        """Cancel a streaming kvs_watch_async() future
+        """Cancel a streaming future
 
         If stop=True, then deactivate the multi-response future so no
         further callbacks are called.
         """
-        RAW.flux_kvs_lookup_cancel(self.pimpl)
+        self.watch_cancel(self.pimpl)
         self.needs_cancel = False
         if stop:
             self.stop()
 
 
+class KVSWatchFuture(WatchImplementation):
+    """
+    A future returned from kvs_watch_async().
+    """
+
+    def __init__(self, future_handle):
+        super().__init__(future_handle)
+
+    def watch_get(self, future):
+        """
+        Implementation of watch_get() for KVSWatchFuture.
+
+        Will be called from WatchABC.get()
+        """
+        valp = ffi.new("char *[1]")
+        RAW.flux_kvs_lookup_get(future, valp)
+        return _get_value(valp)
+
+    def watch_cancel(self, future):
+        """
+        Implementation of watch_cancel() for KVSWatchFuture.
+
+        Will be called from WatchABC.cancel()
+        """
+        RAW.flux_kvs_lookup_cancel(future)
+
+
 def kvs_watch_async(
     flux_handle, key, namespace=None, waitcreate=False, uniq=False, full=False
 ):