refactor to allow use of mes_listener outside of dht
Jeromy committed
Aug 23, 2014 at 22:21 UTC
c5e7273cabd899b02e7e973611164349e646721b
13 files changed
+324
-122
bitswap/bitswap.go
+57
-5
@@ -1,8 +1,10 @@
1
package bitswap
2
3
import (
4
+ "code.google.com/p/goprotobuf/proto"
5
blocks "github.com/jbenet/go-ipfs/blocks"
6
peer "github.com/jbenet/go-ipfs/peer"
7
+ routing "github.com/jbenet/go-ipfs/routing"
8
swarm "github.com/jbenet/go-ipfs/swarm"
9
u "github.com/jbenet/go-ipfs/util"
10
@@ -27,12 +29,18 @@ type BitSwap struct {
29
peer *peer.Peer
30
31
// net holds the connections to all peers.
30
- net swarm.Network
32
+ net swarm.Network
33
+ meschan *swarm.Chan
34
35
// datastore is the local database
36
// Ledgers of known
37
datastore ds.Datastore
38
39
+ // routing interface for communication
40
+ routing routing.IpfsRouting
41
+
42
+ listener *swarm.MesListener
43
+
44
// partners is a map of currently active bitswap relationships.
45
// The Ledger has the peer.ID, and the peer connection works through net.
46
// Ledgers of known relationships (active or inactive) stored in datastore.
@@ -44,27 +52,71 @@ type BitSwap struct {
52
53
// wantList is the set of keys we want values for. a map for fast lookups.
54
wantList KeySet
55
+
56
+ haltChan chan struct{}
57
}
58
59
// NewBitSwap creates a new BitSwap instance. It does not check its parameters.
50
-func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore) *BitSwap {
51
- return &BitSwap{
60
+func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore, r routing.IpfsRouting) *BitSwap {
61
+ bs := &BitSwap{
62
peer: p,
63
net: net,
64
datastore: d,
65
partners: LedgerMap{},
66
wantList: KeySet{},
67
+ routing: r,
68
+ meschan: net.GetChannel(swarm.PBWrapper_BITSWAP),
69
+ haltChan: make(chan struct{}),
70
}
71
+
72
+ go bs.handleMessages()
73
+ return bs
74
}
75
76
// GetBlock attempts to retrieve a particular block from peers, within timeout.
61
-func (s *BitSwap) GetBlock(k u.Key, timeout time.Time) (
77
+func (bs *BitSwap) GetBlock(k u.Key, timeout time.Time) (
78
*blocks.Block, error) {
79
+ begin := time.Now()
80
+ _, err := bs.routing.FindProviders(k, timeout)
81
+ if err != nil {
82
+ u.PErr("GetBlock error: %s\n", err)
83
+ return
84
+ }
85
+ tleft := timeout.Sub(time.Now().Sub(begin))
86
return nil, errors.New("not implemented")
87
}
88
89
+func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byte, error) {
90
+ mes := new(PBMessage)
91
+ mes.Id = proto.Uint64(swarm.GenerateID())
92
+ mes.Key = proto.String(k)
93
+ typ := PBMessage_GET_BLOCK
94
+ mes.Type = &typ
95
+
96
+ after := time.After(timeout)
97
+ resp := bs.listener.Listen(mes.GetId(), 1, timeout)
98
+ smes := swarm.NewMessage(p, mes)
99
+ bs.meschan.Outgoing <- smes
100
+
101
+ select {
102
+ case resp_mes := <-resp:
103
+ case <-after:
104
+ u.PErr("getBlock for '%s' timed out.", k)
105
+ return nil, u.ErrTimeout
106
+ }
107
+}
108
+
109
// HaveBlock announces the existance of a block to BitSwap, potentially sending
110
// it to peers (Partners) whose WantLists include it.
68
-func (s *BitSwap) HaveBlock(k u.Key) (*blocks.Block, error) {
111
+func (bs *BitSwap) HaveBlock(k u.Key) (*blocks.Block, error) {
112
return nil, errors.New("not implemented")
113
}
114
+
115
+func (bs *BitSwap) handleMessages() {
116
+ for {
117
+ select {
118
+ case mes := bs.meschan.Incoming:
119
+ case <-bs.haltChan:
120
+ }
121
+ }
122
+}
bitswap/message.pb.go
new
+103
@@ -0,0 +1,103 @@
1
+// Code generated by protoc-gen-go.
2
+// source: message.proto
3
+// DO NOT EDIT!
4
+
5
+/*
6
+Package bitswap is a generated protocol buffer package.
7
+
8
+It is generated from these files:
9
+ message.proto
10
+
11
+It has these top-level messages:
12
+ PBMessage
13
+*/
14
+package bitswap
15
+
16
+import proto "code.google.com/p/goprotobuf/proto"
17
+import math "math"
18
+
19
+// Reference imports to suppress errors if they are not otherwise used.
20
+var _ = proto.Marshal
21
+var _ = math.Inf
22
+
23
+type PBMessage_MessageType int32
24
+
25
+const (
26
+ PBMessage_GET_BLOCK PBMessage_MessageType = 0
27
+)
28
+
29
+var PBMessage_MessageType_name = map[int32]string{
30
+ 0: "GET_BLOCK",
31
+}
32
+var PBMessage_MessageType_value = map[string]int32{
33
+ "GET_BLOCK": 0,
34
+}
35
+
36
+func (x PBMessage_MessageType) Enum() *PBMessage_MessageType {
37
+ p := new(PBMessage_MessageType)
38
+ *p = x
39
+ return p
40
+}
41
+func (x PBMessage_MessageType) String() string {
42
+ return proto.EnumName(PBMessage_MessageType_name, int32(x))
43
+}
44
+func (x *PBMessage_MessageType) UnmarshalJSON(data []byte) error {
45
+ value, err := proto.UnmarshalJSONEnum(PBMessage_MessageType_value, data, "PBMessage_MessageType")
46
+ if err != nil {
47
+ return err
48
+ }
49
+ *x = PBMessage_MessageType(value)
50
+ return nil
51
+}
52
+
53
+type PBMessage struct {
54
+ Type *PBMessage_MessageType `protobuf:"varint,1,req,enum=bitswap.PBMessage_MessageType" json:"Type,omitempty"`
55
+ Id *uint64 `protobuf:"varint,2,req,name=id" json:"id,omitempty"`
56
+ Key *string `protobuf:"bytes,3,req,name=key" json:"key,omitempty"`
57
+ Value []byte `protobuf:"bytes,4,opt,name=value" json:"value,omitempty"`
58
+ Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
59
+ XXX_unrecognized []byte `json:"-"`
60
+}
61
+
62
+func (m *PBMessage) Reset() { *m = PBMessage{} }
63
+func (m *PBMessage) String() string { return proto.CompactTextString(m) }
64
+func (*PBMessage) ProtoMessage() {}
65
+
66
+func (m *PBMessage) GetType() PBMessage_MessageType {
67
+ if m != nil && m.Type != nil {
68
+ return *m.Type
69
+ }
70
+ return PBMessage_GET_BLOCK
71
+}
72
+
73
+func (m *PBMessage) GetId() uint64 {
74
+ if m != nil && m.Id != nil {
75
+ return *m.Id
76
+ }
77
+ return 0
78
+}
79
+
80
+func (m *PBMessage) GetKey() string {
81
+ if m != nil && m.Key != nil {
82
+ return *m.Key
83
+ }
84
+ return ""
85
+}
86
+
87
+func (m *PBMessage) GetValue() []byte {
88
+ if m != nil {
89
+ return m.Value
90
+ }
91
+ return nil
92
+}
93
+
94
+func (m *PBMessage) GetResponse() bool {
95
+ if m != nil && m.Response != nil {
96
+ return *m.Response
97
+ }
98
+ return false
99
+}
100
+
101
+func init() {
102
+ proto.RegisterEnum("bitswap.PBMessage_MessageType", PBMessage_MessageType_name, PBMessage_MessageType_value)
103
+}
bitswap/message.proto
new
+13
@@ -0,0 +1,13 @@
1
+package bitswap;
2
+
3
+message PBMessage {
4
+ enum MessageType {
5
+ GET_BLOCK = 0;
6
+ }
7
+
8
+ required MessageType Type = 1;
9
+ required uint64 id = 2;
10
+ required string key = 3;
11
+ optional bytes value = 4;
12
+ optional bool response = 5;
13
+}
routing/dht/dht.go
+18
-15
@@ -2,7 +2,7 @@ package dht
2
3
import (
4
"bytes"
5
- "errors"
5
+ "fmt"
6
"sync"
7
"time"
8
@@ -49,7 +49,7 @@ type IpfsDHT struct {
49
diaglock sync.Mutex
50
51
// listener is a server to register to listen for responses to messages
52
- listener *mesListener
52
+ listener *swarm.MesListener
53
}
54
55
// NewDHT creates a new DHT object with the given peer as the 'local' host
@@ -66,7 +66,7 @@ func NewDHT(p *peer.Peer, net swarm.Network) *IpfsDHT {
66
dht.routingTables[0] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*30)
67
dht.routingTables[1] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*100)
68
dht.routingTables[2] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Hour)
69
- dht.listener = newMesListener()
69
+ dht.listener = swarm.NewMesListener()
70
dht.birth = time.Now()
71
return dht
72
}
@@ -89,7 +89,7 @@ func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
89
// NOTE: this should be done better...
90
err = dht.Ping(npeer, time.Second*2)
91
if err != nil {
92
- return nil, errors.New("failed to ping newly connected peer\n")
92
+ return nil, fmt.Errorf("failed to ping newly connected peer: %s\n", err)
93
}
94
95
dht.Update(npeer)
@@ -132,19 +132,19 @@ func (dht *IpfsDHT) handleMessages() {
132
pmes.GetId(), mes.Peer.ID.Pretty())
133
switch pmes.GetType() {
134
case PBDHTMessage_GET_VALUE:
135
- dht.handleGetValue(mes.Peer, pmes)
135
+ go dht.handleGetValue(mes.Peer, pmes)
136
case PBDHTMessage_PUT_VALUE:
137
- dht.handlePutValue(mes.Peer, pmes)
137
+ go dht.handlePutValue(mes.Peer, pmes)
138
case PBDHTMessage_FIND_NODE:
139
- dht.handleFindPeer(mes.Peer, pmes)
139
+ go dht.handleFindPeer(mes.Peer, pmes)
140
case PBDHTMessage_ADD_PROVIDER:
141
- dht.handleAddProvider(mes.Peer, pmes)
141
+ go dht.handleAddProvider(mes.Peer, pmes)
142
case PBDHTMessage_GET_PROVIDERS:
143
- dht.handleGetProviders(mes.Peer, pmes)
143
+ go dht.handleGetProviders(mes.Peer, pmes)
144
case PBDHTMessage_PING:
145
- dht.handlePing(mes.Peer, pmes)
145
+ go dht.handlePing(mes.Peer, pmes)
146
case PBDHTMessage_DIAGNOSTIC:
147
- dht.handleDiagnostic(mes.Peer, pmes)
147
+ go dht.handleDiagnostic(mes.Peer, pmes)
148
default:
149
u.PErr("Recieved invalid message type")
150
}
@@ -162,7 +162,7 @@ func (dht *IpfsDHT) putValueToNetwork(p *peer.Peer, key string, value []byte) er
162
Type: PBDHTMessage_PUT_VALUE,
163
Key: key,
164
Value: value,
165
- ID: GenerateMessageID(),
165
+ ID: swarm.GenerateMessageID(),
166
}
167
168
mes := swarm.NewMessage(p, pmes.ToProtobuf())
@@ -242,6 +242,7 @@ func (dht *IpfsDHT) handlePutValue(p *peer.Peer, pmes *PBDHTMessage) {
242
}
243
244
func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *PBDHTMessage) {
245
+ u.DOut("[%s] Responding to ping from [%s]!\n", dht.self.ID.Pretty(), p.ID.Pretty())
246
resp := Message{
247
Type: pmes.GetType(),
248
Response: true,
@@ -328,6 +329,8 @@ func (dht *IpfsDHT) handleAddProvider(p *peer.Peer, pmes *PBDHTMessage) {
329
func (dht *IpfsDHT) Halt() {
330
dht.shutdown <- struct{}{}
331
dht.network.Close()
332
+ dht.providers.Halt()
333
+ dht.listener.Halt()
334
}
335
336
// NOTE: not yet finished, low priority
@@ -424,7 +427,7 @@ func (dht *IpfsDHT) getValueSingle(p *peer.Peer, key u.Key, timeout time.Duratio
427
Type: PBDHTMessage_GET_VALUE,
428
Key: string(key),
429
Value: []byte{byte(level)},
427
- ID: GenerateMessageID(),
430
+ ID: swarm.GenerateMessageID(),
431
}
432
responseChan := dht.listener.Listen(pmes.ID, 1, time.Minute)
433
@@ -539,7 +542,7 @@ func (dht *IpfsDHT) findPeerSingle(p *peer.Peer, id peer.ID, timeout time.Durati
542
pmes := Message{
543
Type: PBDHTMessage_FIND_NODE,
544
Key: string(id),
542
- ID: GenerateMessageID(),
545
+ ID: swarm.GenerateMessageID(),
546
Value: []byte{byte(level)},
547
}
548
@@ -575,7 +578,7 @@ func (dht *IpfsDHT) findProvidersSingle(p *peer.Peer, key u.Key, level int, time
578
pmes := Message{
579
Type: PBDHTMessage_GET_PROVIDERS,
580
Key: string(key),
578
- ID: GenerateMessageID(),
581
+ ID: swarm.GenerateMessageID(),
582
Value: []byte{byte(level)},
583
}
584
routing/dht/ext_test.go
+1
-1
@@ -153,7 +153,7 @@ func TestGetFailures(t *testing.T) {
153
req := Message{
154
Type: PBDHTMessage_GET_VALUE,
155
Key: "hello",
156
- ID: GenerateMessageID(),
156
+ ID: swarm.GenerateMessageID(),
157
Value: []byte{0},
158
}
159
fn.Chan.Incoming <- swarm.NewMessage(other, req.ToProtobuf())
routing/dht/providers.go
+4
@@ -81,3 +81,7 @@ func (pm *ProviderManager) GetProviders(k u.Key) []*peer.Peer {
81
pm.getprovs <- gp
82
return <-gp.resp
83
}
84
+
85
+func (pm *ProviderManager) Halt() {
86
+ pm.halt <- struct{}{}
87
+}
routing/dht/routing.go
+19
-81
@@ -4,8 +4,6 @@ import (
4
"bytes"
5
"encoding/json"
6
"errors"
7
- "math/rand"
8
- "sync"
7
"time"
8
9
proto "code.google.com/p/goprotobuf/proto"
@@ -18,21 +16,6 @@ import (
16
u "github.com/jbenet/go-ipfs/util"
17
)
18
21
-// Pool size is the number of nodes used for group find/set RPC calls
22
-var PoolSize = 6
23
-
24
-// We put the 'K' in kademlia!
25
-var KValue = 10
26
-
27
-// Its in the paper, i swear
28
-var AlphaValue = 3
29
-
30
-// GenerateMessageID creates and returns a new message ID
31
-// TODO: determine a way of creating and managing message IDs
32
-func GenerateMessageID() uint64 {
33
- return (uint64(rand.Uint32()) << 32) | uint64(rand.Uint32())
34
-}
35
-
19
// This file implements the Routing interface for the IpfsDHT struct.
20
21
// Basic Put/Get
@@ -64,60 +47,6 @@ func (dht *IpfsDHT) PutValue(key u.Key, value []byte) {
47
}
48
}
49
67
-// A counter for incrementing a variable across multiple threads
68
-type counter struct {
69
- n int
70
- mut sync.RWMutex
71
-}
72
-
73
-func (c *counter) Increment() {
74
- c.mut.Lock()
75
- c.n++
76
- c.mut.Unlock()
77
-}
78
-
79
-func (c *counter) Decrement() {
80
- c.mut.Lock()
81
- c.n--
82
- c.mut.Unlock()
83
-}
84
-
85
-func (c *counter) Size() int {
86
- c.mut.RLock()
87
- defer c.mut.RUnlock()
88
- return c.n
89
-}
90
-
91
-type peerSet struct {
92
- ps map[string]bool
93
- lk sync.RWMutex
94
-}
95
-
96
-func newPeerSet() *peerSet {
97
- ps := new(peerSet)
98
- ps.ps = make(map[string]bool)
99
- return ps
100
-}
101
-
102
-func (ps *peerSet) Add(p *peer.Peer) {
103
- ps.lk.Lock()
104
- ps.ps[string(p.ID)] = true
105
- ps.lk.Unlock()
106
-}
107
-
108
-func (ps *peerSet) Contains(p *peer.Peer) bool {
109
- ps.lk.RLock()
110
- _, ok := ps.ps[string(p.ID)]
111
- ps.lk.RUnlock()
112
- return ok
113
-}
114
-
115
-func (ps *peerSet) Size() int {
116
- ps.lk.RLock()
117
- defer ps.lk.RUnlock()
118
- return len(ps.ps)
119
-}
120
-
50
// GetValue searches for the value corresponding to given Key.
51
// If the search does not succeed, a multiaddr string of a closer peer is
52
// returned along with util.ErrSearchIncomplete
@@ -159,9 +88,13 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
88
89
count := 0
90
go func() {
91
+ defer close(procPeer)
92
for {
93
select {
164
- case p := <-npeerChan:
94
+ case p, ok := <-npeerChan:
95
+ if !ok {
96
+ return
97
+ }
98
count++
99
if count >= KValue {
100
errChan <- u.ErrNotFound
@@ -171,8 +104,11 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
104
105
procPeer <- p
106
default:
174
- if c.Size() == 0 {
175
- errChan <- u.ErrNotFound
107
+ if c.Size() <= 0 {
108
+ select {
109
+ case errChan <- u.ErrNotFound:
110
+ default:
111
+ }
112
return
113
}
114
}
@@ -180,20 +116,22 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
116
}()
117
118
process := func() {
119
+ defer c.Decrement()
120
for p := range procPeer {
121
if p == nil {
185
- c.Decrement()
122
return
123
}
124
val, peers, err := dht.getValueOrPeers(p, key, timeout/4, routeLevel)
125
if err != nil {
126
u.DErr("%v\n", err.Error())
191
- c.Decrement()
127
continue
128
}
129
if val != nil {
195
- valChan <- val
196
- c.Decrement()
130
+ select {
131
+ case valChan <- val:
132
+ default:
133
+ u.DOut("Wasnt the first to return the value!")
134
+ }
135
return
136
}
137
@@ -347,7 +285,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
285
// Thoughts: maybe this should accept an ID and do a peer lookup?
286
u.DOut("Enter Ping.")
287
350
- pmes := Message{ID: GenerateMessageID(), Type: PBDHTMessage_PING}
288
+ pmes := Message{ID: swarm.GenerateMessageID(), Type: PBDHTMessage_PING}
289
mes := swarm.NewMessage(p, pmes.ToProtobuf())
290
291
before := time.Now()
@@ -363,7 +301,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
301
return nil
302
case <-tout:
303
// Timed out, think about removing peer from network
366
- u.DOut("Ping peer timed out.")
304
+ u.DOut("[%s] Ping peer [%s] timed out.", dht.self.ID.Pretty(), p.ID.Pretty())
305
dht.listener.Unlisten(pmes.ID)
306
return u.ErrTimeout
307
}
@@ -377,7 +315,7 @@ func (dht *IpfsDHT) getDiagnostic(timeout time.Duration) ([]*diagInfo, error) {
315
// TODO: Add timeout to this struct so nodes know when to return
316
pmes := Message{
317
Type: PBDHTMessage_DIAGNOSTIC,
380
- ID: GenerateMessageID(),
318
+ ID: swarm.GenerateMessageID(),
319
}
320
321
listenChan := dht.listener.Listen(pmes.ID, len(targets), time.Minute*2)
routing/dht/util.go
new
+71
@@ -0,0 +1,71 @@
1
+package dht
2
+
3
+import (
4
+ "sync"
5
+
6
+ peer "github.com/jbenet/go-ipfs/peer"
7
+)
8
+
9
+// Pool size is the number of nodes used for group find/set RPC calls
10
+var PoolSize = 6
11
+
12
+// We put the 'K' in kademlia!
13
+var KValue = 10
14
+
15
+// Its in the paper, i swear
16
+var AlphaValue = 3
17
+
18
+// A counter for incrementing a variable across multiple threads
19
+type counter struct {
20
+ n int
21
+ mut sync.Mutex
22
+}
23
+
24
+func (c *counter) Increment() {
25
+ c.mut.Lock()
26
+ c.n++
27
+ c.mut.Unlock()
28
+}
29
+
30
+func (c *counter) Decrement() {
31
+ c.mut.Lock()
32
+ c.n--
33
+ c.mut.Unlock()
34
+}
35
+
36
+func (c *counter) Size() (s int) {
37
+ c.mut.Lock()
38
+ s = c.n
39
+ c.mut.Unlock()
40
+ return
41
+}
42
+
43
+type peerSet struct {
44
+ ps map[string]bool
45
+ lk sync.RWMutex
46
+}
47
+
48
+func newPeerSet() *peerSet {
49
+ ps := new(peerSet)
50
+ ps.ps = make(map[string]bool)
51
+ return ps
52
+}
53
+
54
+func (ps *peerSet) Add(p *peer.Peer) {
55
+ ps.lk.Lock()
56
+ ps.ps[string(p.ID)] = true
57
+ ps.lk.Unlock()
58
+}
59
+
60
+func (ps *peerSet) Contains(p *peer.Peer) bool {
61
+ ps.lk.RLock()
62
+ _, ok := ps.ps[string(p.ID)]
63
+ ps.lk.RUnlock()
64
+ return ok
65
+}
66
+
67
+func (ps *peerSet) Size() int {
68
+ ps.lk.RLock()
69
+ defer ps.lk.RUnlock()
70
+ return len(ps.ps)
71
+}
swarm/mes_listener.go
renamed
+18
-13
@@ -1,14 +1,14 @@
1
-package dht
1
+package swarm
2
3
import (
4
+ "math/rand"
5
"sync"
6
"time"
7
7
- swarm "github.com/jbenet/go-ipfs/swarm"
8
u "github.com/jbenet/go-ipfs/util"
9
)
10
11
-type mesListener struct {
11
+type MesListener struct {
12
listeners map[uint64]*listenInfo
13
haltchan chan struct{}
14
unlist chan uint64
@@ -16,10 +16,15 @@ type mesListener struct {
16
send chan *respMes
17
}
18
19
+// GenerateMessageID creates and returns a new message ID
20
+func GenerateMessageID() uint64 {
21
+ return (uint64(rand.Uint32()) << 32) | uint64(rand.Uint32())
22
+}
23
+
24
// The listen info struct holds information about a message that is being waited for
25
type listenInfo struct {
26
// Responses matching the listen ID will be sent through resp
22
- resp chan *swarm.Message
27
+ resp chan *Message
28
29
// count is the number of responses to listen for
30
count int
@@ -36,8 +41,8 @@ type listenInfo struct {
41
id uint64
42
}
43
39
-func newMesListener() *mesListener {
40
- ml := new(mesListener)
44
+func NewMesListener() *MesListener {
45
+ ml := new(MesListener)
46
ml.haltchan = make(chan struct{})
47
ml.listeners = make(map[uint64]*listenInfo)
48
ml.nlist = make(chan *listenInfo, 16)
@@ -47,37 +52,37 @@ func newMesListener() *mesListener {
52
return ml
53
}
54
50
-func (ml *mesListener) Listen(id uint64, count int, timeout time.Duration) <-chan *swarm.Message {
55
+func (ml *MesListener) Listen(id uint64, count int, timeout time.Duration) <-chan *Message {
56
li := new(listenInfo)
57
li.count = count
58
li.eol = time.Now().Add(timeout)
54
- li.resp = make(chan *swarm.Message, count)
59
+ li.resp = make(chan *Message, count)
60
li.id = id
61
ml.nlist <- li
62
return li.resp
63
}
64
60
-func (ml *mesListener) Unlisten(id uint64) {
65
+func (ml *MesListener) Unlisten(id uint64) {
66
ml.unlist <- id
67
}
68
69
type respMes struct {
70
id uint64
66
- mes *swarm.Message
71
+ mes *Message
72
}
73
69
-func (ml *mesListener) Respond(id uint64, mes *swarm.Message) {
74
+func (ml *MesListener) Respond(id uint64, mes *Message) {
75
ml.send <- &respMes{
76
id: id,
77
mes: mes,
78
}
79
}
80
76
-func (ml *mesListener) Halt() {
81
+func (ml *MesListener) Halt() {
82
ml.haltchan <- struct{}{}
83
}
84
80
-func (ml *mesListener) run() {
85
+func (ml *MesListener) run() {
86
for {
87
select {
88
case <-ml.haltchan:
swarm/mes_listener_test.go
renamed
+6
-7
@@ -1,23 +1,22 @@
1
-package dht
1
+package swarm
2
3
import (
4
"testing"
5
"time"
6
7
- "github.com/jbenet/go-ipfs/peer"
8
- "github.com/jbenet/go-ipfs/swarm"
7
+ peer "github.com/jbenet/go-ipfs/peer"
8
)
9
10
// Ensure that the Message Listeners basic functionality works
11
func TestMesListenerBasic(t *testing.T) {
13
- ml := newMesListener()
12
+ ml := NewMesListener()
13
a := GenerateMessageID()
14
resp := ml.Listen(a, 1, time.Minute)
15
17
- pmes := new(swarm.PBWrapper)
16
+ pmes := new(PBWrapper)
17
pmes.Message = []byte("Hello")
19
- pmes.Type = new(swarm.PBWrapper_MessageType)
20
- mes := swarm.NewMessage(new(peer.Peer), pmes)
18
+ pmes.Type = new(PBWrapper_MessageType)
19
+ mes := NewMessage(new(peer.Peer), pmes)
20
21
go ml.Respond(a, mes)
22
swarm/mes_wrapper.pb.go
+3
@@ -25,15 +25,18 @@ type PBWrapper_MessageType int32
25
const (
26
PBWrapper_TEST PBWrapper_MessageType = 0
27
PBWrapper_DHT_MESSAGE PBWrapper_MessageType = 1
28
+ PBWrapper_BITSWAP PBWrapper_MessageType = 2
29
)
30
31
var PBWrapper_MessageType_name = map[int32]string{
32
0: "TEST",
33
1: "DHT_MESSAGE",
34
+ 2: "BITSWAP",
35
}
36
var PBWrapper_MessageType_value = map[string]int32{
37
"TEST": 0,
38
"DHT_MESSAGE": 1,
39
+ "BITSWAP": 2,
40
}
41
42
func (x PBWrapper_MessageType) Enum() *PBWrapper_MessageType {
swarm/mes_wrapper.proto
+1
@@ -4,6 +4,7 @@ message PBWrapper {
4
enum MessageType {
5
TEST = 0;
6
DHT_MESSAGE = 1;
7
+ BITSWAP = 2;
8
}
9
10
required MessageType Type = 1;
swarm/swarm.go
+10
@@ -90,6 +90,7 @@ type Swarm struct {
90
91
local *peer.Peer
92
listeners []net.Listener
93
+ haltroute chan struct{}
94
}
95
96
// NewSwarm constructs a Swarm, with a Chan.
@@ -101,6 +102,7 @@ func NewSwarm(local *peer.Peer) *Swarm {
102
filterChans: make(map[PBWrapper_MessageType]*Chan),
103
toFilter: make(chan *Message, 32),
104
newFilters: make(chan *newFilterInfo),
105
+ haltroute: make(chan struct{}),
106
}
107
go s.routeMessages()
108
go s.fanOut()
@@ -203,6 +205,12 @@ func (s *Swarm) Close() {
205
for _, list := range s.listeners {
206
list.Close()
207
}
208
+
209
+ s.haltroute <- struct{}{}
210
+
211
+ for _, filter := range s.filterChans {
212
+ filter.Close <- true
213
+ }
214
}
215
216
// Dial connects to a peer.
@@ -347,6 +355,8 @@ func (s *Swarm) routeMessages() {
355
go s.muxChan(nch, gchan.Type)
356
}
357
gchan.resp <- nch
358
+ case <-s.haltroute:
359
+ return
360
}
361
}
362
}