net: move Network implementation to own pkg
I needed the network implementation in its own package, because I'll be writing several services that will plug into _it_ that shouldn't be part of the core net package. and then there were dependency conflicts. yay. mux + identify are good examples of what i mean.
Juan Batiz-Benet committed
Dec 24, 2014 at 10:17 UTC
4807127def5d44f2cc71f15f3023e4f0190b54df
13 files changed
+165
-150
core/core.go
+2
-1
@@ -21,6 +21,7 @@ import (
21
merkledag "github.com/jbenet/go-ipfs/merkledag"
22
namesys "github.com/jbenet/go-ipfs/namesys"
23
inet "github.com/jbenet/go-ipfs/net"
24
+ ipfsnet "github.com/jbenet/go-ipfs/net/ipfsnet"
25
path "github.com/jbenet/go-ipfs/path"
26
peer "github.com/jbenet/go-ipfs/peer"
27
pin "github.com/jbenet/go-ipfs/pin"
@@ -121,7 +122,7 @@ func NewIpfsNode(ctx context.Context, cfg *config.Config, online bool) (n *IpfsN
122
return nil, debugerror.Wrap(err)
123
}
124
124
- n.Network, err = inet.NewNetwork(ctx, listenAddrs, n.Identity, n.Peerstore)
125
+ n.Network, err = ipfsnet.NewNetwork(ctx, listenAddrs, n.Identity, n.Peerstore)
126
if err != nil {
127
return nil, debugerror.Wrap(err)
128
}
net/backpressure/backpressure_test.go
+10
-34
@@ -8,30 +8,15 @@ import (
8
"time"
9
10
inet "github.com/jbenet/go-ipfs/net"
11
+ netutil "github.com/jbenet/go-ipfs/net/ipfsnet/util"
12
peer "github.com/jbenet/go-ipfs/peer"
13
eventlog "github.com/jbenet/go-ipfs/util/eventlog"
13
- testutil "github.com/jbenet/go-ipfs/util/testutil"
14
15
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
16
)
17
18
var log = eventlog.Logger("backpressure")
19
20
-func GenNetwork(t *testing.T, ctx context.Context) (inet.Network, error) {
21
- p := testutil.RandPeerNetParamsOrFatal(t)
22
- ps := peer.NewPeerstore()
23
- ps.AddAddress(p.ID, p.Addr)
24
- ps.AddPubKey(p.ID, p.PubKey)
25
- ps.AddPrivKey(p.ID, p.PrivKey)
26
- return inet.NewNetwork(ctx, ps.Addresses(p.ID), p.ID, ps)
27
-}
28
-
29
-func divulgeAddresses(a, b inet.Network) {
30
- id := a.LocalPeer()
31
- addrs := a.Peerstore().Addresses(id)
32
- b.Peerstore().AddAddresses(id, addrs)
33
-}
34
-
20
// TestBackpressureStreamHandler tests whether mux handler
21
// ratelimiting works. Meaning, since the handler is sequential
22
// it should block senders.
@@ -149,14 +134,8 @@ a problem.
134
// ok that's enough setup. let's do it!
135
136
ctx := context.Background()
152
- n1, err := GenNetwork(t, ctx)
153
- if err != nil {
154
- t.Fatal(err)
155
- }
156
- n2, err := GenNetwork(t, ctx)
157
- if err != nil {
158
- t.Fatal(err)
159
- }
137
+ n1 := netutil.GenNetwork(t, ctx)
138
+ n2 := netutil.GenNetwork(t, ctx)
139
140
// setup receiver handler
141
n1.SetHandler(inet.ProtocolTesting, receiver)
@@ -291,17 +270,11 @@ func TestStBackpressureStreamWrite(t *testing.T) {
270
271
// setup the networks
272
ctx := context.Background()
294
- n1, err := GenNetwork(t, ctx)
295
- if err != nil {
296
- t.Fatal(err)
297
- }
298
- n2, err := GenNetwork(t, ctx)
299
- if err != nil {
300
- t.Fatal(err)
301
- }
273
+ n1 := netutil.GenNetwork(t, ctx)
274
+ n2 := netutil.GenNetwork(t, ctx)
275
303
- divulgeAddresses(n1, n2)
304
- divulgeAddresses(n2, n1)
276
+ netutil.DivulgeAddresses(n1, n2)
277
+ netutil.DivulgeAddresses(n2, n1)
278
279
// setup sender handler on 1
280
n1.SetHandler(inet.ProtocolTesting, sender)
@@ -313,6 +286,9 @@ func TestStBackpressureStreamWrite(t *testing.T) {
286
287
// open a stream, from 2->1, this is our reader
288
s, err := n2.NewStream(inet.ProtocolTesting, n1.LocalPeer())
289
+ if err != nil {
290
+ t.Fatal(err)
291
+ }
292
293
// let's make sure r/w works.
294
testSenderWrote := func(bytesE int) {
net/interface.go
+1
-5
@@ -23,6 +23,7 @@ const (
23
ProtocolDHT ProtocolID = "/ipfs/dht"
24
ProtocolIdentify ProtocolID = "/ipfs/id"
25
ProtocolDiag ProtocolID = "/ipfs/diagnostics"
26
+ ProtocolRelay ProtocolID = "/ipfs/relay"
27
)
28
29
// MessageSizeMax is a soft (recommended) maximum for network messages.
@@ -96,11 +97,6 @@ type Network interface {
97
98
// CtxGroup returns the network's contextGroup
99
CtxGroup() ctxgroup.ContextGroup
99
-
100
- // IdentifyProtocol returns the instance of the object running the Identify
101
- // Protocol. This is what runs the ifps handshake-- this should be removed
102
- // if this abstracted out to its own package.
103
- IdentifyProtocol() *IDService
100
}
101
102
// Dialer represents a service that can dial out to peers
net/ipfsnet/net.go
renamed
+50
-48
@@ -5,14 +5,20 @@ import (
5
"fmt"
6
7
ic "github.com/jbenet/go-ipfs/crypto"
8
+ inet "github.com/jbenet/go-ipfs/net"
9
+ ids "github.com/jbenet/go-ipfs/net/services/identify"
10
+ mux "github.com/jbenet/go-ipfs/net/services/mux"
11
swarm "github.com/jbenet/go-ipfs/net/swarm"
12
peer "github.com/jbenet/go-ipfs/peer"
13
14
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
15
ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
16
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
17
+ eventlog "github.com/jbenet/go-ipfs/util/eventlog"
18
)
19
20
+var log = eventlog.Logger("net/mux")
21
+
22
type stream swarm.Stream
23
24
func (s *stream) SwarmStream() *swarm.Stream {
@@ -20,7 +26,7 @@ func (s *stream) SwarmStream() *swarm.Stream {
26
}
27
28
// Conn returns the connection this stream is part of.
23
-func (s *stream) Conn() Conn {
29
+func (s *stream) Conn() inet.Conn {
30
c := s.SwarmStream().Conn()
31
return (*conn_)(c)
32
}
@@ -50,7 +56,7 @@ func (c *conn_) SwarmConn() *swarm.Conn {
56
return (*swarm.Conn)(c)
57
}
58
53
-func (c *conn_) NewStreamWithProtocol(pr ProtocolID) (Stream, error) {
59
+func (c *conn_) NewStreamWithProtocol(pr inet.ProtocolID) (inet.Stream, error) {
60
s, err := (*swarm.Conn)(c).NewStream()
61
if err != nil {
62
return nil, err
@@ -58,7 +64,7 @@ func (c *conn_) NewStreamWithProtocol(pr ProtocolID) (Stream, error) {
64
65
ss := (*stream)(s)
66
61
- if err := WriteProtocolHeader(pr, ss); err != nil {
67
+ if err := mux.WriteProtocolHeader(pr, ss); err != nil {
68
ss.Close()
69
return nil, err
70
}
@@ -90,30 +96,32 @@ func (c *conn_) RemotePublicKey() ic.PubKey {
96
return c.SwarmConn().RemotePublicKey()
97
}
98
93
-// network implements the Network interface,
94
-type network struct {
95
- local peer.ID // local peer
96
- mux Mux // protocol multiplexing
97
- swarm *swarm.Swarm // peer connection multiplexing
99
+// Network implements the inet.Network interface.
100
+// It uses a swarm to connect to remote hosts.
101
+type Network struct {
102
+ local peer.ID // local peer
103
ps peer.Peerstore
99
- ids *IDService
104
+
105
+ swarm *swarm.Swarm // peer connection multiplexing
106
+ mux mux.Mux // protocol multiplexing
107
+ ids *ids.IDService
108
109
cg ctxgroup.ContextGroup // for Context closing
110
}
111
112
// NewNetwork constructs a new network and starts listening on given addresses.
113
func NewNetwork(ctx context.Context, listen []ma.Multiaddr, local peer.ID,
106
- peers peer.Peerstore) (Network, error) {
114
+ peers peer.Peerstore) (*Network, error) {
115
116
s, err := swarm.NewSwarm(ctx, listen, local, peers)
117
if err != nil {
118
return nil, err
119
}
120
113
- n := &network{
121
+ n := &Network{
122
local: local,
123
swarm: s,
116
- mux: Mux{Handlers: StreamHandlerMap{}},
124
+ mux: mux.Mux{Handlers: inet.StreamHandlerMap{}},
125
cg: ctxgroup.WithContext(ctx),
126
ps: peers,
127
}
@@ -127,20 +135,20 @@ func NewNetwork(ctx context.Context, listen []ma.Multiaddr, local peer.ID,
135
136
// setup a conn handler that immediately "asks the other side about them"
137
// this is ProtocolIdentify.
130
- n.ids = NewIDService(n)
138
+ n.ids = ids.NewIDService(n)
139
s.SetConnHandler(n.newConnHandler)
140
141
return n, nil
142
}
143
136
-func (n *network) newConnHandler(c *swarm.Conn) {
144
+func (n *Network) newConnHandler(c *swarm.Conn) {
145
cc := (*conn_)(c)
146
n.ids.IdentifyConn(cc)
147
}
148
149
// DialPeer attempts to establish a connection to a given peer.
150
// Respects the context.
143
-func (n *network) DialPeer(ctx context.Context, p peer.ID) error {
151
+func (n *Network) DialPeer(ctx context.Context, p peer.ID) error {
152
log.Debugf("[%s] network dialing peer [%s]", n.local, p)
153
sc, err := n.swarm.Dial(ctx, p)
154
if err != nil {
@@ -165,39 +173,40 @@ func (n *network) DialPeer(ctx context.Context, p peer.ID) error {
173
return nil
174
}
175
168
-func (n *network) Protocols() []ProtocolID {
176
+// Protocols returns the ProtocolIDs of all the registered handlers.
177
+func (n *Network) Protocols() []inet.ProtocolID {
178
return n.mux.Protocols()
179
}
180
181
// CtxGroup returns the network's ContextGroup
173
-func (n *network) CtxGroup() ctxgroup.ContextGroup {
182
+func (n *Network) CtxGroup() ctxgroup.ContextGroup {
183
return n.cg
184
}
185
186
// Swarm returns the network's peerstream.Swarm
178
-func (n *network) Swarm() *swarm.Swarm {
187
+func (n *Network) Swarm() *swarm.Swarm {
188
return n.Swarm()
189
}
190
191
// LocalPeer the network's LocalPeer
183
-func (n *network) LocalPeer() peer.ID {
192
+func (n *Network) LocalPeer() peer.ID {
193
return n.swarm.LocalPeer()
194
}
195
196
// Peers returns the connected peers
188
-func (n *network) Peers() []peer.ID {
197
+func (n *Network) Peers() []peer.ID {
198
return n.swarm.Peers()
199
}
200
201
// Peers returns the connected peers
193
-func (n *network) Peerstore() peer.Peerstore {
202
+func (n *Network) Peerstore() peer.Peerstore {
203
return n.ps
204
}
205
206
// Conns returns the connected peers
198
-func (n *network) Conns() []Conn {
207
+func (n *Network) Conns() []inet.Conn {
208
conns1 := n.swarm.Connections()
200
- out := make([]Conn, len(conns1))
209
+ out := make([]inet.Conn, len(conns1))
210
for i, c := range conns1 {
211
out[i] = (*conn_)(c)
212
}
@@ -205,9 +214,9 @@ func (n *network) Conns() []Conn {
214
}
215
216
// ConnsToPeer returns the connections in this Netowrk for given peer.
208
-func (n *network) ConnsToPeer(p peer.ID) []Conn {
217
+func (n *Network) ConnsToPeer(p peer.ID) []inet.Conn {
218
conns1 := n.swarm.ConnectionsToPeer(p)
210
- out := make([]Conn, len(conns1))
219
+ out := make([]inet.Conn, len(conns1))
220
for i, c := range conns1 {
221
out[i] = (*conn_)(c)
222
}
@@ -215,53 +224,53 @@ func (n *network) ConnsToPeer(p peer.ID) []Conn {
224
}
225
226
// ClosePeer connection to peer
218
-func (n *network) ClosePeer(p peer.ID) error {
227
+func (n *Network) ClosePeer(p peer.ID) error {
228
return n.swarm.CloseConnection(p)
229
}
230
231
// close is the real teardown function
223
-func (n *network) close() error {
232
+func (n *Network) close() error {
233
return n.swarm.Close()
234
}
235
236
// Close calls the ContextCloser func
228
-func (n *network) Close() error {
237
+func (n *Network) Close() error {
238
return n.cg.Close()
239
}
240
241
// BandwidthTotals returns the total amount of bandwidth transferred
233
-func (n *network) BandwidthTotals() (in uint64, out uint64) {
242
+func (n *Network) BandwidthTotals() (in uint64, out uint64) {
243
// need to implement this. probably best to do it in swarm this time.
244
// need a "metrics" object
245
return 0, 0
246
}
247
248
// ListenAddresses returns a list of addresses at which this network listens.
240
-func (n *network) ListenAddresses() []ma.Multiaddr {
249
+func (n *Network) ListenAddresses() []ma.Multiaddr {
250
return n.swarm.ListenAddresses()
251
}
252
253
// InterfaceListenAddresses returns a list of addresses at which this network
254
// listens. It expands "any interface" addresses (/ip4/0.0.0.0, /ip6/::) to
255
// use the known local interfaces.
247
-func (n *network) InterfaceListenAddresses() ([]ma.Multiaddr, error) {
256
+func (n *Network) InterfaceListenAddresses() ([]ma.Multiaddr, error) {
257
return swarm.InterfaceListenAddresses(n.swarm)
258
}
259
260
// Connectedness returns a state signaling connection capabilities
261
// For now only returns Connected || NotConnected. Expand into more later.
253
-func (n *network) Connectedness(p peer.ID) Connectedness {
262
+func (n *Network) Connectedness(p peer.ID) inet.Connectedness {
263
c := n.swarm.ConnectionsToPeer(p)
264
if c != nil && len(c) > 0 {
256
- return Connected
265
+ return inet.Connected
266
}
258
- return NotConnected
267
+ return inet.NotConnected
268
}
269
270
// NewStream returns a new stream to given peer p.
271
// If there is no connection to p, attempts to create one.
272
// If ProtocolID is "", writes no header.
264
-func (n *network) NewStream(pr ProtocolID, p peer.ID) (Stream, error) {
273
+func (n *Network) NewStream(pr inet.ProtocolID, p peer.ID) (inet.Stream, error) {
274
log.Debugf("[%s] network opening stream to peer [%s]: %s", n.local, p, pr)
275
s, err := n.swarm.NewStreamWithPeer(p)
276
if err != nil {
@@ -270,7 +279,7 @@ func (n *network) NewStream(pr ProtocolID, p peer.ID) (Stream, error) {
279
280
ss := (*stream)(s)
281
273
- if err := WriteProtocolHeader(pr, ss); err != nil {
282
+ if err := mux.WriteProtocolHeader(pr, ss); err != nil {
283
ss.Close()
284
return nil, err
285
}
@@ -280,23 +289,16 @@ func (n *network) NewStream(pr ProtocolID, p peer.ID) (Stream, error) {
289
290
// SetHandler sets the protocol handler on the Network's Muxer.
291
// This operation is threadsafe.
283
-func (n *network) SetHandler(p ProtocolID, h StreamHandler) {
292
+func (n *Network) SetHandler(p inet.ProtocolID, h inet.StreamHandler) {
293
n.mux.SetHandler(p, h)
294
}
295
287
-func (n *network) String() string {
296
+// String returns a string representation of Network.
297
+func (n *Network) String() string {
298
return fmt.Sprintf("<Network %s>", n.LocalPeer())
299
}
300
291
-func (n *network) IdentifyProtocol() *IDService {
301
+// IdentifyProtocol returns the network's IDService
302
+func (n *Network) IdentifyProtocol() *ids.IDService {
303
return n.ids
304
}
294
-
295
-func WriteProtocolHeader(pr ProtocolID, s Stream) error {
296
- if pr != "" { // only write proper protocol headers
297
- if err := WriteLengthPrefix(s, string(pr)); err != nil {
298
- return err
299
- }
300
- }
301
- return nil
302
-}
net/ipfsnet/net_test.go
renamed
+4
-2
@@ -6,7 +6,9 @@ import (
6
"time"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
+
10
inet "github.com/jbenet/go-ipfs/net"
11
+ netutil "github.com/jbenet/go-ipfs/net/ipfsnet/util"
12
)
13
14
// TestConnectednessCorrect starts a few networks, connects a few
@@ -17,13 +19,13 @@ func TestConnectednessCorrect(t *testing.T) {
19
20
nets := make([]inet.Network, 4)
21
for i := 0; i < 4; i++ {
20
- nets[i] = GenNetwork(t, ctx)
22
+ nets[i] = netutil.GenNetwork(t, ctx)
23
}
24
25
// connect 0-1, 0-2, 0-3, 1-2, 2-3
26
27
dial := func(a, b inet.Network) {
26
- DivulgeAddresses(b, a)
28
+ netutil.DivulgeAddresses(b, a)
29
if err := a.DialPeer(ctx, b.LocalPeer()); err != nil {
30
t.Fatalf("Failed to dial: %s", err)
31
}
net/ipfsnet/util/util.go
new
+31
@@ -0,0 +1,31 @@
1
+package testutil
2
+
3
+import (
4
+ "testing"
5
+
6
+ inet "github.com/jbenet/go-ipfs/net"
7
+ in "github.com/jbenet/go-ipfs/net/ipfsnet"
8
+ peer "github.com/jbenet/go-ipfs/peer"
9
+ tu "github.com/jbenet/go-ipfs/util/testutil"
10
+
11
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
12
+)
13
+
14
+func GenNetwork(t *testing.T, ctx context.Context) *in.Network {
15
+ p := tu.RandPeerNetParamsOrFatal(t)
16
+ ps := peer.NewPeerstore()
17
+ ps.AddAddress(p.ID, p.Addr)
18
+ ps.AddPubKey(p.ID, p.PubKey)
19
+ ps.AddPrivKey(p.ID, p.PrivKey)
20
+ n, err := in.NewNetwork(ctx, ps.Addresses(p.ID), p.ID, ps)
21
+ if err != nil {
22
+ t.Fatal(err)
23
+ }
24
+ return n
25
+}
26
+
27
+func DivulgeAddresses(a, b inet.Network) {
28
+ id := a.LocalPeer()
29
+ addrs := a.Peerstore().Addresses(id)
30
+ b.Peerstore().AddAddresses(id, addrs)
31
+}
net/mock/mock_conn.go
+2
-1
@@ -6,6 +6,7 @@ import (
6
7
ic "github.com/jbenet/go-ipfs/crypto"
8
inet "github.com/jbenet/go-ipfs/net"
9
+ mux "github.com/jbenet/go-ipfs/net/services/mux"
10
peer "github.com/jbenet/go-ipfs/peer"
11
12
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
@@ -86,7 +87,7 @@ func (c *conn) NewStreamWithProtocol(pr inet.ProtocolID) (inet.Stream, error) {
87
log.Debugf("Conn.NewStreamWithProtocol: %s --> %s", c.local, c.remote)
88
89
s := c.openStream()
89
- if err := inet.WriteProtocolHeader(pr, s); err != nil {
90
+ if err := mux.WriteProtocolHeader(pr, s); err != nil {
91
s.Close()
92
return nil, err
93
}
net/mock/mock_peernet.go
+7
-5
@@ -7,6 +7,8 @@ import (
7
8
ic "github.com/jbenet/go-ipfs/crypto"
9
inet "github.com/jbenet/go-ipfs/net"
10
+ ids "github.com/jbenet/go-ipfs/net/services/identify"
11
+ mux "github.com/jbenet/go-ipfs/net/services/mux"
12
peer "github.com/jbenet/go-ipfs/peer"
13
14
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -28,8 +30,8 @@ type peernet struct {
30
connsByLink map[*link]map[*conn]struct{}
31
32
// needed to implement inet.Network
31
- mux inet.Mux
32
- ids *inet.IDService
33
+ mux mux.Mux
34
+ ids *ids.IDService
35
36
cg ctxgroup.ContextGroup
37
sync.RWMutex
@@ -54,7 +56,7 @@ func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
56
mocknet: m,
57
peer: p,
58
ps: ps,
57
- mux: inet.Mux{Handlers: inet.StreamHandlerMap{}},
59
+ mux: mux.Mux{Handlers: inet.StreamHandlerMap{}},
60
cg: ctxgroup.WithContext(ctx),
61
62
connsByPeer: map[peer.ID]map[*conn]struct{}{},
@@ -65,7 +67,7 @@ func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
67
68
// setup a conn handler that immediately "asks the other side about them"
69
// this is ProtocolIdentify.
68
- n.ids = inet.NewIDService(n)
70
+ n.ids = ids.NewIDService(n)
71
72
return n, nil
73
}
@@ -338,6 +340,6 @@ func (pn *peernet) SetHandler(p inet.ProtocolID, h inet.StreamHandler) {
340
pn.mux.SetHandler(p, h)
341
}
342
341
-func (pn *peernet) IdentifyProtocol() *inet.IDService {
343
+func (pn *peernet) IdentifyProtocol() *ids.IDService {
344
return pn.ids
345
}
net/services/identify/id.go
renamed
+21
-14
@@ -1,15 +1,22 @@
1
-package net
1
+package identify
2
3
import (
4
"sync"
5
6
- handshake "github.com/jbenet/go-ipfs/net/handshake"
7
- pb "github.com/jbenet/go-ipfs/net/handshake/pb"
8
-
6
ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
7
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
8
+
9
+ inet "github.com/jbenet/go-ipfs/net"
10
+ handshake "github.com/jbenet/go-ipfs/net/handshake"
11
+ pb "github.com/jbenet/go-ipfs/net/handshake/pb"
12
+ eventlog "github.com/jbenet/go-ipfs/util/eventlog"
13
)
14
15
+var log = eventlog.Logger("net/identify")
16
+
17
+// ProtocolIdentify is the ProtocolID of the Identify Service.
18
+const ProtocolIdentify inet.ProtocolID = "/ipfs/identify"
19
+
20
// IDService is a structure that implements ProtocolIdentify.
21
// It is a trivial service that gives the other peer some
22
// useful information about the local peer. A sort of hello.
@@ -19,24 +26,24 @@ import (
26
// * Our IPFS Agent Version
27
// * Our public Listen Addresses
28
type IDService struct {
22
- Network Network
29
+ Network inet.Network
30
31
// connections undergoing identification
32
// for wait purposes
26
- currid map[Conn]chan struct{}
33
+ currid map[inet.Conn]chan struct{}
34
currmu sync.RWMutex
35
}
36
30
-func NewIDService(n Network) *IDService {
37
+func NewIDService(n inet.Network) *IDService {
38
s := &IDService{
39
Network: n,
33
- currid: make(map[Conn]chan struct{}),
40
+ currid: make(map[inet.Conn]chan struct{}),
41
}
42
n.SetHandler(ProtocolIdentify, s.RequestHandler)
43
return s
44
}
45
39
-func (ids *IDService) IdentifyConn(c Conn) {
46
+func (ids *IDService) IdentifyConn(c inet.Conn) {
47
ids.currmu.Lock()
48
if wait, found := ids.currid[c]; found {
49
ids.currmu.Unlock()
@@ -70,7 +77,7 @@ func (ids *IDService) IdentifyConn(c Conn) {
77
close(ch) // release everyone waiting.
78
}
79
73
-func (ids *IDService) RequestHandler(s Stream) {
80
+func (ids *IDService) RequestHandler(s inet.Stream) {
81
defer s.Close()
82
c := s.Conn()
83
@@ -83,7 +90,7 @@ func (ids *IDService) RequestHandler(s Stream) {
90
c.RemotePeer(), c.RemoteMultiaddr())
91
}
92
86
-func (ids *IDService) ResponseHandler(s Stream) {
93
+func (ids *IDService) ResponseHandler(s inet.Stream) {
94
defer s.Close()
95
c := s.Conn()
96
@@ -100,7 +107,7 @@ func (ids *IDService) ResponseHandler(s Stream) {
107
c.RemotePeer(), c.RemoteMultiaddr())
108
}
109
103
-func (ids *IDService) populateMessage(mes *pb.Handshake3, c Conn) {
110
+func (ids *IDService) populateMessage(mes *pb.Handshake3, c inet.Conn) {
111
112
// set protocols this node is currently handling
113
protos := ids.Network.Protocols()
@@ -129,7 +136,7 @@ func (ids *IDService) populateMessage(mes *pb.Handshake3, c Conn) {
136
mes.H1 = handshake.NewHandshake1("", "")
137
}
138
132
-func (ids *IDService) consumeMessage(mes *pb.Handshake3, c Conn) {
139
+func (ids *IDService) consumeMessage(mes *pb.Handshake3, c inet.Conn) {
140
p := c.RemotePeer()
141
142
// mes.Protocols
@@ -164,7 +171,7 @@ func (ids *IDService) consumeMessage(mes *pb.Handshake3, c Conn) {
171
// This happens async so the connection can start to be used
172
// even if handshake3 knowledge is not necesary.
173
// Users **MUST** call IdentifyWait _after_ IdentifyConn
167
-func (ids *IDService) IdentifyWait(c Conn) <-chan struct{} {
174
+func (ids *IDService) IdentifyWait(c inet.Conn) <-chan struct{} {
175
ids.currmu.Lock()
176
ch, found := ids.currid[c]
177
ids.currmu.Unlock()
net/services/identify/id_test.go
renamed
+5
-24
@@ -1,4 +1,4 @@
1
-package net_test
1
+package identify_test
2
3
import (
4
"testing"
@@ -6,38 +6,19 @@ import (
6
7
inet "github.com/jbenet/go-ipfs/net"
8
handshake "github.com/jbenet/go-ipfs/net/handshake"
9
+ netutil "github.com/jbenet/go-ipfs/net/ipfsnet/util"
10
peer "github.com/jbenet/go-ipfs/peer"
10
- testutil "github.com/jbenet/go-ipfs/util/testutil"
11
12
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
14
)
15
16
-func GenNetwork(t *testing.T, ctx context.Context) inet.Network {
17
- p := testutil.RandPeerNetParamsOrFatal(t)
18
- ps := peer.NewPeerstore()
19
- ps.AddAddress(p.ID, p.Addr)
20
- ps.AddPubKey(p.ID, p.PubKey)
21
- ps.AddPrivKey(p.ID, p.PrivKey)
22
- n, err := inet.NewNetwork(ctx, ps.Addresses(p.ID), p.ID, ps)
23
- if err != nil {
24
- t.Fatal(err)
25
- }
26
- return n
27
-}
28
-
29
-func DivulgeAddresses(a, b inet.Network) {
30
- id := a.LocalPeer()
31
- addrs := a.Peerstore().Addresses(id)
32
- b.Peerstore().AddAddresses(id, addrs)
33
-}
34
-
16
func subtestIDService(t *testing.T, postDialWait time.Duration) {
17
18
// the generated networks should have the id service wired in.
19
ctx := context.Background()
39
- n1 := GenNetwork(t, ctx)
40
- n2 := GenNetwork(t, ctx)
20
+ n1 := netutil.GenNetwork(t, ctx)
21
+ n2 := netutil.GenNetwork(t, ctx)
22
23
n1p := n1.LocalPeer()
24
n2p := n2.LocalPeer()
@@ -46,7 +27,7 @@ func subtestIDService(t *testing.T, postDialWait time.Duration) {
27
testKnowsAddrs(t, n2, n1p, []ma.Multiaddr{}) // nothing
28
29
// have n2 tell n1, so we can dial...
49
- DivulgeAddresses(n2, n1)
30
+ netutil.DivulgeAddresses(n2, n1)
31
32
testKnowsAddrs(t, n1, n2p, n2.Peerstore().Addresses(n2p)) // has them
33
testKnowsAddrs(t, n2, n1p, []ma.Multiaddr{}) // nothing
net/services/mux/mux.go
renamed
+24
-10
@@ -1,4 +1,4 @@
1
-package net
1
+package mux
2
3
import (
4
"fmt"
@@ -6,11 +6,13 @@ import (
6
"sync"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
+
10
+ inet "github.com/jbenet/go-ipfs/net"
11
eventlog "github.com/jbenet/go-ipfs/util/eventlog"
12
lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
13
)
14
13
-var log = eventlog.Logger("network")
15
+var log = eventlog.Logger("net/mux")
16
17
// Mux provides simple stream multixplexing.
18
// It helps you precisely when:
@@ -30,16 +32,16 @@ var log = eventlog.Logger("network")
32
// WARNING: this datastructure IS NOT threadsafe.
33
// do not modify it once the network is using it.
34
type Mux struct {
33
- Default StreamHandler // handles unknown protocols.
34
- Handlers StreamHandlerMap
35
+ Default inet.StreamHandler // handles unknown protocols.
36
+ Handlers inet.StreamHandlerMap
37
38
sync.RWMutex
39
}
40
41
// Protocols returns the list of protocols this muxer has handlers for
40
-func (m *Mux) Protocols() []ProtocolID {
42
+func (m *Mux) Protocols() []inet.ProtocolID {
43
m.RLock()
42
- l := make([]ProtocolID, 0, len(m.Handlers))
44
+ l := make([]inet.ProtocolID, 0, len(m.Handlers))
45
for p := range m.Handlers {
46
l = append(l, p)
47
}
@@ -49,7 +51,7 @@ func (m *Mux) Protocols() []ProtocolID {
51
52
// ReadProtocolHeader reads the stream and returns the next Handler function
53
// according to the muxer encoding.
52
-func (m *Mux) ReadProtocolHeader(s io.Reader) (string, StreamHandler, error) {
54
+func (m *Mux) ReadProtocolHeader(s io.Reader) (string, inet.StreamHandler, error) {
55
// log.Error("ReadProtocolHeader")
56
name, err := ReadLengthPrefix(s)
57
if err != nil {
@@ -58,7 +60,7 @@ func (m *Mux) ReadProtocolHeader(s io.Reader) (string, StreamHandler, error) {
60
61
// log.Debug("ReadProtocolHeader got:", name)
62
m.RLock()
61
- h, found := m.Handlers[ProtocolID(name)]
63
+ h, found := m.Handlers[inet.ProtocolID(name)]
64
m.RUnlock()
65
66
switch {
@@ -80,7 +82,7 @@ func (m *Mux) String() string {
82
83
// SetHandler sets the protocol handler on the Network's Muxer.
84
// This operation is threadsafe.
83
-func (m *Mux) SetHandler(p ProtocolID, h StreamHandler) {
85
+func (m *Mux) SetHandler(p inet.ProtocolID, h inet.StreamHandler) {
86
log.Debugf("%s setting handler for protocol: %s (%d)", m, p, len(p))
87
m.Lock()
88
m.Handlers[p] = h
@@ -88,7 +90,8 @@ func (m *Mux) SetHandler(p ProtocolID, h StreamHandler) {
90
}
91
92
// Handle reads the next name off the Stream, and calls a function
91
-func (m *Mux) Handle(s Stream) {
93
+func (m *Mux) Handle(s inet.Stream) {
94
+
95
ctx := context.Background()
96
97
name, handler, err := m.ReadProtocolHeader(s)
@@ -133,3 +136,14 @@ func WriteLengthPrefix(w io.Writer, name string) error {
136
_, err := w.Write(s)
137
return err
138
}
139
+
140
+// WriteProtocolHeader defines how a protocol is written into the header of
141
+// a stream. This is so the muxer can multiplex between services.
142
+func WriteProtocolHeader(pr inet.ProtocolID, s inet.Stream) error {
143
+ if pr != "" { // only write proper protocol headers
144
+ if err := WriteLengthPrefix(s, string(pr)); err != nil {
145
+ return err
146
+ }
147
+ }
148
+ return nil
149
+}
net/services/mux/mux_test.go
renamed
+6
-4
@@ -1,8 +1,10 @@
1
-package net
1
+package mux
2
3
import (
4
"bytes"
5
"testing"
6
+
7
+ inet "github.com/jbenet/go-ipfs/net"
8
)
9
10
var testCases = map[string]string{
@@ -28,13 +30,13 @@ func TestHandler(t *testing.T) {
30
31
outs := make(chan string, 10)
32
31
- h := func(n string) func(s Stream) {
32
- return func(s Stream) {
33
+ h := func(n string) func(s inet.Stream) {
34
+ return func(s inet.Stream) {
35
outs <- n
36
}
37
}
38
37
- m := Mux{Handlers: StreamHandlerMap{}}
39
+ m := Mux{Handlers: inet.StreamHandlerMap{}}
40
m.Default = h("default")
41
m.Handlers["dht"] = h("bitswap")
42
// m.Handlers["ipfs"] = h("bitswap") // default!
routing/dht/dht_test.go
+2
-2
@@ -14,7 +14,7 @@ import (
14
dssync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
15
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
16
17
- inet "github.com/jbenet/go-ipfs/net"
17
+ ipfsnet "github.com/jbenet/go-ipfs/net/ipfsnet"
18
peer "github.com/jbenet/go-ipfs/peer"
19
routing "github.com/jbenet/go-ipfs/routing"
20
u "github.com/jbenet/go-ipfs/util"
@@ -49,7 +49,7 @@ func setupDHT(ctx context.Context, t *testing.T, addr ma.Multiaddr) *IpfsDHT {
49
peerstore.AddPubKey(p, pk)
50
peerstore.AddAddress(p, addr)
51
52
- n, err := inet.NewNetwork(ctx, []ma.Multiaddr{addr}, p, peerstore)
52
+ n, err := ipfsnet.NewNetwork(ctx, []ma.Multiaddr{addr}, p, peerstore)
53
if err != nil {
54
t.Fatal(err)
55
}