@@ -0,0 +1,2847 @@
+# -*- coding: utf-8 -*-%0A%0A#    Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.%0A#%0A#    Licensed under the Apache License, Version 2.0 (the %22License%22); you may%0A#    not use this file except in compliance with the License. You may obtain%0A#    a copy of the License at%0A#%0A#         http://www.apache.org/licenses/LICENSE-2.0%0A#%0A#    Unless required by applicable law or agreed to in writing, software%0A#    distributed under the License is distributed on an %22AS IS%22 BASIS, WITHOUT%0A#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the%0A#    License for the specific language governing permissions and limitations%0A#    under the License.%0A%0Aimport logging%0Aimport os%0Aimport sys%0A%0Afrom concurrent import futures%0A%0Alogging.basicConfig(level=logging.ERROR)%0A%0Aself_dir = os.path.abspath(os.path.dirname(__file__))%0Atop_dir = os.path.abspath(os.path.join(os.path.dirname(__file__),%0A                                       os.pardir,%0A                                       os.pardir))%0Asys.path.insert(0, top_dir)%0Asys.path.insert(0, self_dir)%0A%0A# INTRO: in this example linear_flow we will attach a listener to an engine%0A# and delay the return from a function until after the result of a task has%0A# occured in that engine. The engine will continue running (in the background)%0A# while the function will have returned.%0A%0Aimport taskflow.engines%0A%0Afrom taskflow.listeners import base%0Afrom taskflow.patterns import linear_flow as lf%0Afrom taskflow import states%0Afrom taskflow import task%0Afrom taskflow.utils import misc%0A%0A%0Aclass PokeFutureListener(base.ListenerBase):%0A    def __init__(self, engine, future, task_name):%0A        super(PokeFutureListener, self).__init__(%0A            engine,%0A            task_listen_for=(misc.Notifier.ANY,),%0A            flow_listen_for=%5B%5D)%0A        self._future = future%0A        self._task_name = task_name%0A%0A    def _task_receiver(self, state, details):%0A        if state in (states.SUCCESS, states.FAILURE):%0A            if details.get('task_name') == self._task_name:%0A                if state == states.SUCCESS:%0A                    self._future.set_result(details%5B'result'%5D)%0A                else:%0A                    failure = details%5B'result'%5D%0A                    self._future.set_exception(failure.exception)%0A%0A%0Aclass Hi(task.Task):%0A    def execute(self):%0A        # raise IOError(%22I broken%22)%0A        return 'hi'%0A%0A%0Aclass Bye(task.Task):%0A    def execute(self):%0A        return 'bye'%0A%0A%0Adef return_from_flow(pool):%0A    wf = lf.Flow(%22root%22).add(Hi(%22hi%22), Bye(%22bye%22))%0A    eng = taskflow.engines.load(wf, engine_conf='serial')%0A    f = futures.Future()%0A    watcher = PokeFutureListener(eng, f, 'hi')%0A    watcher.register()%0A    pool.submit(eng.run)%0A    return (eng, f.result())%0A%0A%0Awith futures.ThreadPoolExecutor(1) as pool:%0A    engine, hi_result = return_from_flow(pool)%0A    print(hi_result)%0A%0Aprint(engine.storage.get_flow_state())%0A

Add a example that activates a future when a result is ready