make net work with new stream + mux
Juan Batiz-Benet committed
Dec 16, 2014 at 04:43 UTC
6334e19374969f217dcd821f0c2c84d37a18700d
8 files changed
+183
-166
net/conn/interface.go
+10
-7
@@ -15,12 +15,7 @@ import (
15
// Map maps Keys (Peer.IDs) to Connections.
16
type Map map[u.Key]Conn
17
18
-// Conn is a generic message-based Peer-to-Peer connection.
19
-type Conn interface {
20
-
21
- // ID is an identifier unique to this connection.
22
- ID() string
23
-
18
+type PeerConn interface {
19
// LocalMultiaddr is the Multiaddr on this side
20
LocalMultiaddr() ma.Multiaddr
21
@@ -32,8 +27,16 @@ type Conn interface {
27
28
// RemotePeer is the Peer on the remote side
29
RemotePeer() peer.Peer
30
+}
31
+
32
+// Conn is a generic message-based Peer-to-Peer connection.
33
+type Conn interface {
34
+ PeerConn
35
+
36
+ // ID is an identifier unique to this connection.
37
+ ID() string
38
36
- // net.Conn, cause duplicates.
39
+ // can't just say "net.Conn" cause we have duplicate methods.
40
LocalAddr() net.Addr
41
RemoteAddr() net.Addr
42
SetDeadline(t time.Time) error
net/interface.go
+63
-44
@@ -1,56 +1,84 @@
1
package net
2
3
import (
4
- "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
4
+ "io"
5
+
6
conn "github.com/jbenet/go-ipfs/net/conn"
6
- msg "github.com/jbenet/go-ipfs/net/message"
7
- mux "github.com/jbenet/go-ipfs/net/mux"
8
- srv "github.com/jbenet/go-ipfs/net/service"
7
+ swarm "github.com/jbenet/go-ipfs/net/swarm2"
8
peer "github.com/jbenet/go-ipfs/peer"
10
- ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
9
10
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
12
)
13
15
-// Network is the interface IPFS uses for connecting to the world.
16
-type Network interface {
17
- ctxc.ContextCloser
18
-
19
- // Listen handles incoming connections on given Multiaddr.
20
- // Listen(*ma.Muliaddr) error
21
- // TODO: for now, only listen on addrs in local peer when initializing.
22
-
23
- // LocalPeer returns the local peer associated with this network
24
- LocalPeer() peer.Peer
14
+// Stream represents a bidirectional channel between two agents in
15
+// the IPFS network. "agent" is as granular as desired, potentially
16
+// being a "request -> reply" pair, or whole protocols.
17
+// Streams are backed by SPDY streams underneath the hood.
18
+type Stream interface {
19
+ io.Reader
20
+ io.Writer
21
+ io.Closer
22
+
23
+ // Conn returns the connection this stream is part of.
24
+ Conn() Conn
25
+}
26
26
- // DialPeer attempts to establish a connection to a given peer
27
- DialPeer(context.Context, peer.Peer) error
27
+// StreamHandler is the function protocols who wish to listen to
28
+// incoming streams must implement.
29
+type StreamHandler func(Stream)
30
29
- // ClosePeer connection to peer
30
- ClosePeer(peer.Peer) error
31
+// Conn is a connection to a remote peer. It multiplexes streams.
32
+// Usually there is no need to use a Conn directly, but it may
33
+// be useful to get information about the peer on the other side:
34
+// stream.Conn().RemotePeer()
35
+type Conn interface {
36
+ conn.PeerConn
37
32
- // Connectedness returns a state signaling connection capabilities
33
- Connectedness(peer.Peer) Connectedness
38
+ // NewStream constructs a new Stream directly connected to p.
39
+ NewStream(p peer.Peer) (Stream, error)
40
+}
41
35
- // GetProtocols returns the protocols registered in the network.
36
- GetProtocols() *mux.ProtocolMap
42
+// Mux provides simple stream multixplexing.
43
+// It helps you precisely when:
44
+// * You have many streams
45
+// * You have function handlers
46
+//
47
+// It contains the handlers for each protocol accepted.
48
+// It dispatches handlers for streams opened by remote peers.
49
+//
50
+// We use a totally ad-hoc encoding:
51
+// <1 byte length in bytes><string name>
52
+// So "bitswap" is 0x0762697473776170
53
+//
54
+// NOTE: only the dialer specifies this muxing line.
55
+// This is because we're using Streams :)
56
+//
57
+// WARNING: this datastructure IS NOT threadsafe.
58
+// do not modify it once the network is using it.
59
+type Mux struct {
60
+ Default StreamHandler // handles unknown protocols.
61
+ Handlers map[string]StreamHandler
62
+}
63
38
- // GetPeerList returns the list of peers currently connected in this network.
39
- GetPeerList() []peer.Peer
64
+// Network is the interface IPFS uses for connecting to the world.
65
+// It dials and listens for connections. it uses a Swarm to pool
66
+// connnections (see swarm pkg, and peerstream.Swarm). Connections
67
+// are encrypted with a TLS-like protocol.
68
+type Network interface {
69
+ Dialer
70
+ io.Closer
71
41
- // GetConnections returns the list of connections currently open in this network.
42
- GetConnections() []conn.Conn
72
+ // NewStream returns a new stream to given peer p.
73
+ // If there is no connection to p, attempts to create one.
74
+ NewStream(p peer.Peer) (Stream, error)
75
44
- // GetBandwidthTotals returns the total number of bytes passed through
45
- // the network since it was instantiated
46
- GetBandwidthTotals() (uint64, uint64)
76
+ // Swarm returns the connection Swarm
77
+ Swarm() *swarm.Swarm
78
48
- // GetMessageCounts returns the total number of messages passed through
79
+ // BandwidthTotals returns the total number of bytes passed through
80
// the network since it was instantiated
50
- GetMessageCounts() (uint64, uint64)
51
-
52
- // SendMessage sends given Message out
53
- SendMessage(msg.NetMessage) error
81
+ BandwidthTotals() (uint64, uint64)
82
83
// ListenAddresses returns a list of addresses at which this network listens.
84
ListenAddresses() []ma.Multiaddr
@@ -61,15 +89,6 @@ type Network interface {
89
InterfaceListenAddresses() ([]ma.Multiaddr, error)
90
}
91
64
-// Sender interface for network services.
65
-type Sender srv.Sender
66
-
67
-// Handler interface for network services.
68
-type Handler srv.Handler
69
-
70
-// Service interface for network resources.
71
-type Service srv.Service
72
-
92
// Dialer represents a service that can dial out to peers
93
// (this is usually just a Network, but other services may not need the whole
94
// stack, and thus it becomes easier to mock)
net/mux.go
+2
-26
@@ -1,4 +1,4 @@
1
-package mux
1
+package net
2
3
import (
4
"errors"
@@ -6,36 +6,12 @@ import (
6
"io"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
- swarm "github.com/jbenet/go-ipfs/net/swarm2"
9
eventlog "github.com/jbenet/go-ipfs/util/eventlog"
10
lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
11
)
12
13
var log = eventlog.Logger("mux2")
14
16
-// Mux provides simple stream multixplexing.
17
-// It helps you precisely when:
18
-// * You have many streams
19
-// * You have function handlers
20
-//
21
-// We use a totally ad-hoc encoding:
22
-//
23
-// <1 byte length in bytes><string name>
24
-//
25
-// So "bitswap" is 0x0762697473776170
26
-//
27
-// NOTE: only the dialer specifies this muxing line.
28
-// This is because we're using Streams :)
29
-//
30
-// WARNING: this datastructure IS NOT threadsafe.
31
-// do not modify it once it's begun serving.
32
-type Mux struct {
33
- Default StreamHandler
34
- Handlers map[string]StreamHandler
35
-}
36
-
37
-type StreamHandler func(s *swarm.Stream)
38
-
15
// NextName reads the stream and returns the next protocol name
16
// according to the muxer encoding.
17
func (m *Mux) NextName(s io.Reader) (string, error) {
@@ -78,7 +54,7 @@ func (m *Mux) NextHandler(s io.Reader) (string, StreamHandler, error) {
54
}
55
56
// Handle reads the next name off the Stream, and calls a function
81
-func (m *Mux) Handle(s *swarm.Stream) {
57
+func (m *Mux) Handle(s Stream) {
58
ctx := context.Background()
59
60
name, handler, err := m.NextHandler(s)
net/mux_test.go
+3
-5
@@ -1,10 +1,8 @@
1
-package mux
1
+package net
2
3
import (
4
"bytes"
5
"testing"
6
-
7
- swarm "github.com/jbenet/go-ipfs/net/swarm2"
6
)
7
8
var testCases = map[string]string{
@@ -30,8 +28,8 @@ func TestHandler(t *testing.T) {
28
29
outs := make(chan string, 10)
30
33
- h := func(n string) func(s *swarm.Stream) {
34
- return func(s *swarm.Stream) {
31
+ h := func(n string) func(s Stream) {
32
+ return func(s Stream) {
33
outs <- n
34
}
35
}
net/net.go
+93
-80
@@ -2,66 +2,112 @@
2
package net
3
4
import (
5
- conn "github.com/jbenet/go-ipfs/net/conn"
6
- msg "github.com/jbenet/go-ipfs/net/message"
7
- mux "github.com/jbenet/go-ipfs/net/mux"
8
- swarm "github.com/jbenet/go-ipfs/net/swarm"
5
+ swarm "github.com/jbenet/go-ipfs/net/swarm2"
6
peer "github.com/jbenet/go-ipfs/peer"
7
util "github.com/jbenet/go-ipfs/util"
11
- ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
8
9
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
+ ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
11
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
12
)
13
17
-// IpfsNetwork implements the Network interface,
18
-type IpfsNetwork struct {
14
+type stream swarm.Stream
15
20
- // local peer
21
- local peer.Peer
16
+func (s *stream) SwarmStream() *swarm.Stream {
17
+ return (*swarm.Stream)(s)
18
+}
19
23
- // protocol multiplexing
24
- muxer *mux.Muxer
20
+// Conn returns the connection this stream is part of.
21
+func (s *stream) Conn() Conn {
22
+ c := s.SwarmStream().Conn()
23
+ return (*conn_)(c)
24
+}
25
26
- // peer connection multiplexing
27
- swarm *swarm.Swarm
26
+// Conn returns the connection this stream is part of.
27
+func (s *stream) Close() error {
28
+ return s.SwarmStream().Close()
29
+}
30
29
- // network context closer
30
- ctxc.ContextCloser
31
+// Read reads bytes from a stream.
32
+func (s *stream) Read(p []byte) (n int, err error) {
33
+ return s.SwarmStream().Read(p)
34
}
35
33
-// NewIpfsNetwork is the structure that implements the network interface
34
-func NewIpfsNetwork(ctx context.Context, listen []ma.Multiaddr, local peer.Peer,
35
- peers peer.Peerstore, pmap *mux.ProtocolMap) (*IpfsNetwork, error) {
36
+// Write writes bytes to a stream, flushing for each call.
37
+func (s *stream) Write(p []byte) (n int, err error) {
38
+ return s.SwarmStream().Write(p)
39
+}
40
37
- in := &IpfsNetwork{
38
- local: local,
39
- muxer: mux.NewMuxer(ctx, *pmap),
40
- ContextCloser: ctxc.NewContextCloser(ctx, nil),
41
- }
41
+type conn_ swarm.Conn
42
43
- var err error
44
- in.swarm, err = swarm.NewSwarm(ctx, listen, local, peers)
43
+func (c *conn_) SwarmConn() *swarm.Conn {
44
+ return (*swarm.Conn)(c)
45
+}
46
+
47
+func (c *conn_) NewStream(p peer.Peer) (Stream, error) {
48
+ s, err := (*swarm.Conn)(c).NewStream()
49
if err != nil {
46
- in.Close()
50
return nil, err
51
}
52
+ return (*stream)(s), nil
53
+}
54
50
- in.AddCloserChild(in.swarm)
51
- in.AddCloserChild(in.muxer)
55
+// LocalMultiaddr is the Multiaddr on this side
56
+func (c *conn_) LocalMultiaddr() ma.Multiaddr {
57
+ return c.SwarmConn().LocalMultiaddr()
58
+}
59
53
- // remember to wire components together.
54
- in.muxer.Pipe.ConnectTo(in.swarm.Pipe)
60
+// LocalPeer is the Peer on our side of the connection
61
+func (c *conn_) LocalPeer() peer.Peer {
62
+ return c.SwarmConn().LocalPeer()
63
+}
64
56
- return in, nil
65
+// RemoteMultiaddr is the Multiaddr on the remote side
66
+func (c *conn_) RemoteMultiaddr() ma.Multiaddr {
67
+ return c.SwarmConn().RemoteMultiaddr()
68
}
69
59
-// Listen handles incoming connections on given Multiaddr.
60
-// func (n *IpfsNetwork) Listen(*ma.Muliaddr) error {}
70
+// RemotePeer is the Peer on the remote side
71
+func (c *conn_) RemotePeer() peer.Peer {
72
+ return c.SwarmConn().RemotePeer()
73
+}
74
+
75
+// network implements the Network interface,
76
+type network struct {
77
+ local peer.Peer // local peer
78
+ mux Mux // protocol multiplexing
79
+ swarm *swarm.Swarm // peer connection multiplexing
80
+
81
+ cg ctxgroup.ContextGroup // for Context closing
82
+}
83
+
84
+// NewConn is the structure that implements the network interface
85
+func NewConn(ctx context.Context, listen []ma.Multiaddr, local peer.Peer,
86
+ peers peer.Peerstore, m Mux) (*network, error) {
87
+
88
+ s, err := swarm.NewSwarm(ctx, listen, local, peers)
89
+ if err != nil {
90
+ return nil, err
91
+ }
92
+
93
+ n := &network{
94
+ local: local,
95
+ swarm: s,
96
+ mux: m,
97
+ cg: ctxgroup.WithContext(ctx),
98
+ }
99
+
100
+ s.SetStreamHandler(func(s *swarm.Stream) {
101
+ m.Handle((*stream)(s))
102
+ })
103
+
104
+ n.cg.AddChildGroup(s.CtxGroup())
105
+ return n, nil
106
+}
107
108
// DialPeer attempts to establish a connection to a given peer.
109
// Respects the context.
64
-func (n *IpfsNetwork) DialPeer(ctx context.Context, p peer.Peer) error {
110
+func (n *network) DialPeer(ctx context.Context, p peer.Peer) error {
111
err := util.ContextDo(ctx, func() error {
112
_, err := n.swarm.Dial(p)
113
return err
@@ -70,72 +116,39 @@ func (n *IpfsNetwork) DialPeer(ctx context.Context, p peer.Peer) error {
116
}
117
118
// LocalPeer the network's LocalPeer
73
-func (n *IpfsNetwork) LocalPeer() peer.Peer {
119
+func (n *network) LocalPeer() peer.Peer {
120
return n.swarm.LocalPeer()
121
}
122
123
// ClosePeer connection to peer
78
-func (n *IpfsNetwork) ClosePeer(p peer.Peer) error {
124
+func (n *network) ClosePeer(p peer.Peer) error {
125
return n.swarm.CloseConnection(p)
126
}
127
82
-// IsConnected returns whether a connection to given peer exists.
83
-func (n *IpfsNetwork) IsConnected(p peer.Peer) bool {
84
- return n.swarm.GetConnection(p.ID()) != nil
85
-}
86
-
87
-// GetProtocols returns the protocols registered in the network.
88
-func (n *IpfsNetwork) GetProtocols() *mux.ProtocolMap {
89
- // copy over because this map should be read only.
90
- pmap := mux.ProtocolMap{}
91
- for id, proto := range n.muxer.Protocols {
92
- pmap[id] = proto
93
- }
94
- return &pmap
95
-}
96
-
97
-// SendMessage sends given Message out
98
-func (n *IpfsNetwork) SendMessage(m msg.NetMessage) error {
99
- n.swarm.Outgoing <- m
100
- return nil
101
-}
102
-
103
-// GetPeerList returns the networks list of connected peers
104
-func (n *IpfsNetwork) GetPeerList() []peer.Peer {
105
- return n.swarm.GetPeerList()
106
-}
107
-
108
-// GetConnections returns the networks list of open connections
109
-func (n *IpfsNetwork) GetConnections() []conn.Conn {
110
- return n.swarm.Connections()
111
-}
112
-
113
-// GetBandwidthTotals returns the total amount of bandwidth transferred
114
-func (n *IpfsNetwork) GetBandwidthTotals() (in uint64, out uint64) {
115
- return n.muxer.GetBandwidthTotals()
116
-}
117
-
118
-// GetBandwidthTotals returns the total amount of messages transferred
119
-func (n *IpfsNetwork) GetMessageCounts() (in uint64, out uint64) {
120
- return n.muxer.GetMessageCounts()
128
+// BandwidthTotals returns the total amount of bandwidth transferred
129
+func (n *network) BandwidthTotals() (in uint64, out uint64) {
130
+ // need to implement this. probably best to do it in swarm this time.
131
+ // need a "metrics" object
132
+ return 0, 0
133
}
134
135
// ListenAddresses returns a list of addresses at which this network listens.
124
-func (n *IpfsNetwork) ListenAddresses() []ma.Multiaddr {
136
+func (n *network) ListenAddresses() []ma.Multiaddr {
137
return n.swarm.ListenAddresses()
138
}
139
140
// InterfaceListenAddresses returns a list of addresses at which this network
141
// listens. It expands "any interface" addresses (/ip4/0.0.0.0, /ip6/::) to
142
// use the known local interfaces.
131
-func (n *IpfsNetwork) InterfaceListenAddresses() ([]ma.Multiaddr, error) {
132
- return n.swarm.InterfaceListenAddresses()
143
+func (n *network) InterfaceListenAddresses() ([]ma.Multiaddr, error) {
144
+ return swarm.InterfaceListenAddresses(n.swarm)
145
}
146
147
// Connectedness returns a state signaling connection capabilities
148
// For now only returns Connecter || NotConnected. Expand into more later.
137
-func (n *IpfsNetwork) Connectedness(p peer.Peer) Connectedness {
138
- if n.swarm.GetConnection(p.ID()) != nil {
149
+func (n *network) Connectedness(p peer.Peer) Connectedness {
150
+ c := n.swarm.ConnectionsToPeer(p)
151
+ if c != nil && len(c) < 1 {
152
return Connected
153
}
154
return NotConnected
net/net_test.go
deleted
-1
@@ -1 +0,0 @@
1
-package net
net/swarm2/swarm.go
+7
-2
@@ -50,6 +50,11 @@ func (s *Swarm) teardown() error {
50
return s.swarm.Close()
51
}
52
53
+// CtxGroup returns the Context Group of the swarm
54
+func (s *Swarm) CtxGroup() ctxgroup.ContextGroup {
55
+ return s.cg
56
+}
57
+
58
// Close stops the Swarm.
59
func (s *Swarm) Close() error {
60
return s.cg.Close()
@@ -106,8 +111,8 @@ func (s *Swarm) CloseConnection(p peer.Peer) error {
111
return nil
112
}
113
109
-// GetPeerList returns a copy of the set of peers swarm is connected to.
110
-func (s *Swarm) GetPeerList() []peer.Peer {
114
+// Peers returns a copy of the set of peers swarm is connected to.
115
+func (s *Swarm) Peers() []peer.Peer {
116
conns := s.Connections()
117
118
seen := make(map[peer.Peer]struct{})
net/swarm2/swarm_stream.go
+5
-1
@@ -22,19 +22,23 @@ func (s *Stream) Conn() *Conn {
22
return (*Conn)(s.Stream().Conn())
23
}
24
25
-// Write writes bytes to a stream, calling write data for each call.
25
+// Wait waits for the stream to receive a reply.
26
func (s *Stream) Wait() error {
27
return s.Stream().Wait()
28
}
29
30
+// Read reads bytes from a stream.
31
func (s *Stream) Read(p []byte) (n int, err error) {
32
return s.Stream().Read(p)
33
}
34
35
+// Write writes bytes to a stream, flushing for each call.
36
func (s *Stream) Write(p []byte) (n int, err error) {
37
return s.Stream().Write(p)
38
}
39
40
+// Close closes the stream, indicating this side is finished
41
+// with the stream.
42
func (s *Stream) Close() error {
43
return s.Stream().Close()
44
}