allow removal of stream handlers
Jeromy committed
Feb 15, 2015 at 01:50 UTC
2309266f73a993a709705d2a1e41cea04885d30c
4 files changed
+27
p2p/host/basic/basic_host.go
+4
@@ -118,6 +118,10 @@ func (h *BasicHost) SetStreamHandler(pid protocol.ID, handler inet.StreamHandler
118
h.Mux().SetHandler(pid, handler)
119
}
120
121
+func (h *BasicHost) RemoveStreamHandler(pid protocol.ID) {
122
+ h.Mux().RemoveHandler(pid)
123
+}
124
+
125
// NewStream opens a new stream to given peer p, and writes a p2p/protocol
126
// header with given protocol.ID. If there is no connection to p, attempts
127
// to create one. If ProtocolID is "", writes no header.
p2p/host/host.go
+4
@@ -46,6 +46,10 @@ type Host interface {
46
// (Threadsafe)
47
SetStreamHandler(pid protocol.ID, handler inet.StreamHandler)
48
49
+ // RemoveStreamHandler removes a handler on the mux that was set by
50
+ // SetStreamHandler
51
+ RemoveStreamHandler(pid protocol.ID)
52
+
53
// NewStream opens a new stream to given peer p, and writes a p2p/protocol
54
// header with given protocol.ID. If there is no connection to p, attempts
55
// to create one. If ProtocolID is "", writes no header.
p2p/host/routed/routed.go
+10
@@ -84,21 +84,31 @@ func logRoutingErrDifferentPeers(ctx context.Context, wanted, got peer.ID, err e
84
func (rh *RoutedHost) ID() peer.ID {
85
return rh.host.ID()
86
}
87
+
88
func (rh *RoutedHost) Peerstore() peer.Peerstore {
89
return rh.host.Peerstore()
90
}
91
+
92
func (rh *RoutedHost) Addrs() []ma.Multiaddr {
93
return rh.host.Addrs()
94
}
95
+
96
func (rh *RoutedHost) Network() inet.Network {
97
return rh.host.Network()
98
}
99
+
100
func (rh *RoutedHost) Mux() *protocol.Mux {
101
return rh.host.Mux()
102
}
103
+
104
func (rh *RoutedHost) SetStreamHandler(pid protocol.ID, handler inet.StreamHandler) {
105
rh.host.SetStreamHandler(pid, handler)
106
}
107
+
108
+func (rh *RoutedHost) RemoveStreamHandler(pid protocol.ID) {
109
+ rh.host.RemoveStreamHandler(pid)
110
+}
111
+
112
func (rh *RoutedHost) NewStream(pid protocol.ID, p peer.ID) (inet.Stream, error) {
113
return rh.host.NewStream(pid, p)
114
}
p2p/protocol/mux.go
+9
@@ -90,6 +90,15 @@ func (m *Mux) SetHandler(p ID, h inet.StreamHandler) {
90
m.lock.Unlock()
91
}
92
93
+// RemoveHandler removes the protocol handler on the Network's Muxer.
94
+// This operation is threadsafe.
95
+func (m *Mux) RemoveHandler(p ID) {
96
+ log.Debugf("%s removing handler for protocol: %s (%d)", m, p, len(p))
97
+ m.lock.Lock()
98
+ delete(m.handlers, p)
99
+ m.lock.Unlock()
100
+}
101
+
102
// Handle reads the next name off the Stream, and calls a handler function
103
// This is done in its own goroutine, to avoid blocking the caller.
104
func (m *Mux) Handle(s inet.Stream) {