Last active
December 14, 2015 16:38
-
-
Save tito/5116183 to your computer and use it in GitHub Desktop.
You don't like callback programming? Use asynchronous!
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from kivy.app import App | |
from kivy.uix.button import Button | |
from kivy.uix.gridlayout import GridLayout | |
from kivy.clock import Clock | |
from kivy.network.urlrequest import UrlRequest | |
class _async_generator(object): | |
_calls = [] | |
def __init__(self, func, args, callback): | |
object.__init__(self) | |
_async_generator._calls.append(self) | |
self.func = func | |
self.args = args | |
self.callback = callback | |
self.gen = func(*args) | |
self.pump() | |
def stop(self): | |
try: | |
if self.gen: | |
self.gen.send(True) | |
except StopIteration: | |
pass | |
self._end() | |
def pump(self, *args): | |
gen = self.gen | |
if gen is not None: | |
try: | |
ret = gen.next() | |
if ret: | |
Clock.schedule_once(self.pump, ret) | |
return | |
except StopIteration: | |
pass | |
self._end() | |
def _end(self): | |
if self not in _async_generator._calls: | |
return | |
if self.callback: | |
self.callback(*self.args) | |
_async_generator._calls.remove(self) | |
Clock.unschedule(self.pump) | |
class asynchronous(object): | |
def __init__(self, callback=None): | |
super(asynchronous, self).__init__() | |
self.callback = callback | |
def __call__(self, func): | |
def wrapped_f(*args): | |
return _async_generator(func, args, self.callback) | |
return wrapped_f | |
class TestApp(App): | |
# | |
# example with callback | |
# | |
def add_many_finished(self, *args): | |
print 'end!', self, args | |
@asynchronous(add_many_finished) | |
def add_many(self, *args): | |
for x in xrange(5): | |
self.root.add_widget(Button(text='Label {}'.format(x))) | |
# wait one second. If we received anything from the yield, we need | |
# to stop (maybe the user asked us to stop.) | |
if (yield 1): return | |
# | |
# example with urlrequest, no callback | |
# | |
@asynchronous() | |
def sequential_call(self, *args): | |
req = UrlRequest('https://api.twitter.com/1/trends/1.json') | |
while not req.is_finished: | |
yield .1 | |
print req.result | |
# | |
# example with stop() | |
# | |
def call_add_many(self, *args): | |
if self.gen: | |
self.gen.stop() | |
self.gen = self.add_many() | |
def stop_add_many(self, *args): | |
self.gen.stop() | |
# | |
# app | |
# | |
def build(self): | |
self.gen = None | |
self.root = root = GridLayout(cols=2) | |
btn = Button(text='Create 5 buttons (unstoppable)') | |
btn.bind(on_release=self.add_many) | |
root.add_widget(btn) | |
btn = Button(text='Get trends (urlrequest)') | |
btn.bind(on_release=self.sequential_call) | |
root.add_widget(btn) | |
btn = Button(text='Create 5 buttons (stoppable)') | |
btn.bind(on_release=self.call_add_many) | |
root.add_widget(btn) | |
btn = Button(text='Stop generation') | |
btn.bind(on_release=self.stop_add_many) | |
root.add_widget(btn) | |
return root | |
TestApp().run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Line 32 needs to be
if ret == None:
otherwise it's going to skip it when ret == 0 because bool(0) = False.