mirror of
https://github.com/fluencelabs/go-libp2p-kad-dht
synced 2025-04-25 06:42:13 +00:00
9bc5b3d5d3
- updated go-ctxgroup and goprocess ctxgroup: AddChildGroup was changed to AddChild. Used in two files: - p2p/net/mock/mock_net.go - routing/dht/dht.go - updated context from hg repo to git prev. commit in hg was ad01a6fcc8a19d3a4478c836895ffe883bd2ceab. (context: make parentCancelCtx iterative) represents commit 84f8955a887232b6308d79c68b8db44f64df455c in git repo - updated context to master (b6fdb7d8a4ccefede406f8fe0f017fb58265054c) Aaron Jacobs (2): net/context: Don't accept a context in the DoSomethingSlow example. context: Be clear that users must cancel the result of WithCancel. Andrew Gerrand (1): go.net: use golang.org/x/... import paths Bryan C. Mills (1): net/context: Don't leak goroutines in Done example. Damien Neil (1): context: fix removal of cancelled timer contexts from parent David Symonds (2): context: Fix WithValue example code. net: add import comments. Sameer Ajmani (1): context: fix TestAllocs to account for ints in interfaces
133 lines
3.4 KiB
Go
133 lines
3.4 KiB
Go
package dht
|
|
|
|
import (
|
|
"errors"
|
|
"time"
|
|
|
|
inet "github.com/jbenet/go-ipfs/p2p/net"
|
|
peer "github.com/jbenet/go-ipfs/p2p/peer"
|
|
pb "github.com/jbenet/go-ipfs/routing/dht/pb"
|
|
ctxutil "github.com/jbenet/go-ipfs/util/ctx"
|
|
|
|
ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
|
|
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
|
|
)
|
|
|
|
// handleNewStream implements the inet.StreamHandler
|
|
func (dht *IpfsDHT) handleNewStream(s inet.Stream) {
|
|
go dht.handleNewMessage(s)
|
|
}
|
|
|
|
func (dht *IpfsDHT) handleNewMessage(s inet.Stream) {
|
|
defer s.Close()
|
|
|
|
ctx := dht.Context()
|
|
cr := ctxutil.NewReader(ctx, s) // ok to use. we defer close stream in this func
|
|
cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
|
|
r := ggio.NewDelimitedReader(cr, inet.MessageSizeMax)
|
|
w := ggio.NewDelimitedWriter(cw)
|
|
mPeer := s.Conn().RemotePeer()
|
|
|
|
// receive msg
|
|
pmes := new(pb.Message)
|
|
if err := r.ReadMsg(pmes); err != nil {
|
|
log.Debugf("Error unmarshaling data: %s", err)
|
|
return
|
|
}
|
|
|
|
// update the peer (on valid msgs only)
|
|
dht.updateFromMessage(ctx, mPeer, pmes)
|
|
|
|
// get handler for this msg type.
|
|
handler := dht.handlerForMsgType(pmes.GetType())
|
|
if handler == nil {
|
|
log.Debug("got back nil handler from handlerForMsgType")
|
|
return
|
|
}
|
|
|
|
// dispatch handler.
|
|
rpmes, err := handler(ctx, mPeer, pmes)
|
|
if err != nil {
|
|
log.Debugf("handle message error: %s", err)
|
|
return
|
|
}
|
|
|
|
// if nil response, return it before serializing
|
|
if rpmes == nil {
|
|
log.Warning("Got back nil response from request.")
|
|
return
|
|
}
|
|
|
|
// send out response msg
|
|
if err := w.WriteMsg(rpmes); err != nil {
|
|
log.Debugf("send response error: %s", err)
|
|
return
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// sendRequest sends out a request, but also makes sure to
|
|
// measure the RTT for latency measurements.
|
|
func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
|
|
|
|
log.Debugf("%s dht starting stream", dht.self)
|
|
s, err := dht.host.NewStream(ProtocolDHT, p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer s.Close()
|
|
|
|
cr := ctxutil.NewReader(ctx, s) // ok to use. we defer close stream in this func
|
|
cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
|
|
r := ggio.NewDelimitedReader(cr, inet.MessageSizeMax)
|
|
w := ggio.NewDelimitedWriter(cw)
|
|
|
|
start := time.Now()
|
|
|
|
if err := w.WriteMsg(pmes); err != nil {
|
|
return nil, err
|
|
}
|
|
log.Event(ctx, "dhtSentMessage", dht.self, p, pmes)
|
|
|
|
rpmes := new(pb.Message)
|
|
if err := r.ReadMsg(rpmes); err != nil {
|
|
return nil, err
|
|
}
|
|
if rpmes == nil {
|
|
return nil, errors.New("no response to request")
|
|
}
|
|
|
|
// update the peer (on valid msgs only)
|
|
dht.updateFromMessage(ctx, p, rpmes)
|
|
|
|
dht.peerstore.RecordLatency(p, time.Since(start))
|
|
log.Event(ctx, "dhtReceivedMessage", dht.self, p, rpmes)
|
|
return rpmes, nil
|
|
}
|
|
|
|
// sendMessage sends out a message
|
|
func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.ID, pmes *pb.Message) error {
|
|
|
|
log.Debugf("%s dht starting stream", dht.self)
|
|
s, err := dht.host.NewStream(ProtocolDHT, p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer s.Close()
|
|
|
|
cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
|
|
w := ggio.NewDelimitedWriter(cw)
|
|
|
|
if err := w.WriteMsg(pmes); err != nil {
|
|
return err
|
|
}
|
|
log.Event(ctx, "dhtSentMessage", dht.self, p, pmes)
|
|
return nil
|
|
}
|
|
|
|
func (dht *IpfsDHT) updateFromMessage(ctx context.Context, p peer.ID, mes *pb.Message) error {
|
|
dht.Update(ctx, p)
|
|
return nil
|
|
}
|