Skip to content

Instantly share code, notes, and snippets.

@tito
Last active December 14, 2015 16:38
Show Gist options
  • Save tito/5116183 to your computer and use it in GitHub Desktop.
Save tito/5116183 to your computer and use it in GitHub Desktop.
You don't like callback programming? Use asynchronous!
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()
@depau
Copy link

depau commented Jan 14, 2015

Line 32 needs to be

if ret == None:

otherwise it's going to skip it when ret == 0 because bool(0) = False.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment