Rev 6: Implement the core of advertise_branch, allowing revision messages to be emitted. in file:///home/robertc/source/baz/plugins/dbus/trunk/
Robert Collins
robertc at robertcollins.net
Tue Feb 6 07:37:02 GMT 2007
------------------------------------------------------------
revno: 6
revision-id: robertc at robertcollins.net-20070206073700-2vi6qnfqfchehdzn
parent: robertc at robertcollins.net-20070206071242-f1htyq1f1utkj77a
committer: Robert Collins <robertc at robertcollins.net>
branch nick: trunk
timestamp: Tue 2007-02-06 18:37:00 +1100
message:
Implement the core of advertise_branch, allowing revision messages to be emitted.
modified:
TODO todo-20070206032729-2b5teqiwctqrfgei-3
activity.py activity.py-20070206034725-q208d0jtkshwu0fx-1
tests/test_activity.py test_activity.py-20070206035206-bnhzvtm6m6hpgylz-1
=== modified file 'TODO'
--- a/TODO 2007-02-06 07:12:42 +0000
+++ b/TODO 2007-02-06 07:37:00 +0000
@@ -1,8 +1,6 @@
Plans for dbus and bzr
----------------------
- * advertise_branch sends the message 'announce_revision' when the service is
- running
* simple subscriber that sits in a corner and shows revisions
* dbus service to act as the broadcast point for revisions
* .service details for the dbus service.
=== modified file 'activity.py'
--- a/activity.py 2007-02-06 07:12:42 +0000
+++ b/activity.py 2007-02-06 07:37:00 +0000
@@ -26,6 +26,8 @@
import dbus.service
import gobject
+from bzrlib.revision import NULL_REVISION
+
class Activity(object):
"""bzrlib object activity tracking."""
@@ -54,6 +56,44 @@
:return: None
:raises: Nothing should be raised.
"""
+ self.announce_revision(branch.last_revision(), branch.base)
+
+ def announce_revision(self, revision, url):
+ """Low level revision-specific announce logic.
+
+ The recommended API is advertise_branch, announce_revision, while
+ public is not stable or supported. Use at your own warranty.
+ """
+ try:
+ dbus_object = self.bus.get_object(Broadcast.DBUS_NAME,
+ Broadcast.DBUS_PATH)
+ except dbus.DBusException, e:
+ if e.message.startswith(
+ 'org.freedesktop.DBus.Error.ServiceUnknown:'):
+ # service not available
+ return
+ else:
+ # some other error
+ raise
+ dbus_iface = dbus.Interface(dbus_object, Broadcast.DBUS_INTERFACE)
+ # make a non-blocking call, which we can then ignore as we dont
+ # care about responses: Apparently there is some dbus foo to help
+ # make this not need the stub function
+ def handle_reply():
+ """Ignore all replies: we want nothing back."""
+ def handle_error(error):
+ """If an error has happened, lets raise it.
+
+ Note that this will not raise it at the right point, but it should
+ hit some handler somewhere.
+ """
+ raise error
+ if revision in (None, NULL_REVISION):
+ revision = '' # avoid sending None or NULL_REVISION (its internal
+ # only) on the wire.
+ dbus_iface.announce_revision(revision, url,
+ reply_handler=handle_reply,
+ error_handler=handle_error)
def serve_agency(self):
"""Serve an advertising agency.
=== modified file 'tests/test_activity.py'
--- a/tests/test_activity.py 2007-02-06 07:12:42 +0000
+++ b/tests/test_activity.py 2007-02-06 07:37:00 +0000
@@ -134,7 +134,8 @@
# should not error: just call it
activity.Activity(bus=self.bus).advertise_branch(self.make_branch('.'))
- def in_dev_test_advertise_branch_service_running(self):
+ def test_advertise_branch_service_running_no_commits(self):
+ """No commits in a branch leads to it being show with a revid of ''."""
# We could construct a test-broadcaster to test with, but for now the
# sheer overhead of that scares me.
# attach a Broadcaster to our test branch: creates a running service.
@@ -146,6 +147,7 @@
branches = []
def catch_branch(revision, url):
branches.append((revision, url))
+ gobject.idle_add(mainloop.quit)
dbus_object.connect_to_signal("Revision", catch_branch,
dbus_interface=activity.Broadcast.DBUS_INTERFACE)
@@ -155,10 +157,42 @@
# now, let the broadcast method interactions all happen
mainloop = gobject.MainLoop()
- gobject.idle_add(mainloop.quit)
- mainloop.run()
- self.assertEqual([('revision1', 'url1'), ('revision2', 'url2')],
- branches)
+ mainloop.run()
+ # '' because the branch has no commits.
+ self.assertEqual([('', branch.base)], branches)
+
+ def test_advertise_branch_service_running_with_commits(self):
+ """With commits, the commit is utf8 encoded."""
+ # We could construct a test-broadcaster to test with, but for now the
+ # sheer overhead of that scares me.
+ # attach a Broadcaster to our test branch: creates a running service.
+ broadcaster = activity.Broadcast(self.bus)
+ # get the object so we can subscribe to callbacks from it.
+ dbus_object = self.bus.get_object(activity.Broadcast.DBUS_NAME,
+ activity.Broadcast.DBUS_PATH)
+ # we want to recieve the callbacks to inspect them.
+ branches = []
+ def catch_branch(revision, url):
+ branches.append((revision, url))
+ gobject.idle_add(mainloop.quit)
+ dbus_object.connect_to_signal("Revision", catch_branch,
+ dbus_interface=activity.Broadcast.DBUS_INTERFACE)
+
+ # now call our convenience JustDoIt method.
+ tree = self.make_branch_and_memory_tree('.')
+ tree.lock_write()
+ tree.add('')
+ tree.commit('commit', rev_id=u'\xc8')
+ tree.unlock()
+ activity.Activity(bus=self.bus).advertise_branch(tree.branch)
+ # now, let the broadcast method interactions all happen
+
+ mainloop = gobject.MainLoop()
+ mainloop.run()
+ # in theory, the revid here would be utf8 encoded, but dbus seems to
+ # consider 'string' == 'unicode', and 'Magic happens' as far as wire
+ # level decoding -> unicode results. CRY.
+ self.assertEqual([(u'\xc8', tree.branch.base)], branches)
def test_constructor(self):
"""The constructor should setup a good working environment."""
@@ -200,6 +234,8 @@
# now try to call the announce method, which should call the signal
# handlers - all of them..
+ # we dont use announce_revision here because we want to catch errors
+ # should they occur.
errors = []
results = []
def handle_reply():
More information about the bazaar-commits
mailing list