Merge pull request #158 from xssnick/andreypfau-dht-v2

DHT v2

adnl/adnl.go

@@ -541,6 +541,11 @@ func (a *ADNL) sendRequest(ctx context.Context, ch *Channel, req tl.Serializable
 }
 
 func (a *ADNL) send(ctx context.Context, buf []byte) error {
+	if err := ctx.Err(); err != nil {
+		// check if context is failed to not try to write
+		return err
+	}
+
 	dl, ok := ctx.Deadline()
 	if !ok {
 		dl = time.Now().Add(10 * time.Second)

adnl/dht/bucket.go

@@ -0,0 +1,62 @@
+package dht
+
+import (
+	"sort"
+	"sync"
+)
+
+type Bucket struct {
+	k     uint
+	nodes dhtNodeList
+	mx    sync.RWMutex
+}
+
+func newBucket(k uint) *Bucket {
+	b := &Bucket{
+		k:     k,
+		nodes: make([]*dhtNode, 0),
+	}
+	return b
+}
+
+func (b *Bucket) getNodes() dhtNodeList {
+	b.mx.RLock()
+	defer b.mx.RUnlock()
+
+	return b.nodes
+}
+
+func (b *Bucket) getNode(id string) *dhtNode {
+	b.mx.RLock()
+	defer b.mx.RUnlock()
+
+	for _, n := range b.nodes {
+		if n != nil && n.id() == id {
+			return n
+		}
+	}
+
+	return nil
+}
+
+func (b *Bucket) addNode(node *dhtNode) {
+	b.mx.Lock()
+	defer b.mx.Unlock()
+	defer b.sortAndFilter()
+
+	for i, n := range b.nodes {
+		if n != nil && n.id() == node.id() {
+			b.nodes[i] = node
+			return
+		}
+	}
+
+	b.nodes = append(b.nodes, node)
+}
+
+func (b *Bucket) sortAndFilter() {
+	sort.Sort(b.nodes)
+	if len(b.nodes) > int(b.k*2) {
+		b.nodes = b.nodes[:b.k*2]
+	}
+}