ipfs_impl.go 5.11 KB
Newer Older
1 2 3
package network

import (
4
	key "github.com/ipfs/go-ipfs/blocks/key"
5 6
	bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
	routing "github.com/ipfs/go-ipfs/routing"
7 8
	ma "gx/ipfs/QmYzDkkgAEmrcNzFCiYo6L1dTX4EAG1gZkbtdbd9trL4vd/go-multiaddr"
	peer "gx/ipfs/QmZpD74pUj6vuxTp1o6LhA3JavC2Bvh9fsWPPVvHnD9sE7/go-libp2p-peer"
Jeromy's avatar
Jeromy committed
9
	context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10
	logging "gx/ipfs/QmaDNZ4QMdBdku1YZWBysufYyoQt1negQGNav6PLYarbY8/go-log"
11 12
	host "gx/ipfs/QmcQTVCQWCN2MYgBHpFXE5S56rcg2mRsxaRgMYmA1UWgA8/go-libp2p/p2p/host"
	inet "gx/ipfs/QmcQTVCQWCN2MYgBHpFXE5S56rcg2mRsxaRgMYmA1UWgA8/go-libp2p/p2p/net"
13 14
)

Jeromy's avatar
Jeromy committed
15
var log = logging.Logger("bitswap_network")
Jeromy's avatar
Jeromy committed
16

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
17 18
// NewFromIpfsHost returns a BitSwapNetwork supported by underlying IPFS host
func NewFromIpfsHost(host host.Host, r routing.IpfsRouting) BitSwapNetwork {
19
	bitswapNetwork := impl{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
20
		host:    host,
21
		routing: r,
22
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
23
	host.SetStreamHandler(ProtocolBitswap, bitswapNetwork.handleNewStream)
24 25 26
	host.Network().Notify((*netNotifiee)(&bitswapNetwork))
	// TODO: StopNotify.

27
	return &bitswapNetwork
28 29
}

30 31
// impl transforms the ipfs network interface, which sends and receives
// NetMessage objects, into the bitswap network interface.
32
type impl struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
33
	host    host.Host
34
	routing routing.IpfsRouting
35 36 37

	// inbound messages from the network are forwarded to the receiver
	receiver Receiver
38 39
}

40
func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (inet.Stream, error) {
41

42 43
	// first, make sure we're connected.
	// if this fails, we cannot connect to given peer.
44 45
	//TODO(jbenet) move this into host.NewStream?
	if err := bsnet.host.Connect(ctx, peer.PeerInfo{ID: p}); err != nil {
46
		return nil, err
47 48
	}

49
	return bsnet.host.NewStream(ctx, ProtocolBitswap, p)
50 51 52 53 54 55 56 57
}

func (bsnet *impl) SendMessage(
	ctx context.Context,
	p peer.ID,
	outgoing bsmsg.BitSwapMessage) error {

	s, err := bsnet.newStreamToPeer(ctx, p)
58 59 60
	if err != nil {
		return err
	}
61 62
	defer s.Close()

63
	if err := outgoing.ToNet(s); err != nil {
64
		log.Debugf("error: %s", err)
65 66 67 68
		return err
	}

	return err
69 70
}

71
func (bsnet *impl) SendRequest(
72
	ctx context.Context,
73
	p peer.ID,
74 75
	outgoing bsmsg.BitSwapMessage) (bsmsg.BitSwapMessage, error) {

76
	s, err := bsnet.newStreamToPeer(ctx, p)
77 78 79
	if err != nil {
		return nil, err
	}
80 81 82
	defer s.Close()

	if err := outgoing.ToNet(s); err != nil {
83
		log.Debugf("error: %s", err)
84 85
		return nil, err
	}
86

87 88
	incoming, err := bsmsg.FromNet(s)
	if err != nil {
89
		log.Debugf("error: %s", err)
90 91 92 93
		return incoming, err
	}

	return incoming, nil
94 95
}

96 97
func (bsnet *impl) SetDelegate(r Receiver) {
	bsnet.receiver = r
98
}
99

100 101 102 103
func (bsnet *impl) ConnectTo(ctx context.Context, p peer.ID) error {
	return bsnet.host.Connect(ctx, peer.PeerInfo{ID: p})
}

104
// FindProvidersAsync returns a channel of providers for the given key
105
func (bsnet *impl) FindProvidersAsync(ctx context.Context, k key.Key, max int) <-chan peer.ID {
106 107 108 109 110 111 112 113 114

	// Since routing queries are expensive, give bitswap the peers to which we
	// have open connections. Note that this may cause issues if bitswap starts
	// precisely tracking which peers provide certain keys. This optimization
	// would be misleading. In the long run, this may not be the most
	// appropriate place for this optimization, but it won't cause any harm in
	// the short term.
	connectedPeers := bsnet.host.Network().Peers()
	out := make(chan peer.ID, len(connectedPeers)) // just enough buffer for these connectedPeers
115
	for _, id := range connectedPeers {
116 117 118
		if id == bsnet.host.ID() {
			continue // ignore self as provider
		}
119 120 121
		out <- id
	}

122 123 124 125
	go func() {
		defer close(out)
		providers := bsnet.routing.FindProvidersAsync(ctx, k, max)
		for info := range providers {
126 127
			if info.ID == bsnet.host.ID() {
				continue // ignore self as provider
128
			}
129
			bsnet.host.Peerstore().AddAddrs(info.ID, info.Addrs, peer.TempAddrTTL)
130 131
			select {
			case <-ctx.Done():
132
				return
133 134 135 136 137
			case out <- info.ID:
			}
		}
	}()
	return out
138 139 140
}

// Provide provides the key to the network
141
func (bsnet *impl) Provide(ctx context.Context, k key.Key) error {
142 143 144
	return bsnet.routing.Provide(ctx, k)
}

145 146
// handleNewStream receives a new stream from the network.
func (bsnet *impl) handleNewStream(s inet.Stream) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
147
	defer s.Close()
148 149 150 151 152

	if bsnet.receiver == nil {
		return
	}

153 154 155 156 157 158 159
	for {
		received, err := bsmsg.FromNet(s)
		if err != nil {
			go bsnet.receiver.ReceiveError(err)
			log.Debugf("bitswap net handleNewStream from %s error: %s", s.Conn().RemotePeer(), err)
			return
		}
160

161 162 163 164 165
		p := s.Conn().RemotePeer()
		ctx := context.Background()
		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())
		bsnet.receiver.ReceiveMessage(ctx, p, received)
	}
166
}
167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183

type netNotifiee impl

func (nn *netNotifiee) impl() *impl {
	return (*impl)(nn)
}

func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
	nn.impl().receiver.PeerConnected(v.RemotePeer())
}

func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
	nn.impl().receiver.PeerDisconnected(v.RemotePeer())
}

func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {}
func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
184 185
func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr)      {}
func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {}