upgrade protobuf and switch to bytes keys

fixes #177

dht.go

@@ -143,14 +143,13 @@ func makeDHT(ctx context.Context, h host.Host, dstore ds.Batching, protocols []p
 }
 
 // putValueToPeer stores the given key/value pair at the peer 'p'
-func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID,
-	key string, rec *recpb.Record) error {
+func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID, rec *recpb.Record) error {
 
-	pmes := pb.NewMessage(pb.Message_PUT_VALUE, key, 0)
+	pmes := pb.NewMessage(pb.Message_PUT_VALUE, rec.Key, 0)
 	pmes.Record = rec
 	rpmes, err := dht.sendRequest(ctx, p, pmes)
 	if err != nil {
-		log.Debugf("putValueToPeer: %v. (peer: %s, key: %s)", err, p.Pretty(), key)
+		log.Debugf("putValueToPeer: %v. (peer: %s, key: %s)", err, p.Pretty(), loggableKey(string(rec.Key)))
 		return err
 	}
 
@@ -183,7 +182,7 @@ func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p peer.ID, key string)
 		log.Debug("getValueOrPeers: got value")
 
 		// make sure record is valid.
-		err = dht.Validator.Validate(record.GetKey(), record.GetValue())
+		err = dht.Validator.Validate(string(record.GetKey()), record.GetValue())
 		if err != nil {
 			log.Info("Received invalid record! (discarded)")
 			// return a sentinal to signify an invalid record was received
@@ -212,7 +211,7 @@ func (dht *IpfsDHT) getValueSingle(ctx context.Context, p peer.ID, key string) (
 	eip := log.EventBegin(ctx, "getValueSingle", meta)
 	defer eip.Done()
 
-	pmes := pb.NewMessage(pb.Message_GET_VALUE, key, 0)
+	pmes := pb.NewMessage(pb.Message_GET_VALUE, []byte(key), 0)
 	resp, err := dht.sendRequest(ctx, p, pmes)
 	switch err {
 	case nil:
@@ -236,7 +235,7 @@ func (dht *IpfsDHT) getLocal(key string) (*recpb.Record, error) {
 	}
 
 	// Double check the key. Can't hurt.
-	if rec != nil && rec.GetKey() != key {
+	if rec != nil && string(rec.GetKey()) != key {
 		log.Errorf("BUG getLocal: found a DHT record that didn't match it's key: %s != %s", rec.GetKey(), key)
 		return nil, nil
 
@@ -293,7 +292,7 @@ func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p peer.ID, id peer.ID) (
 		})
 	defer eip.Done()
 
-	pmes := pb.NewMessage(pb.Message_FIND_NODE, string(id), 0)
+	pmes := pb.NewMessage(pb.Message_FIND_NODE, []byte(id), 0)
 	resp, err := dht.sendRequest(ctx, p, pmes)
 	switch err {
 	case nil:
@@ -311,7 +310,7 @@ func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key *cid
 	eip := log.EventBegin(ctx, "findProvidersSingle", p, key)
 	defer eip.Done()
 
-	pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, key.KeyString(), 0)
+	pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, key.Bytes(), 0)
 	resp, err := dht.sendRequest(ctx, p, pmes)
 	switch err {
 	case nil:
@@ -327,7 +326,7 @@ func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key *cid
 
 // nearestPeersToQuery returns the routing tables closest peers.
 func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
-	closer := dht.routingTable.NearestPeers(kb.ConvertKey(pmes.GetKey()), count)
+	closer := dht.routingTable.NearestPeers(kb.ConvertKey(string(pmes.GetKey())), count)
 	return closer
 }
 

ext_test.go

@@ -105,8 +105,8 @@ func TestGetFailures(t *testing.T) {
 
 		rec := record.MakePutRecord(str, []byte("blah"))
 		req := pb.Message{
-			Type:   &typ,
-			Key:    &str,
+			Type:   typ,
+			Key:    []byte(str),
 			Record: rec,
 		}
 

handlers.go

@@ -1,6 +1,7 @@
 package dht
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -59,7 +60,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 
 	// first, is there even a key?
 	k := pmes.GetKey()
-	if k == "" {
+	if len(k) == 0  {
 		return nil, errors.New("handleGetValue but no key was provided")
 		// TODO: send back an error response? could be bad, but the other node's hanging.
 	}
@@ -90,7 +91,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 	return resp, nil
 }
 
-func (dht *IpfsDHT) checkLocalDatastore(k string) (*recpb.Record, error) {
+func (dht *IpfsDHT) checkLocalDatastore(k []byte) (*recpb.Record, error) {
 	log.Debugf("%s handleGetValue looking into ds", dht.self)
 	dskey := convertToDsKey(k)
 	iVal, err := dht.datastore.Get(dskey)
@@ -150,8 +151,7 @@ func (dht *IpfsDHT) checkLocalDatastore(k string) (*recpb.Record, error) {
 
 // Cleans the record (to avoid storing arbitrary data).
 func cleanRecord(rec *recpb.Record) {
-	rec.XXX_unrecognized = nil
-	rec.TimeReceived = nil
+	rec.TimeReceived = ""
 }
 
 // Store a value in this peer local storage
@@ -170,14 +170,14 @@ func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 		return nil, errors.New("nil record")
 	}
 
-	if pmes.GetKey() != rec.GetKey() {
+	if !bytes.Equal(pmes.GetKey(), rec.GetKey()) {
 		return nil, errors.New("put key doesn't match record key")
 	}
 
 	cleanRecord(rec)
 
 	// Make sure the record is valid (not expired, valid signature etc)
-	if err = dht.Validator.Validate(rec.GetKey(), rec.GetValue()); err != nil {
+	if err = dht.Validator.Validate(string(rec.GetKey()), rec.GetValue()); err != nil {
 		log.Warningf("Bad dht record in PUT from: %s. %s", p.Pretty(), err)
 		return nil, err
 	}
@@ -194,7 +194,7 @@ func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 
 	if existing != nil {
 		recs := [][]byte{rec.GetValue(), existing.GetValue()}
-		i, err := dht.Validator.Select(rec.GetKey(), recs)
+		i, err := dht.Validator.Select(string(rec.GetKey()), recs)
 		if err != nil {
 			log.Warningf("Bad dht record in PUT from %s: %s", p.Pretty(), err)
 			return nil, err
@@ -206,7 +206,7 @@ func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 	}
 
 	// record the time we receive every record
-	rec.TimeReceived = proto.String(u.FormatRFC3339(time.Now()))
+	rec.TimeReceived = u.FormatRFC3339(time.Now())
 
 	data, err := proto.Marshal(rec)
 	if err != nil {
@@ -245,7 +245,7 @@ func (dht *IpfsDHT) getRecordFromDatastore(dskey ds.Key) (*recpb.Record, error)
 		return nil, nil
 	}
 
-	err = dht.Validator.Validate(rec.GetKey(), rec.GetValue())
+	err = dht.Validator.Validate(string(rec.GetKey()), rec.GetValue())
 	if err != nil {
 		// Invalid record in datastore, probably expired but don't return an error,
 		// we'll just overwrite it
@@ -263,7 +263,7 @@ func (dht *IpfsDHT) handlePing(_ context.Context, p peer.ID, pmes *pb.Message) (
 
 func (dht *IpfsDHT) handleFindPeer(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
 	defer log.EventBegin(ctx, "handleFindPeer", p).Done()
-	resp := pb.NewMessage(pmes.GetType(), "", pmes.GetClusterLevel())
+	resp := pb.NewMessage(pmes.GetType(), nil, pmes.GetClusterLevel())
 	var closest []peer.ID
 
 	// if looking for self... special case where we send it on CloserPeers.
@@ -331,7 +331,7 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.
 	defer log.Debugf("%s end", reqDesc)
 
 	// check if we have this value, to add ourselves as provider.
-	has, err := dht.datastore.Has(convertToDsKey(c.KeyString()))
+	has, err := dht.datastore.Has(convertToDsKey(c.Bytes()))
 	if err != nil && err != ds.ErrNotFound {
 		log.Debugf("unexpected datastore error: %v\n", err)
 		has = false
@@ -403,6 +403,6 @@ func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.M
 	return nil, nil
 }
 
-func convertToDsKey(s string) ds.Key {
-	return ds.NewKey(base32.RawStdEncoding.EncodeToString([]byte(s)))
+func convertToDsKey(s []byte) ds.Key {
+	return ds.NewKey(base32.RawStdEncoding.EncodeToString(s))
 }

handlers_test.go

@@ -10,10 +10,9 @@ import (
 
 func TestCleanRecordSigned(t *testing.T) {
 	actual := new(recpb.Record)
-	actual.TimeReceived = proto.String("time")
-	actual.XXX_unrecognized = []byte("extra data")
+	actual.TimeReceived = "time"
 	actual.Value = []byte("value")
-	actual.Key = proto.String("key")
+	actual.Key = []byte("key")
 
 	cleanRecord(actual)
 	actualBytes, err := proto.Marshal(actual)
@@ -23,7 +22,7 @@ func TestCleanRecordSigned(t *testing.T) {
 
 	expected := new(recpb.Record)
 	expected.Value = []byte("value")
-	expected.Key = proto.String("key")
+	expected.Key = []byte("key")
 	expectedBytes, err := proto.Marshal(expected)
 	if err != nil {
 		t.Fatal(err)
@@ -36,9 +35,8 @@ func TestCleanRecordSigned(t *testing.T) {
 
 func TestCleanRecord(t *testing.T) {
 	actual := new(recpb.Record)
-	actual.TimeReceived = proto.String("time")
-	actual.XXX_unrecognized = []byte("extra data")
-	actual.Key = proto.String("key")
+	actual.TimeReceived = "time"
+	actual.Key = []byte("key")
 	actual.Value = []byte("value")
 
 	cleanRecord(actual)
@@ -48,7 +46,7 @@ func TestCleanRecord(t *testing.T) {
 	}
 
 	expected := new(recpb.Record)
-	expected.Key = proto.String("key")
+	expected.Key = []byte("key")
 	expected.Value = []byte("value")
 	expectedBytes, err := proto.Marshal(expected)
 	if err != nil {

package.json

@@ -30,7 +30,7 @@
       "version": "1.0.0"
     },
     {
-      "hash": "QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV",
+      "hash": "QmdxUuburamoF6zF9qjeQC4WYcWGbWuRmdLacMEsW8ioD8",
       "name": "gogo-protobuf",
       "version": "0.0.0"
     },
@@ -72,9 +72,9 @@
     },
     {
       "author": "whyrusleeping",
-      "hash": "QmVsp2KdPYE6M8ryzCk5KHLo3zprcY5hBDaYx6uPCFUdxA",
+      "hash": "Qmf3oiykN5MQyJHhtAqUaLFrUD3MFAk7WijqVWq6rSajJV",
       "name": "go-libp2p-record",
-      "version": "4.1.3"
+      "version": "4.1.4"
     },
     {
       "author": "whyrusleeping",

pb/Makefile

@@ -4,7 +4,7 @@ GO = $(PB:.proto=.pb.go)
 all: $(GO)
 
 %.pb.go: %.proto
-		protoc --gogo_out=. --proto_path=../../../../../../:/usr/local/opt/protobuf/include:. $<
+		protoc --proto_path=$(GOPATH)/src:. --gogofast_out=. $<
 
 clean:
 		rm -f *.pb.go