Skip to content

Instantly share code, notes, and snippets.

@ejconlon
Created February 27, 2011 03:26
Show Gist options
  • Save ejconlon/845877 to your computer and use it in GitHub Desktop.
Save ejconlon/845877 to your computer and use it in GitHub Desktop.
Translated from the example in "Deprecating the Observer Pattern" (Maier, Rompf, Odersky)
#!/usr/bin/env python
"""
Translated from the example in
Deprecating the Observer Pattern (Maier, Rompf, Odersky)
http://lamp.epfl.ch/~imaier/pub/DeprecatingObserversTR2010.pdf
Their Scala example:
Reactor.once { self =>
// step 1:
val path = new Path((self next mouseDown).position)
// step 2:
self loopUntil mouseUp {
val m = self next mouseMove
path.lineTo(m.position)
draw(path)
}
// step 3:
path.close()
draw(path)
}
"""
import logging
def accept_line():
logger = logging.getLogger("accept_line")
logger.debug("waiting for mouse down")
mouseDownEvent = (yield ("pop", "mouse_down"))
logger.debug("got mouse down")
path = Path(mouseDownEvent["position"])
logger.debug("continuing until mouse up")
while (yield ("until", "mouse_up")):
logger.debug("continue, not a mouse up")
mouseMoveEvent = (yield ("pop", "mouse_move"))
logger.debug("got mouse move")
path.lineTo(mouseMoveEvent["position"])
path.draw()
logger.debug("continuing until mouse up")
path.close()
path.draw()
logger.debug("done")
def main():
logging.basicConfig(level=logging.DEBUG)
reactor = Reactor(accept_line)
events = [
make_mouse_move_event(0),
make_mouse_down_event(1),
make_mouse_move_event(2),
make_key_press_event('a'),
make_mouse_move_event(3),
make_mouse_move_event(4),
make_mouse_up_event(5),
make_mouse_down_event(6)
]
for event in events:
reactor.send(event)
class Reactor(object):
def __init__(self, closure):
self.generator = closure()
self.stopped = False
self.filter_type = None
self.event_type = None
self.logger = logging.getLogger("Reactor")
self._advance_filter()
def _advance_filter(self):
if not self.stopped:
try:
self.filter_type, self.event_type = self.generator.next()
self.logger.debug("filter %s %s", self.filter_type, self.event_type)
except StopIteration:
self.stopped = True
self.logger.debug("stopping")
def _proxy_send(self, value):
if not self.stopped:
try:
self.logger.debug("sending %s", value)
self.filter_type, self.event_type = self.generator.send(value)
self.logger.debug("filter %s %s", self.filter_type, self.event_type)
except StopIteration:
self.stopped = True
self.logger.debug("stopping")
def send(self, event):
self.logger.debug("event %s", event)
if self.stopped: return
is_match = (event["type"] == self.event_type)
if "peek" == self.filter_type:
self._proxy_send(is_match)
if is_match:
self.send(event)
elif "until" == self.filter_type:
self._proxy_send(not is_match)
if not is_match:
self.send(event)
elif is_match and "pop" == self.filter_type:
self._proxy_send(event["value"])
class Path(object):
def __init__(self, x0):
self.xs = [x0]
self.logger = logging.getLogger("Path")
def lineTo(self, xi):
self.xs.append(xi)
def draw(self):
self.logger.debug("draw: xs = %s", self.xs)
def close(self):
self.logger.debug("closed")
def make_mouse_down_event(position):
return {"type": "mouse_down", "value" : {"position" : position}}
def make_mouse_move_event(position):
return {"type": "mouse_move", "value" : {"position" : position}}
def make_mouse_up_event(position):
return {"type": "mouse_up", "value" : {"position" : position}}
def make_key_press_event(key):
return {"type": "key_press", "value" : {"key" : key}}
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment