Rewrite of the network stuff. Stuff broke likely.

chat/chat.py

@@ -14,6 +14,8 @@ import sys
 try:
 	import activity
 	from Group import *
+	from StreamReader import *
+	from StreamWriter import *
 	from sugar_globals import *
 except ImportError:
 	from sugar import activity
@@ -139,7 +141,7 @@ class Chat(activity.Activity):
 		self._controller.notify_activate(self)
 
 	def recv_message(self, buddy, msg):
-		self._insert_rich_message(buddy.nick(), msg)
+		self._insert_rich_message(buddy.get_nick_name(), msg)
 		self._controller.notify_new_message(self, buddy)
 
 	def _insert_rich_message(self, nick, msg):
@@ -160,18 +162,18 @@ class Chat(activity.Activity):
 			aniter = buffer.get_end_iter()
 			buffer.insert(aniter, message)
 		else:
-			nick = p2p.Owner.get_instance().get_nick()
+			nick = self._group.get_owner().get_nick_name()
 			self._insert_rich_message(nick, text)
 
 class BuddyChat(Chat):
 	def __init__(self, controller, buddy):
 		self._buddy = buddy
-		self._act_name = "Chat: %s" % buddy.nick()
+		self._act_name = "Chat: %s" % buddy.get_nick_name()
 		Chat.__init__(self, controller)
 
 	def _start(self):
-		group = p2p.Group.get_instance()
+		service_name = buddy.get_service_name()
-		self._output_pipe = p2p.OutputPipe(group, self._buddy, "buddy-chat")
+		self._stream_writer = StreamWriter(self._group, service_name)
 
 	def activity_on_connected_to_shell(self):
 		Chat.activity_on_connected_to_shell(self)
@@ -185,7 +187,7 @@ class BuddyChat(Chat):
 
 	def send_message(self, text):
 		if len(text) > 0:
-			success = self._output_pipe.send(text)
+			success = self._stream_writer.write(text)
 			self._local_message(success, text)
 
 	def activity_on_close_from_user(self):
@@ -212,6 +214,15 @@ class GroupChat(Chat):
 		self._group.add_listener(self._on_group_event)
 		self._group.join()
 		
+		owner_service = self._group.get_owner().get_service_name()
+		self._buddy_reader = StreamReader(self._group, owner_service)
+		self._buddy_reader.set_listener(self._buddy_recv_message)
+
+		self._buddy_reader = StreamReader(self._group, "localgroup_multicast")
+		self._buddy_reader.set_listener(self.recv_message)
+		
+		self._stream_writer = StreamWriter(self._group, "localgroup_multicast")
+
 	def _create_sidebar(self):
 		vbox = gtk.VBox(False, 6)
 		
@@ -322,12 +333,12 @@ class GroupChat(Chat):
 
 	def send_message(self, text):
 		if len(text) > 0:
-			self._output_pipe.send(text)
+			self._stream_writer.write(text)
 		self._local_message(True, text)
 
 	def recv_message(self, buddy, msg):
 		if buddy:
-			self._insert_rich_message(buddy.nick(), msg)
+			self._insert_rich_message(buddy.get_nick_name(), msg)
 			self._controller.notify_new_message(self, None)
 
 	def _buddy_recv_message(self, sender, msg):

p2p/Buddy.py

@@ -1,7 +1,14 @@
 import pwd
 import os
 
-import Service
+from Service import *
+import presence
+
+BUDDY_SERVICE_TYPE = "_olpc_buddy._tcp"
+BUDDY_SERVICE_PORT = 666
+
+GROUP_SERVICE_TYPE = "_olpc_buddy._udp"
+GROUP_SERVICE_PORT = 6666
 
 class Buddy:
 	def __init__(self, service, nick_name):
@@ -11,22 +18,27 @@ class Buddy:
 	def get_service(self):
 		return self._service
 
+	def get_service_name(self):
+		return self._service.get_name()
+		
 	def get_nick_name(self):
 		return self._nick_name
 		
 class Owner(Buddy):
-	instance = None
-
 	def __init__(self):
 		ent = pwd.getpwuid(os.getuid())
 		nick = ent[0]
 		if not nick or not len(nick):
 			nick = "n00b"
-		Buddy.__init__(self, None, nick)
 
-	def get_instance():
+		service_name = nick + '.' + GROUP_SERVICE_TYPE
-		if not Owner.instance:
+		service = Service(service_name, '', '', GROUP_SERVICE_PORT)
-			Owner.instance = Owner()
-		return Owner.instance
 
-	get_instance = staticmethod(get_instance)
+		Buddy.__init__(self, service, nick)
+		
+	def register(self):
+		pannounce = presence.PresenceAnnounce()
+		pannounce.register_service(self._nick_name,
+								   BUDDY_SERVICE_PORT,
+								   BUDDY_SERVICE_TYPE,
+								   nickname = self._nick_name)

p2p/Group.py

@@ -7,9 +7,6 @@ from Service import *
 BUDDY_JOIN = "join"
 BUDDY_LEAVE = "leave"
 
-BUDDY_SERVICE_TYPE = "_olpc_buddy._tcp"
-BUDDY_SERVICE_PORT = 666
-
 class Group:
 	def __init__(self):
 		self._listeners = []
@@ -39,11 +36,26 @@ class LocalGroup(Group):
 		self._pdiscovery.add_service_listener(self._on_service_change)
 		self._pdiscovery.start()
 
+	def get_owner(self):
+		return self._owner
+
 	def join(self):
-		self._pannounce = presence.PresenceAnnounce()
+		self._owner = Owner()
-		name = Owner.get_instance().get_nick_name()
+		self._owner.register()
-		self._pannounce.register_service(name, BUDDY_SERVICE_PORT, BUDDY_SERVICE_TYPE,
+
-										 nickname = name)
+	def get_service_from_name(self, name):
+		if name == 'localgroup_multicast':
+			return Service('localgroup_multicast', '', '224.0.0.221', 6666, True)
+		elif name == self._owner.get_service().get_name():
+			return self._owner.get_service()
+		else:
+			return self._services[name]
+	
+	def get_buddy_from_address(self, address):
+		for buddy in self._buddies.values():
+			if buddy.get_service().get_address() == address:
+				return buddy
+		return None
 
 	def _on_service_change(self, action, interface, protocol, name, stype, domain, flags):
 		if action == presence.ACTION_SERVICE_NEW:
@@ -58,12 +70,12 @@ class LocalGroup(Group):
 		service = Service(name, host, address, port)
 		self._services[name] = service
 		if stype == BUDDY_SERVICE_TYPE:
-			self._add_buddy(service, txt)
+			data = self._pair_to_dict(avahi.txt_array_to_string_array(txt))
+			self._add_buddy(service, data)
 
-	def _add_buddy(self, service, txt):
+	def _add_buddy(self, service, data):
 		name = service.get_name()
 		if not self._buddies.has_key(name):
-			data = self._pair_to_dict(avahi.txt_array_to_string_array(txt))
 			buddy = Buddy(service, data['nickname'])
 			self._buddies[name] = buddy
 			self._notify_buddy_join(buddy)

p2p/Service.py

@@ -1,9 +1,10 @@
 class Service(object):
-	def __init__(self, name, host, address, port):
+	def __init__(self, name, host, address, port, multicast=False):
 		self._name = name
 		self._host = host
 		self._address = str(address)
 		self._port = int(port)
+		self._multicast = multicast
 
 	def get_name(self):
 		return self._name
@@ -16,3 +17,6 @@ class Service(object):
 
 	def get_port(self):
 		return self._port
+		
+	def is_multicast(self):
+		return self._multicast

p2p/StreamReader.py

@@ -1,4 +1,41 @@
+from network import *
+
+class StreamReaderRequestHandler(object):
+	def __init__(self, reader):
+		self._reader = reader
+
+	def message(self, message):
+		address = network.get_authinfo()
+		self._reader.recv(address[0], message)
+		return True
+
 class StreamReader:
-	def __init__(service, callback):
+	def __init__(self, group, service_name):
-		self._service == service
+		self._group = group
+		self._service_name = service_name
+		
+		self._service = group.get_service_from_name(service_name)
+		if self._service.is_multicast():
+			self._setup_multicast()
+		else:
+			self._setup_unicast()
+
+	def set_listener(self, callback):
 		self._callback = callback
+
+	def _setup_multicast(self):
+		address = self._service.get_address()
+		port = self._service.get_port()
+		server = GroupServer(address, port, self._recv_multicast)
+		server.start()
+		
+	def _setup_unicast(self):
+		p2p_server = GlibXMLRPCServer(("", self._service.get_port()))
+		p2p_server.register_instance(StreamReaderRequestHandler(self))
+		
+	def _recv_multicast(self, msg):
+		self._recv(msg['addr'], msg['data'])
+	
+	def _recv(self, address, data):
+		buddy = self._group.get_buddy_from_address(address)
+		self._callback(buddy, data)

p2p/StreamWriter.py

@@ -1,6 +1,40 @@
+import socket
+
+import network
+
 class StreamWriter:
-	def __init__(service):
+	def __init__(self, group, service_name):
-		self._service = service
+		self._group = group
+		self._service_name = service_name
+		self._service = group.get_service_from_name(service_name)
+		self._address = self._service.get_address()
+		self._port = self._service.get_port()
+
+		if self._service.is_multicast():
+			self._setup_multicast()
+		else:
+			self._setup_unicast()
 		
 	def write(self, data):
-		pass
+		if self._service.is_multicast():
+			self._multicast_write(data)
+		else:
+			self._unicast_write(data)
+
+	def _setup_unicast(self):
+		xmlrpc_addr = "http://%s:%d" % (self._address, self._port)
+		self._uclient = xmlrpclib.ServerProxy(xmlrpc_addr)
+
+	def _unicast_write(self, data):
+		try:
+			self._uclient.message(data)
+			return True
+		except (socket.error, xmlrpclib.Fault, xmlrpclib.ProtocolError), e:
+			traceback.print_exc()
+			return False
+
+	def _setup_multicast(self):
+		self._mclient = network.GroupClient(self._address, self._port)
+		
+	def _multicast_write(self, data):
+		self._mclient.send_msg(data)

p2p/network.py

@@ -12,7 +12,6 @@ import gobject
 import SimpleXMLRPCServer
 import SocketServer
 
-
 __authinfos = {}
 
 def _add_authinfo(authinfo):
@@ -117,8 +116,7 @@ class GlibXMLRPCServer(GlibTCPServer, SimpleXMLRPCServer.SimpleXMLRPCDispatcher)
 
 		return response
 
-
+class GroupServer(object):
-class GroupChatController(object):
 
 	_MAX_MSG_SIZE = 500
 
@@ -127,14 +125,8 @@ class GroupChatController(object):
 		self._port = port
 		self._data_cb = data_cb
 
-		self._setup_sender()
 		self._setup_listener()
 
-	def _setup_sender(self):
-		self._send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-		# Make the socket multicast-aware, and set TTL.
-		self._send_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20) # Change TTL (=20) to suit
-
 	def _setup_listener(self):
 		# Listener socket
 		self._listen_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -168,6 +160,17 @@ class GroupChatController(object):
 			self._data_cb(msg)
 		return True
 
+class GroupClient(object):
+
+	_MAX_MSG_SIZE = 500
+
+	def __init__(self, address, port):
+		self._address = address
+		self._port = port
+
+		self._send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+		# Make the socket multicast-aware, and set TTL.
+		self._send_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20) # Change TTL (=20) to suit
+
 	def send_msg(self, data):
 		self._send_sock.sendto(data, (self._address, self._port))
-