updated for base64 encoding of hashes and values

diff --git a/actions.py b/actions.py
index c3cea32821a09eb49a1d680efd615695a2ff763f..58252943436afacbe5231cf71dbee87917a1f58e 100644 (file)
--- a/actions.py
+++ b/actions.py
@@ -46,14 +46,15 @@ class FindNode(ActionBase):
     """ find node action merits it's own class as it is a long running stateful process """
     def handleGotNodes(self, args):
        l, sender = args
     """ find node action merits it's own class as it is a long running stateful process """
     def handleGotNodes(self, args):
        l, sender = args

-       if self.finished or self.answered.has_key(sender['id']):
+       sender = Node().initWithDict(sender)
+       if self.finished or self.answered.has_key(sender.id):

            # a day late and a dollar short
            return
        self.outstanding = self.outstanding - 1
            # a day late and a dollar short
            return
        self.outstanding = self.outstanding - 1

-       self.answered[sender['id']] = 1
+       self.answered[sender.id] = 1

        for node in l:
        for node in l:

-           if not self.found.has_key(node['id']):
-               n = Node(node['id'], node['host'], node['port'])
+           n = Node().initWithDict(node)
+           if not self.found.has_key(n.id):

                self.found[n.id] = n
                self.table.insertNode(n)
        self.schedule()
                self.found[n.id] = n
                self.table.insertNode(n)
        self.schedule()


@@ -115,17 +116,18 @@ class GetValue(FindNode):
     """ get value task """
     def handleGotNodes(self, args):
        l, sender = args
     """ get value task """
     def handleGotNodes(self, args):
        l, sender = args

-       if self.finished or self.answered.has_key(sender['id']):
+       sender = Node().initWithDict(sender)
+       if self.finished or self.answered.has_key(sender.id):

            # a day late and a dollar short
            return
        self.outstanding = self.outstanding - 1
            # a day late and a dollar short
            return
        self.outstanding = self.outstanding - 1

-       self.answered[sender['id']] = 1
+       self.answered[sender.id] = 1

        # go through nodes
        # if we have any closer than what we already got, query them
        if l.has_key('nodes'):
            for node in l['nodes']:
        # go through nodes
        # if we have any closer than what we already got, query them
        if l.has_key('nodes'):
            for node in l['nodes']:

-               if not self.found.has_key(node['id']):
-                   n = Node(node['id'], node['host'], node['port'])
+               n = Node().initWithDict(node)
+               if not self.found.has_key(n.id):

                    self.found[n.id] = n
                    self.table.insertNode(n)
        elif l.has_key('values'):
                    self.found[n.id] = n
                    self.table.insertNode(n)
        elif l.has_key('values'):

diff --git a/khashmir.py b/khashmir.py
index 9e5a26cc17e7c13e592081e00bdf5fa26f8942a9..20cc062f0b9a2128ba1d358634666510329e7295 100644 (file)
--- a/khashmir.py
+++ b/khashmir.py
@@ -21,6 +21,8 @@ threadable.init()
 from bsddb3 import db ## find this at http://pybsddb.sf.net/
 from bsddb3._db import DBNotFoundError
 
 from bsddb3 import db ## find this at http://pybsddb.sf.net/
 from bsddb3._db import DBNotFoundError
 

+from base64 import decodestring as decode
+

 # don't ping unless it's been at least this many seconds since we've heard from a peer
 MAX_PING_INTERVAL = 60 * 15 # fifteen minutes
 
 # don't ping unless it's been at least this many seconds since we've heard from a peer
 MAX_PING_INTERVAL = 60 * 15 # fifteen minutes
 


@@ -30,7 +32,7 @@ MAX_PING_INTERVAL = 60 * 15 # fifteen minutes
 class Khashmir(xmlrpc.XMLRPC):
     __slots__ = ['listener', 'node', 'table', 'store', 'itime', 'kw', 'app']
     def __init__(self, host, port):
 class Khashmir(xmlrpc.XMLRPC):
     __slots__ = ['listener', 'node', 'table', 'store', 'itime', 'kw', 'app']
     def __init__(self, host, port):

-       self.node = Node(newID(), host, port)
+       self.node = Node().init(newID(), host, port)

        self.table = KTable(self.node)
        self.app = Application("xmlrpc")
        self.app.listenTCP(port, server.Site(self))
        self.table = KTable(self.node)
        self.app = Application("xmlrpc")
        self.app.listenTCP(port, server.Site(self))


@@ -67,7 +69,7 @@ class Khashmir(xmlrpc.XMLRPC):
        """
         ping this node and add the contact info to the table on pong!
        """
        """
         ping this node and add the contact info to the table on pong!
        """

-       n =Node(" "*20, host, port)  # note, we 
+       n =Node().init(" "*20, host, port)  # note, we 

        self.sendPing(n)
 
 
        self.sendPing(n)
 
 


@@ -90,8 +92,12 @@ class Khashmir(xmlrpc.XMLRPC):
     def valueForKey(self, key, callback):
        """ returns the values found for key in global table """
        nodes = self.table.findNodes(key)
     def valueForKey(self, key, callback):
        """ returns the values found for key in global table """
        nodes = self.table.findNodes(key)

+       # decode values, they will be base64 encoded
+       def cbwrap(values, cb=callback):
+           values = map(lambda x: decode(x), values)
+           callback(values)

        # create our search state
        # create our search state

-       state = GetValue(self, key, callback)
+       state = GetValue(self, key, cbwrap)

        reactor.callFromThread(state.goWithNodes, nodes)
 
 
        reactor.callFromThread(state.goWithNodes, nodes)
 
 


@@ -156,7 +162,9 @@ class Khashmir(xmlrpc.XMLRPC):
                pass
            else:
                #print "Got PONG from %s at %s:%s" % (`msg['id']`, t.target.host, t.target.port)
                pass
            else:
                #print "Got PONG from %s at %s:%s" % (`msg['id']`, t.target.host, t.target.port)

-               n = Node(sender['id'], host, port)
+               sender['host'] = host
+               sender['port'] = port
+               n = Node().initWithDict(sender)

                table.insertNode(n)
            return
        def _defaultPong(err):
                table.insertNode(n)
            return
        def _defaultPong(err):


@@ -199,7 +207,8 @@ class Khashmir(xmlrpc.XMLRPC):
            returns sender dict
        """
        ip = self.crequest.getClientIP()
            returns sender dict
        """
        ip = self.crequest.getClientIP()

-       n = Node(sender['id'], ip, sender['port'])
+       sender['host'] = ip
+       n = Node().initWithDict(sender)

        self.insertNode(n)
        return self.node.senderDict()
                
        self.insertNode(n)
        return self.node.senderDict()
                


@@ -207,11 +216,13 @@ class Khashmir(xmlrpc.XMLRPC):
        nodes = self.table.findNodes(target)
        nodes = map(lambda node: node.senderDict(), nodes)
        ip = self.crequest.getClientIP()
        nodes = self.table.findNodes(target)
        nodes = map(lambda node: node.senderDict(), nodes)
        ip = self.crequest.getClientIP()

-       n = Node(sender['id'], ip, sender['port'])
+       sender['host'] = ip
+       n = Node().initWithDict(sender)

        self.insertNode(n)
        return nodes, self.node.senderDict()
     
     def xmlrpc_store_value(self, key, value, sender):
        self.insertNode(n)
        return nodes, self.node.senderDict()
     
     def xmlrpc_store_value(self, key, value, sender):

+       key = decode(key)

        h1 = sha(key+value).digest()
        t = `time.time()`
        if not self.store.has_key(h1):
        h1 = sha(key+value).digest()
        t = `time.time()`
        if not self.store.has_key(h1):


@@ -226,14 +237,18 @@ class Khashmir(xmlrpc.XMLRPC):
            self.itime.put(t, h1)
 
        ip = self.crequest.getClientIP()
            self.itime.put(t, h1)
 
        ip = self.crequest.getClientIP()

-       n = Node(sender['id'], ip, sender['port'])
+       sender['host'] = ip
+       n = Node().initWithDict(sender)

        self.insertNode(n)
        return self.node.senderDict()
        
     def xmlrpc_find_value(self, key, sender):
        ip = self.crequest.getClientIP()
        self.insertNode(n)
        return self.node.senderDict()
        
     def xmlrpc_find_value(self, key, sender):
        ip = self.crequest.getClientIP()

-       n = Node(sender['id'], ip, sender['port'])
+       key = decode(key)
+       sender['host'] = ip
+       n = Node().initWithDict(sender)

        self.insertNode(n)
        self.insertNode(n)

+

        if self.kw.has_key(key):
            c = self.kw.cursor()
            tup = c.set(key)
        if self.kw.has_key(key):
            c = self.kw.cursor()
            tup = c.set(key)


@@ -255,7 +270,7 @@ class Khashmir(xmlrpc.XMLRPC):
 
 #------ testing
 
 
 #------ testing
 

-def test_build_net(quiet=0, peers=8, pause=1):
+def test_build_net(quiet=0, peers=24, host='localhost',  pause=1):

     from whrandom import randrange
     import thread
     port = 2001
     from whrandom import randrange
     import thread
     port = 2001


@@ -265,7 +280,7 @@ def test_build_net(quiet=0, peers=8, pause=1):
        print "Building %s peer table." % peers
        
     for i in xrange(peers):
        print "Building %s peer table." % peers
        
     for i in xrange(peers):

-       a = Khashmir('localhost', port + i)
+       a = Khashmir(host, port + i)

        l.append(a)
     
 
        l.append(a)
     
 


@@ -279,11 +294,11 @@ def test_build_net(quiet=0, peers=8, pause=1):
 
     for peer in l[1:]:
        n = l[randrange(0, len(l))].node
 
     for peer in l[1:]:
        n = l[randrange(0, len(l))].node

-       peer.addContact(n.host, n.port)
+       peer.addContact(host, n.port)

        n = l[randrange(0, len(l))].node
        n = l[randrange(0, len(l))].node

-       peer.addContact(n.host, n.port)
+       peer.addContact(host, n.port)

        n = l[randrange(0, len(l))].node
        n = l[randrange(0, len(l))].node

-       peer.addContact(n.host, n.port)
+       peer.addContact(host, n.port)

        if pause:
            time.sleep(.30)
            
        if pause:
            time.sleep(.30)
            


@@ -310,8 +325,8 @@ def test_find_nodes(l, quiet=0):
     a = l[randrange(0,n)]
     b = l[randrange(0,n)]
     
     a = l[randrange(0,n)]
     b = l[randrange(0,n)]
     

-    def callback(nodes, flag=flag):
-       if (len(nodes) >0) and (nodes[0].id == b.node.id):
+    def callback(nodes, flag=flag, id = b.node.id):
+       if (len(nodes) >0) and (nodes[0].id == id):

            print "test_find_nodes      PASSED"
        else:
            print "test_find_nodes      FAILED"
            print "test_find_nodes      PASSED"
        else:
            print "test_find_nodes      FAILED"

diff --git a/knode.py b/knode.py
index aecf14e4aa753b3e88ab8c64c108426f1772f92e..52abbed41f81d217b029e578579868274faade69 100644 (file)
--- a/knode.py
+++ b/knode.py
@@ -2,6 +2,7 @@ from node import Node
 from twisted.internet.defer import Deferred
 from xmlrpcclient import XMLRPCClientFactory as factory
 from const import reactor
 from twisted.internet.defer import Deferred
 from xmlrpcclient import XMLRPCClientFactory as factory
 from const import reactor

+from base64 import encodestring as encode

 
 class KNode(Node):
     def ping(self, sender):
 
 class KNode(Node):
     def ping(self, sender):


@@ -16,11 +17,11 @@ class KNode(Node):
        return df
     def storeValue(self, key, value, sender):
        df = Deferred()
        return df
     def storeValue(self, key, value, sender):
        df = Deferred()

-       f = factory('store_value', (key, value, sender), df.callback, df.errback)
+       f = factory('store_value', (encode(key), encode(value), sender), df.callback, df.errback)

        reactor.connectTCP(self.host, self.port, f)
        return df
     def findValue(self, key, sender):
        df = Deferred()
        reactor.connectTCP(self.host, self.port, f)
        return df
     def findValue(self, key, sender):
        df = Deferred()

-       f = factory('find_value', (key, sender), df.callback, df.errback)
+       f = factory('find_value', (encode(key), sender), df.callback, df.errback)

        reactor.connectTCP(self.host, self.port, f)
        return df
        reactor.connectTCP(self.host, self.port, f)
        return df

diff --git a/node.py b/node.py
index 6bc2ef2e7da0cf995cf30f9b287b86046d4b41d4..815f00c281778366ff58d24677317ca8a460fd6e 100644 (file)
--- a/node.py
+++ b/node.py
@@ -1,21 +1,33 @@
 import hash
 import time
 from types import *
 import hash
 import time
 from types import *

+from xmlrpclib import Binary

 
 class Node:
     """encapsulate contact info"""
 
 class Node:
     """encapsulate contact info"""

-    def __init__(self, id, host, port):
+    def init(self, id, host, port):

        self.id = id
        self.int = hash.intify(id)
        self.host = host
        self.port = port
        self.lastSeen = time.time()
        self.id = id
        self.int = hash.intify(id)
        self.host = host
        self.port = port
        self.lastSeen = time.time()

+       self._senderDict = {'id': Binary(self.id), 'port' : self.port, 'host' : self.host}
+       return self
+       
+    def initWithDict(self, dict):
+       self._senderDict = dict
+       self.id = dict['id'].data
+       self.int = hash.intify(self.id)
+       self.port = dict['port']
+       self.host = dict['host']
+       self.lastSeen = time.time()
+       return self

        
     def updateLastSeen(self):
        self.lastSeen = time.time()
 
     def senderDict(self):
        
     def updateLastSeen(self):
        self.lastSeen = time.time()
 
     def senderDict(self):

-       return {'id': self.id, 'port' : self.port, 'host' : self.host}
+       return self._senderDict

        
     def __repr__(self):
        return `(self.id, self.host, self.port)`
        
     def __repr__(self):
        return `(self.id, self.host, self.port)`

diff --git a/xmlrpcclient.py b/xmlrpcclient.py
index 230f3ad56762169eae6caa1f2955201fd22d4e70..5c8f3cf2565329cc235cf9f598e296c86eb3abd4 100644 (file)
--- a/xmlrpcclient.py
+++ b/xmlrpcclient.py
@@ -17,9 +17,13 @@ class XMLRPCClient(HTTPClient):
         self.transport.write('\r\n')
        
     def handleResponse(self, buf):
         self.transport.write('\r\n')
        
     def handleResponse(self, buf):

-       args, name = loads(buf)
-       apply(self.d.callback, args)
-
+       try:
+           args, name = loads(buf)
+       except Exception, e:
+           print "response decode error: " + `e`
+           self.d.errback()
+       else:
+           apply(self.d.callback, args)

 
 class XMLRPCClientFactory(ClientFactory):
     def __init__(self, method, args, callback=None, errback=None):
 
 class XMLRPCClientFactory(ClientFactory):
     def __init__(self, method, args, callback=None, errback=None):