moved versionhandshake to conn
Juan Batiz-Benet committed
Oct 18, 2014 at 04:50 UTC
f2e428d41b3f380901a4887e04b83b76f20cd1e3
3 files changed
+75
-58
net/conn/conn.go
+17
-4
@@ -2,6 +2,7 @@ package conn
2
3
import (
4
"fmt"
5
+ "time"
6
7
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8
msgio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
@@ -14,11 +15,16 @@ import (
15
16
var log = u.Logger("conn")
17
17
-// ChanBuffer is the size of the buffer in the Conn Chan
18
-const ChanBuffer = 10
18
+const (
19
+ // ChanBuffer is the size of the buffer in the Conn Chan
20
+ ChanBuffer = 10
21
20
-// 1 MB
21
-const MaxMessageSize = 1 << 20
22
+ // MaxMessageSize is the size of the largest single message
23
+ MaxMessageSize = 1 << 20 // 1 MB
24
+
25
+ // HandshakeTimeout for when nodes first connect
26
+ HandshakeTimeout = time.Second * 5
27
+)
28
29
// msgioPipe is a pipe using msgio channels.
30
type msgioPipe struct {
@@ -62,6 +68,13 @@ func newSingleConn(ctx context.Context, local, remote *peer.Peer,
68
go conn.msgio.outgoing.WriteTo(maconn)
69
go conn.msgio.incoming.ReadFrom(maconn, MaxMessageSize)
70
71
+ // version handshake
72
+ ctxT, _ := context.WithTimeout(ctx, HandshakeTimeout)
73
+ if err := VersionHandshake(ctxT, conn); err != nil {
74
+ conn.Close()
75
+ return nil, fmt.Errorf("Version handshake: %s", err)
76
+ }
77
+
78
return conn, nil
79
}
80
net/conn/handshake.go
new
+58
@@ -0,0 +1,58 @@
1
+package conn
2
+
3
+import (
4
+ "errors"
5
+ "fmt"
6
+
7
+ handshake "github.com/jbenet/go-ipfs/net/handshake"
8
+
9
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
+ proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
11
+)
12
+
13
+// VersionHandshake exchanges local and remote versions and compares them
14
+// closes remote and returns an error in case of major difference
15
+func VersionHandshake(ctx context.Context, c Conn) error {
16
+ rpeer := c.RemotePeer()
17
+ lpeer := c.LocalPeer()
18
+
19
+ var remoteH, localH *handshake.Handshake1
20
+ localH = handshake.CurrentHandshake()
21
+
22
+ myVerBytes, err := proto.Marshal(localH)
23
+ if err != nil {
24
+ return err
25
+ }
26
+
27
+ c.Out() <- myVerBytes
28
+ log.Debug("Sent my version (%s) to %s", localH, rpeer)
29
+
30
+ select {
31
+ case <-ctx.Done():
32
+ return ctx.Err()
33
+
34
+ case <-c.Done():
35
+ return errors.New("remote closed connection during version exchange")
36
+
37
+ case data, ok := <-c.In():
38
+ if !ok {
39
+ return fmt.Errorf("error retrieving from conn: %v", rpeer)
40
+ }
41
+
42
+ remoteH = new(handshake.Handshake1)
43
+ err = proto.Unmarshal(data, remoteH)
44
+ if err != nil {
45
+ return fmt.Errorf("could not decode remote version: %q", err)
46
+ }
47
+
48
+ log.Debug("Received remote version (%s) from %s", remoteH, rpeer)
49
+ }
50
+
51
+ if err := handshake.Compatible(localH, remoteH); err != nil {
52
+ log.Info("%s (%s) incompatible version with %s (%s)", lpeer, localH, rpeer, remoteH)
53
+ return err
54
+ }
55
+
56
+ log.Debug("%s version handshake compatible %s", lpeer, rpeer)
57
+ return nil
58
+}
net/swarm/conn.go
-54
@@ -5,10 +5,8 @@ import (
5
"fmt"
6
7
conn "github.com/jbenet/go-ipfs/net/conn"
8
- handshake "github.com/jbenet/go-ipfs/net/handshake"
8
msg "github.com/jbenet/go-ipfs/net/message"
9
11
- proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
10
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11
)
12
@@ -93,10 +91,6 @@ func (s *Swarm) connSetup(c conn.Conn) error {
91
// addresses should be figured out through the DHT.
92
// c.Remote.AddAddress(c.Conn.RemoteMultiaddr())
93
96
- if err := s.connVersionExchange(c); err != nil {
97
- return fmt.Errorf("Conn version exchange error: %v", err)
98
- }
99
-
94
// add to conns
95
s.connsLock.Lock()
96
if _, ok := s.conns[c.RemotePeer().Key()]; ok {
@@ -113,54 +107,6 @@ func (s *Swarm) connSetup(c conn.Conn) error {
107
return nil
108
}
109
116
-// connVersionExchange exchanges local and remote versions and compares them
117
-// closes remote and returns an error in case of major difference
118
-func (s *Swarm) connVersionExchange(r conn.Conn) error {
119
- rpeer := r.RemotePeer()
120
-
121
- var remoteH, localH *handshake.Handshake1
122
- localH = handshake.CurrentHandshake()
123
-
124
- myVerBytes, err := proto.Marshal(localH)
125
- if err != nil {
126
- return err
127
- }
128
-
129
- r.Out() <- myVerBytes
130
- log.Debug("Sent my version(%s) [to = %s]", localH, rpeer)
131
-
132
- select {
133
- case <-s.ctx.Done():
134
- return s.ctx.Err()
135
-
136
- // case <-remote.Done():
137
- // return errors.New("remote closed connection during version exchange")
138
-
139
- case data, ok := <-r.In():
140
- if !ok {
141
- return fmt.Errorf("Error retrieving from conn: %v", rpeer)
142
- }
143
-
144
- remoteH = new(handshake.Handshake1)
145
- err = proto.Unmarshal(data, remoteH)
146
- if err != nil {
147
- s.Close()
148
- return fmt.Errorf("connSetup: could not decode remote version: %q", err)
149
- }
150
-
151
- log.Debug("Received remote version(%s) [from = %s]", remoteH, rpeer)
152
- }
153
-
154
- if err := handshake.Compatible(localH, remoteH); err != nil {
155
- log.Info("%s (%s) incompatible version with %s (%s)", s.local, localH, rpeer, remoteH)
156
- r.Close()
157
- return err
158
- }
159
-
160
- log.Debug("[peer: %s] Version compatible", rpeer)
161
- return nil
162
-}
163
-
110
// Handles the unwrapping + sending of messages to the right connection.
111
func (s *Swarm) fanOut() {
112
for {