- Refactor service handling. Buddies can now have more than one service associated

2006-05-22 01:21:12 -04:00
parent cbae21b487
commit 3b2f185e5c
4 changed files with 158 additions and 39 deletions
@@ -2,38 +2,63 @@ import pwd
 import os

 from Service import Service
+from sugar import env

 PRESENCE_SERVICE_TYPE = "_olpc_presence._tcp"
 PRESENCE_SERVICE_PORT = 6000

-class BuddyBase:
-	def __init__(self, service, nick_name):
-		self._service = service
-		self._nick_name = nick_name
+__buddy_service_types = [PRESENCE_SERVICE_TYPE]
+
+def recognize_buddy_service_type(stype):
+	if stype not in __buddy_service_types:
+		__buddy_service_types.append(stype)
+
+def get_recognized_buddy_service_types():
+	return __buddy_service_types[:]
+
+
+class Buddy(object):
+	def __init__(self, service):
+		self._services = {}
+		self._services[service.get_type()] = service
+		self._nick_name = service.get_name()
+		self._address = service.get_address()

 	def get_icon(self):
 		"""Return the buddies icon, if any."""
 		return self._icon
 		
 	def get_address(self):
-		return self._service.get_address()
+		return self._address

-	def get_service_name(self):
-		return self._service.get_name()
+	def add_service(self, service):
+		if service.get_name() != self._nick_name:
+			return False
+		if service.get_address() != self._address:
+			return False
+		if self._services.has_key(service.get_type()):
+			return False
+		self._services[service.get_type()] = service
+
+	def remove_service(self, stype):
+		if self._services.has_key(stype):
+			del self._services[stype]
+
+	def get_service(self, stype):
+		if self._services.has_key(stype):
+			return self._services[stype]
+		return None
 		
 	def get_nick_name(self):
 		return self._nick_name
-		
-class Buddy(BuddyBase):
-	"""Normal buddy class."""

 	def set_icon(self, icon):
 		"""Can only set icon for other buddies.  The Owner
 		takes care of setting it's own icon."""
 		self._icon = icon
-	
+		

-class Owner(BuddyBase):
+class Owner(Buddy):
 	"""Class representing the owner of this machine/instance."""
 	def __init__(self, group):
 		self._group = group
@@ -42,19 +67,20 @@ class Owner(BuddyBase):
 		if not nick or not len(nick):
 			nick = "n00b"

-		service = Service(nick, PRESENCE_SERVICE_TYPE, PRESENCE_SERVICE_PORT)
-		BuddyBase.__init__(self, service, nick)
+		self._presence_service = Service(nick, PRESENCE_SERVICE_TYPE, PRESENCE_SERVICE_PORT)
+		Buddy.__init__(self, self._presence_service)

-		sugar_dir = os.path.abspath(os.path.expanduser("~/.sugar"))
-		icon = None
-		for fname in os.listdir(sugar_dir):
+		for fname in os.listdir(env.get_user_dir()):
 			if not fname.startswith("buddy-icon."):
 				continue
-			fd = open(os.path.join(sugar_dir, fname), "r")
+			fd = open(os.path.join(env.get_user_dir(), fname), "r")
 			self._icon = fd.read()
 			fd.close()
 			break

+	def set_icon(self, icon):
+		"""Can only set icon in constructor for now."""
+		pass
 		
 	def register(self):
-		self._service.register(self._group)
+		self._presence_service.register(self._group)
@@ -1,6 +1,7 @@
 import avahi

 from Buddy import Buddy
+from Buddy import get_recognized_buddy_service_types
 from Buddy import Owner
 from Buddy import PRESENCE_SERVICE_TYPE
 from Service import Service
@@ -105,8 +106,14 @@ class LocalGroup(Group):
 			self._pdiscovery.resolve_service(interface, protocol, name, stype, domain,
 											 self._on_service_resolved)
 		elif action == presence.ACTION_SERVICE_REMOVED:
-			if stype == PRESENCE_SERVICE_TYPE:
-				self._remove_buddy(name)
+			if stype in get_recognized_buddy_service_types():
+				buddy = self.get_buddy(name)
+				if buddy:
+					buddy.remove_service(stype)
+				# Removal of the presence service removes the buddy too
+				if stype == PRESENCE_SERVICE_TYPE:
+					self._remove_buddy(name)
+				self.remove_service((name, stype))
 			elif stype.startswith(_OLPC_SERVICE_TYPE_PREFIX):
 				self.remove_service((name, stype))
 						
@@ -118,9 +125,20 @@ class LocalGroup(Group):
 		for prop in avahi.txt_array_to_string_array(txt):
 			(key, value) = prop.split('=')
 			if key == 'group_address':
-				service.set_group_address(value)		
+				service.set_group_address(value)

-		if stype == PRESENCE_SERVICE_TYPE:
-			self._add_buddy(Buddy(service, name))
-		elif stype.startswith(_OLPC_SERVICE_TYPE_PREFIX):
+		# print "ServiceResolved: name=%s, stype=%s, port=%s, address=%s" % (name, stype, port, address)
+		if stype in get_recognized_buddy_service_types():
+			# Service recognized as Buddy services either create a new
+			# buddy if one doesn't exist yet, or get added to the existing
+			# buddy
+			buddy = self.get_buddy(name)
+			if buddy:
+				buddy.add_service(service)
+			else:
+				self._add_buddy(Buddy(service))
 			self.add_service(service)
+		elif stype.startswith(_OLPC_SERVICE_TYPE_PREFIX):
+			# These services aren't associated with buddies
+			self.add_service(service)
+
@@ -1,6 +1,12 @@
 import xmlrpclib
 import socket
 import traceback
+import threading
+
+import pygtk
+pygtk.require('2.0')
+import gobject
+

 import network
 from MostlyReliablePipe import MostlyReliablePipe
@@ -32,7 +38,7 @@ class Stream(object):
 				self._callback(self._group.get_buddy(nick_name), data)


-class UnicastStreamWriter(object):
+class UnicastStreamWriterBase(object):
 	def __init__(self, stream, service, owner_nick_name):
 		# set up the writer
 		if not service:
@@ -42,6 +48,10 @@ class UnicastStreamWriter(object):
 		self._address = self._service.get_address()
 		self._port = self._service.get_port()
 		self._xmlrpc_addr = "http://%s:%d" % (self._address, self._port)
+
+class UnicastStreamWriter(UnicastStreamWriterBase):
+	def __init__(self, stream, service, owner_nick_name):
+		UnicastStreamWriterBase.__init__(self, stream, service, owner_nick_name)
 		self._writer = xmlrpclib.ServerProxy(self._xmlrpc_addr)

 	def write(self, data):
@@ -63,6 +73,60 @@ class UnicastStreamWriter(object):
 		return None


+class ThreadedRequest(threading.Thread):
+	def __init__(self, controller, addr, method, response_cb, user_data, *args):
+		threading.Thread.__init__(self)
+		self._controller = controller
+		self._method = method
+		self._args = args
+		self._response_cb = response_cb
+		self._user_data = user_data
+		self._writer = xmlrpclib.ServerProxy(addr)
+
+	def run(self):
+		response = None
+		try:
+			method = getattr(self._writer, self._method)
+			response = method(*self._args)
+		except (socket.error, xmlrpclib.Fault, xmlrpclib.ProtocolError):
+			traceback.print_exc()
+		if self._response_cb:
+			gobject.idle_add(self._response_cb, response, self._user_data)
+		self._controller.notify_request_done(self)
+
+class ThreadedUnicastStreamWriter(UnicastStreamWriterBase):
+	def __init__(self, stream, service, owner_nick_name):
+		self._requests_lock = threading.Lock()
+		self._requests = []
+		UnicastStreamWriterBase.__init__(self, stream, service, owner_nick_name)
+
+	def _add_request(self, request):
+		self._requests_lock.acquire()
+		if not request in self._requests:
+			self._requests.append(request)
+		self._requests_lock.release()
+
+	def write(self, response_cb, user_data, data):
+		"""Write some data to the default endpoint of this pipe on the remote server."""
+		request = ThreadedRequest(self, self._xmlrpc_addr, "message", response_cb,
+				user_data, self._owner_nick_name, data)
+		self._add_request(request)
+		request.start()
+
+	def custom_request(self, method_name, response_cb, user_data, *args):
+		"""Call a custom XML-RPC method on the remote server."""
+		request = ThreadedRequest(self, self._xmlrpc_addr, method_name, response_cb,
+				user_data, *args)
+		self._add_request(request)
+		request.start()
+
+	def notify_request_done(self, request):
+		self._requests_lock.acquire()
+		if request in self._requests:
+			self._requests.remove(request)
+		self._requests_lock.release()
+
+
 class UnicastStream(Stream):
 	def __init__(self, service, group):
 		Stream.__init__(self, service, group)
@@ -94,8 +158,11 @@ class UnicastStream(Stream):
 			raise ValueError("Handler name 'message' is a reserved handler.")
 		self._reader.register_function(handler, name)

-	def new_writer(self, service):
-		return UnicastStreamWriter(self, service, self._owner_nick_name)
+	def new_writer(self, service, threaded=False):
+		if threaded:
+			return ThreadedUnicastStreamWriter(self, service, self._owner_nick_name)
+		else:
+			return UnicastStreamWriter(self, service, self._owner_nick_name)


 class MulticastStream(Stream):
@@ -115,5 +182,5 @@ class MulticastStream(Stream):
 		[ nick_name, data ] = data.split(" |**| ", 2)
 		self.recv(nick_name, data)

-	def new_writer(self, service=None):
+	def new_writer(self, service=None, threaded=False):
 		return self