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

import (
4
	"context"
5
	"fmt"
Jeromy's avatar
Jeromy committed
6
	"io"
7
	"sync/atomic"
8
	"time"
Jeromy's avatar
Jeromy committed
9

Jeromy's avatar
Jeromy committed
10
	bsmsg "github.com/ipfs/go-bitswap/message"
Raúl Kripalani's avatar
Raúl Kripalani committed
11
	"github.com/libp2p/go-libp2p-core/helpers"
Jeromy's avatar
Jeromy committed
12 13 14 15

	ggio "github.com/gogo/protobuf/io"
	cid "github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log"
Raúl Kripalani's avatar
Raúl Kripalani committed
16 17 18 19 20 21
	"github.com/libp2p/go-libp2p-core/connmgr"
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-libp2p-core/peer"
	peerstore "github.com/libp2p/go-libp2p-core/peerstore"
	"github.com/libp2p/go-libp2p-core/routing"
Jeromy's avatar
Jeromy committed
22
	ma "github.com/multiformats/go-multiaddr"
23 24
)

Jeromy's avatar
Jeromy committed
25
var log = logging.Logger("bitswap_network")
Jeromy's avatar
Jeromy committed
26

27 28
var sendMessageTimeout = time.Minute * 10

29
// NewFromIpfsHost returns a BitSwapNetwork supported by underlying IPFS host.
30
func NewFromIpfsHost(host host.Host, r routing.ContentRouting) BitSwapNetwork {
31
	bitswapNetwork := impl{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
32
		host:    host,
33
		routing: r,
34
	}
35
	return &bitswapNetwork
36 37
}

38 39
// impl transforms the ipfs network interface, which sends and receives
// NetMessage objects, into the bitswap network interface.
40
type impl struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
41
	host    host.Host
42
	routing routing.ContentRouting
43 44 45

	// inbound messages from the network are forwarded to the receiver
	receiver Receiver
46

47
	stats Stats
48 49
}

Jeromy's avatar
Jeromy committed
50
type streamMessageSender struct {
Raúl Kripalani's avatar
Raúl Kripalani committed
51
	s network.Stream
Jeromy's avatar
Jeromy committed
52 53 54
}

func (s *streamMessageSender) Close() error {
Raúl Kripalani's avatar
Raúl Kripalani committed
55
	return helpers.FullClose(s.s)
Jeromy's avatar
Jeromy committed
56 57
}

58 59 60 61
func (s *streamMessageSender) Reset() error {
	return s.s.Reset()
}

62 63
func (s *streamMessageSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessage) error {
	return msgToStream(ctx, s.s, msg)
64 65
}

Raúl Kripalani's avatar
Raúl Kripalani committed
66
func msgToStream(ctx context.Context, s network.Stream, msg bsmsg.BitSwapMessage) error {
67 68 69 70
	deadline := time.Now().Add(sendMessageTimeout)
	if dl, ok := ctx.Deadline(); ok {
		deadline = dl
	}
Bob Potter's avatar
Bob Potter committed
71

72 73 74 75
	if err := s.SetWriteDeadline(deadline); err != nil {
		log.Warningf("error setting deadline: %s", err)
	}

76 77
	switch s.Protocol() {
	case ProtocolBitswap:
Bob Potter's avatar
Bob Potter committed
78
		if err := msg.ToNetV1(s); err != nil {
79 80 81 82
			log.Debugf("error: %s", err)
			return err
		}
	case ProtocolBitswapOne, ProtocolBitswapNoVers:
Bob Potter's avatar
Bob Potter committed
83
		if err := msg.ToNetV0(s); err != nil {
84 85 86 87 88 89
			log.Debugf("error: %s", err)
			return err
		}
	default:
		return fmt.Errorf("unrecognized protocol on remote: %s", s.Protocol())
	}
90 91 92 93

	if err := s.SetWriteDeadline(time.Time{}); err != nil {
		log.Warningf("error resetting deadline: %s", err)
	}
94
	return nil
Jeromy's avatar
Jeromy committed
95 96 97 98 99 100 101 102 103 104 105
}

func (bsnet *impl) NewMessageSender(ctx context.Context, p peer.ID) (MessageSender, error) {
	s, err := bsnet.newStreamToPeer(ctx, p)
	if err != nil {
		return nil, err
	}

	return &streamMessageSender{s: s}, nil
}

Raúl Kripalani's avatar
Raúl Kripalani committed
106
func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (network.Stream, error) {
107
	return bsnet.host.NewStream(ctx, p, ProtocolBitswap, ProtocolBitswapOne, ProtocolBitswapNoVers)
108 109 110 111 112 113 114 115
}

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

	s, err := bsnet.newStreamToPeer(ctx, p)
116 117 118
	if err != nil {
		return err
	}
119

Steven Allen's avatar
Steven Allen committed
120
	if err = msgToStream(ctx, s, outgoing); err != nil {
121
		s.Reset()
Steven Allen's avatar
Steven Allen committed
122
		return err
123
	}
124 125
	atomic.AddUint64(&bsnet.stats.MessagesSent, 1)

126
	// TODO(https://github.com/libp2p/go-libp2p-net/issues/28): Avoid this goroutine.
Raúl Kripalani's avatar
Raúl Kripalani committed
127
	go helpers.AwaitEOF(s)
128 129
	return s.Close()

130 131
}

132 133
func (bsnet *impl) SetDelegate(r Receiver) {
	bsnet.receiver = r
hannahhoward's avatar
hannahhoward committed
134 135 136 137 138 139
	bsnet.host.SetStreamHandler(ProtocolBitswap, bsnet.handleNewStream)
	bsnet.host.SetStreamHandler(ProtocolBitswapOne, bsnet.handleNewStream)
	bsnet.host.SetStreamHandler(ProtocolBitswapNoVers, bsnet.handleNewStream)
	bsnet.host.Network().Notify((*netNotifiee)(bsnet))
	// TODO: StopNotify.

140
}
141

142
func (bsnet *impl) ConnectTo(ctx context.Context, p peer.ID) error {
Raúl Kripalani's avatar
Raúl Kripalani committed
143
	return bsnet.host.Connect(ctx, peer.AddrInfo{ID: p})
144 145
}

146
// FindProvidersAsync returns a channel of providers for the given key.
147
func (bsnet *impl) FindProvidersAsync(ctx context.Context, k cid.Cid, max int) <-chan peer.ID {
148
	out := make(chan peer.ID, max)
149 150
	go func() {
		defer close(out)
151
		providers := bsnet.routing.FindProvidersAsync(ctx, k, max)
152
		for info := range providers {
153 154
			if info.ID == bsnet.host.ID() {
				continue // ignore self as provider
155
			}
Raúl Kripalani's avatar
Raúl Kripalani committed
156
			bsnet.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.TempAddrTTL)
157 158
			select {
			case <-ctx.Done():
159
				return
160 161 162 163 164
			case out <- info.ID:
			}
		}
	}()
	return out
165 166 167
}

// Provide provides the key to the network
168
func (bsnet *impl) Provide(ctx context.Context, k cid.Cid) error {
169
	return bsnet.routing.Provide(ctx, k, true)
170 171
}

172
// handleNewStream receives a new stream from the network.
Raúl Kripalani's avatar
Raúl Kripalani committed
173
func (bsnet *impl) handleNewStream(s network.Stream) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
174
	defer s.Close()
175 176

	if bsnet.receiver == nil {
177
		s.Reset()
178 179 180
		return
	}

Raúl Kripalani's avatar
Raúl Kripalani committed
181
	reader := ggio.NewDelimitedReader(s, network.MessageSizeMax)
182
	for {
Jeromy's avatar
Jeromy committed
183
		received, err := bsmsg.FromPBReader(reader)
184
		if err != nil {
Jeromy's avatar
Jeromy committed
185
			if err != io.EOF {
186
				s.Reset()
Jeromy's avatar
Jeromy committed
187 188 189
				go bsnet.receiver.ReceiveError(err)
				log.Debugf("bitswap net handleNewStream from %s error: %s", s.Conn().RemotePeer(), err)
			}
190 191
			return
		}
192

193 194 195 196
		p := s.Conn().RemotePeer()
		ctx := context.Background()
		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())
		bsnet.receiver.ReceiveMessage(ctx, p, received)
197
		atomic.AddUint64(&bsnet.stats.MessagesRecvd, 1)
198
	}
199
}
200

Raúl Kripalani's avatar
Raúl Kripalani committed
201
func (bsnet *impl) ConnectionManager() connmgr.ConnManager {
202 203 204
	return bsnet.host.ConnManager()
}

205 206
func (bsnet *impl) Stats() Stats {
	return Stats{
207 208 209 210 211
		MessagesRecvd: atomic.LoadUint64(&bsnet.stats.MessagesRecvd),
		MessagesSent:  atomic.LoadUint64(&bsnet.stats.MessagesSent),
	}
}

212 213 214 215 216 217
type netNotifiee impl

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

Raúl Kripalani's avatar
Raúl Kripalani committed
218
func (nn *netNotifiee) Connected(n network.Network, v network.Conn) {
219 220 221
	nn.impl().receiver.PeerConnected(v.RemotePeer())
}

Raúl Kripalani's avatar
Raúl Kripalani committed
222
func (nn *netNotifiee) Disconnected(n network.Network, v network.Conn) {
223 224 225
	nn.impl().receiver.PeerDisconnected(v.RemotePeer())
}

Raúl Kripalani's avatar
Raúl Kripalani committed
226 227 228 229
func (nn *netNotifiee) OpenedStream(n network.Network, v network.Stream) {}
func (nn *netNotifiee) ClosedStream(n network.Network, v network.Stream) {}
func (nn *netNotifiee) Listen(n network.Network, a ma.Multiaddr)         {}
func (nn *netNotifiee) ListenClose(n network.Network, a ma.Multiaddr)    {}