Merge pull request #21 from iotaledger/feat/peerstorage

Persist known peers

packages/database/database.go

@@ -115,6 +115,40 @@ func (this *databaseImpl) Get(key []byte) ([]byte, error) {
 	return result, err
 }
 
+func (this *databaseImpl) Delete(key []byte) error {
+	err := this.db.Update(func(txn *badger.Txn) error {
+		err := txn.Delete(key)
+		return err
+	})
+	return err
+}
+
+func (this *databaseImpl) ForEach(consumer func([]byte, []byte)) error {
+	err := this.db.View(func(txn *badger.Txn) error {
+		// create an iterator the default options
+		it := txn.NewIterator(badger.DefaultIteratorOptions)
+		defer it.Close()
+
+		// avoid allocations by reusing the value buffer
+		var value []byte
+
+		// loop through every key-value-pair and call the function
+		for it.Rewind(); it.Valid(); it.Next() {
+			item := it.Item()
+
+			var err error
+			value, err = item.ValueCopy(value)
+			if err != nil {
+				return err
+			}
+
+			consumer(item.Key(), value)
+		}
+		return nil
+	})
+	return err
+}
+
 func (this *databaseImpl) Close() error {
 	this.openLock.Lock()
 	defer this.openLock.Unlock()

packages/database/interfaces.go

@@ -3,7 +3,9 @@ package database
 type Database interface {
 	Open() error
 	Set(key []byte, value []byte) error
-	Get(key []byte) ([]byte, error)
 	Contains(key []byte) (bool, error)
+	Get(key []byte) ([]byte, error)
+	ForEach(func(key []byte, value []byte)) error
+	Delete(key []byte) error
 	Close() error
 }

plugins/autopeering/peerstorage/peerstorage.go

+package peerstorage
+
+import (
+	"bytes"
+	"strconv"
+
+	"github.com/iotaledger/goshimmer/packages/database"
+	"github.com/iotaledger/goshimmer/packages/events"
+	"github.com/iotaledger/goshimmer/packages/node"
+	"github.com/iotaledger/goshimmer/plugins/autopeering/instances/knownpeers"
+	"github.com/iotaledger/goshimmer/plugins/autopeering/types/peer"
+)
+
+const peerDbName string = "peers"
+
+var peerDb database.Database
+
+func initDb() {
+	db, err := database.Get(peerDbName)
+	if err != nil {
+		panic(err)
+	}
+
+	peerDb = db
+}
+
+func storePeer(p *peer.Peer) {
+	err := peerDb.Set(p.Identity.Identifier, p.Marshal())
+	if err != nil {
+		panic(err)
+	}
+}
+
+func removePeer(p *peer.Peer) {
+	err := peerDb.Delete(p.Identity.Identifier)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func loadPeers(plugin *node.Plugin) {
+	var count int
+
+	err := peerDb.ForEach(func(key []byte, value []byte) {
+		peer, err := peer.Unmarshal(value)
+		if err != nil {
+			panic(err)
+		}
+		// the peers are stored by identifier in the db
+		if !bytes.Equal(key, peer.Identity.Identifier) {
+			panic("Invalid item in '" + peerDbName + "' database")
+		}
+
+		knownpeers.INSTANCE.AddOrUpdate(peer)
+		count++
+		plugin.LogDebug("Added stored peer: " + peer.Address.String() + " / " + peer.Identity.StringIdentifier)
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	plugin.LogSuccess("Restored " + strconv.Itoa(count) + " peers from database")
+}
+
+func Configure(plugin *node.Plugin) {
+	initDb()
+
+	// do not store the entry nodes by ignoring all peers currently contained in konwnpeers
+	// add peers from db
+	loadPeers(plugin)
+
+	// subscribe to all known peers' events
+	knownpeers.INSTANCE.Events.Add.Attach(events.NewClosure(func(p *peer.Peer) {
+		storePeer(p)
+	}))
+	knownpeers.INSTANCE.Events.Update.Attach(events.NewClosure(func(p *peer.Peer) {
+		storePeer(p)
+	}))
+	knownpeers.INSTANCE.Events.Remove.Attach(events.NewClosure(func(p *peer.Peer) {
+		removePeer(p)
+	}))
+}

plugins/autopeering/plugin.go

@@ -8,6 +8,7 @@ import (
 	"github.com/iotaledger/goshimmer/plugins/autopeering/instances/acceptedneighbors"
 	"github.com/iotaledger/goshimmer/plugins/autopeering/instances/chosenneighbors"
 	"github.com/iotaledger/goshimmer/plugins/autopeering/instances/knownpeers"
+	"github.com/iotaledger/goshimmer/plugins/autopeering/peerstorage"
 	"github.com/iotaledger/goshimmer/plugins/autopeering/protocol"
 	"github.com/iotaledger/goshimmer/plugins/autopeering/saltmanager"
 	"github.com/iotaledger/goshimmer/plugins/autopeering/server"
@@ -22,6 +23,7 @@ func configure(plugin *node.Plugin) {
 	instances.Configure(plugin)
 	server.Configure(plugin)
 	protocol.Configure(plugin)
+	peerstorage.Configure(plugin)
 
 	daemon.Events.Shutdown.Attach(events.NewClosure(func() {
 		server.Shutdown(plugin)

plugins/autopeering/types/peerregister/peer_register.go

@@ -42,6 +42,12 @@ func (this *PeerRegister) AddOrUpdate(peer *peer.Peer, lock ...bool) bool {
 		existingPeer.Address = peer.Address
 		existingPeer.GossipPort = peer.GossipPort
 		existingPeer.PeeringPort = peer.PeeringPort
+		existingPeer.Salt = peer.Salt
+
+		// also update the public key if not yet present
+		if existingPeer.Identity.PublicKey == nil {
+			existingPeer.Identity.PublicKey = peer.Identity.PublicKey
+		}
 
 		this.Events.Update.Trigger(existingPeer)