virtual.go 4.3 KB
Newer Older
Brian Tiger Chow's avatar
Brian Tiger Chow committed
1 2 3
package bitswap

import (
4
	"context"
Brian Tiger Chow's avatar
Brian Tiger Chow committed
5 6
	"errors"

7 8 9 10
	bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
	bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
	mockrouting "github.com/ipfs/go-ipfs/routing/mock"
	delay "github.com/ipfs/go-ipfs/thirdparty/delay"
11
	testutil "github.com/ipfs/go-ipfs/thirdparty/testutil"
12

13
	routing "gx/ipfs/QmPjTrrSfE6TzLv6ya6VWhGcCgPrUAdcgrDcQyRDX2VyW1/go-libp2p-routing"
14
	logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
15 16
	cid "gx/ipfs/QmTprEaAA2A9bst5XH7exuyi5KzNMK3SEDNN8rBDnKWcUS/go-cid"
	peer "gx/ipfs/QmXYjuNuxVzXKJCfWasQk1RqkhVLDM9jtUKhqc2WPQmFSB/go-libp2p-peer"
Brian Tiger Chow's avatar
Brian Tiger Chow committed
17 18
)

19 20
var log = logging.Logger("bstestnet")

21
func VirtualNetwork(rs mockrouting.Server, d delay.D) Network {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
22
	return &network{
23 24
		clients:       make(map[peer.ID]bsnet.Receiver),
		delay:         d,
25
		routingserver: rs,
26
		conns:         make(map[string]struct{}),
Brian Tiger Chow's avatar
Brian Tiger Chow committed
27 28 29 30
	}
}

type network struct {
31 32 33
	clients       map[peer.ID]bsnet.Receiver
	routingserver mockrouting.Server
	delay         delay.D
34
	conns         map[string]struct{}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
35 36
}

37
func (n *network) Adapter(p testutil.Identity) bsnet.BitSwapNetwork {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
38
	client := &networkClient{
39
		local:   p.ID(),
40
		network: n,
Brian Tiger Chow's avatar
Brian Tiger Chow committed
41
		routing: n.routingserver.Client(p),
Brian Tiger Chow's avatar
Brian Tiger Chow committed
42
	}
43
	n.clients[p.ID()] = client
Brian Tiger Chow's avatar
Brian Tiger Chow committed
44 45 46
	return client
}

47 48
func (n *network) HasPeer(p peer.ID) bool {
	_, found := n.clients[p]
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
49 50 51
	return found
}

Brian Tiger Chow's avatar
Brian Tiger Chow committed
52 53 54 55
// TODO should this be completely asynchronous?
// TODO what does the network layer do with errors received from services?
func (n *network) SendMessage(
	ctx context.Context,
56 57
	from peer.ID,
	to peer.ID,
Brian Tiger Chow's avatar
Brian Tiger Chow committed
58 59
	message bsmsg.BitSwapMessage) error {

60
	receiver, ok := n.clients[to]
Brian Tiger Chow's avatar
Brian Tiger Chow committed
61 62 63 64 65 66 67 68 69 70 71 72 73
	if !ok {
		return errors.New("Cannot locate peer on network")
	}

	// nb: terminate the context since the context wouldn't actually be passed
	// over the network in a real scenario

	go n.deliver(receiver, from, message)

	return nil
}

func (n *network) deliver(
74 75
	r bsnet.Receiver, from peer.ID, message bsmsg.BitSwapMessage) error {
	if message == nil || from == "" {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
76 77 78
		return errors.New("Invalid input")
	}

79 80
	n.delay.Wait()

81 82
	r.ReceiveMessage(context.TODO(), from, message)
	return nil
Brian Tiger Chow's avatar
Brian Tiger Chow committed
83 84 85
}

type networkClient struct {
86
	local peer.ID
Brian Tiger Chow's avatar
Brian Tiger Chow committed
87
	bsnet.Receiver
88
	network *network
89
	routing routing.IpfsRouting
Brian Tiger Chow's avatar
Brian Tiger Chow committed
90 91 92 93
}

func (nc *networkClient) SendMessage(
	ctx context.Context,
94
	to peer.ID,
Brian Tiger Chow's avatar
Brian Tiger Chow committed
95 96 97 98
	message bsmsg.BitSwapMessage) error {
	return nc.network.SendMessage(ctx, nc.local, to, message)
}

99
// FindProvidersAsync returns a channel of providers for the given key
100
func (nc *networkClient) FindProvidersAsync(ctx context.Context, k *cid.Cid, max int) <-chan peer.ID {
101 102 103 104 105 106 107 108 109

	// NB: this function duplicates the PeerInfo -> ID transformation in the
	// bitswap network adapter. Not to worry. This network client will be
	// deprecated once the ipfsnet.Mock is added. The code below is only
	// temporary.

	out := make(chan peer.ID)
	go func() {
		defer close(out)
110
		providers := nc.routing.FindProvidersAsync(ctx, k, max)
111 112 113 114 115 116 117 118
		for info := range providers {
			select {
			case <-ctx.Done():
			case out <- info.ID:
			}
		}
	}()
	return out
119 120
}

Jeromy's avatar
Jeromy committed
121 122 123 124 125 126 127
type messagePasser struct {
	net    *network
	target peer.ID
	local  peer.ID
	ctx    context.Context
}

128 129
func (mp *messagePasser) SendMsg(ctx context.Context, m bsmsg.BitSwapMessage) error {
	return mp.net.SendMessage(ctx, mp.local, mp.target, m)
Jeromy's avatar
Jeromy committed
130 131 132 133 134 135 136 137 138 139 140 141 142 143 144
}

func (mp *messagePasser) Close() error {
	return nil
}

func (n *networkClient) NewMessageSender(ctx context.Context, p peer.ID) (bsnet.MessageSender, error) {
	return &messagePasser{
		net:    n.network,
		target: p,
		local:  n.local,
		ctx:    ctx,
	}, nil
}

145
// Provide provides the key to the network
146
func (nc *networkClient) Provide(ctx context.Context, k *cid.Cid) error {
147
	return nc.routing.Provide(ctx, k, true)
148 149
}

Brian Tiger Chow's avatar
Brian Tiger Chow committed
150 151 152
func (nc *networkClient) SetDelegate(r bsnet.Receiver) {
	nc.Receiver = r
}
153 154 155 156 157

func (nc *networkClient) ConnectTo(_ context.Context, p peer.ID) error {
	if !nc.network.HasPeer(p) {
		return errors.New("no such peer in network")
	}
158 159 160 161 162 163 164 165
	tag := tagForPeers(nc.local, p)
	if _, ok := nc.network.conns[tag]; ok {
		log.Warning("ALREADY CONNECTED TO PEER (is this a reconnect? test lib needs fixing)")
		return nil
	}
	nc.network.conns[tag] = struct{}{}
	// TODO: add handling for disconnects

166 167 168 169
	nc.network.clients[p].PeerConnected(nc.local)
	nc.Receiver.PeerConnected(p)
	return nil
}
170 171 172 173 174 175 176

func tagForPeers(a, b peer.ID) string {
	if a < b {
		return string(a + b)
	}
	return string(b + a)
}