integrate the event bus into host implementations
@@ -5,12 +5,15 @@ import (
"io"
"github.com/libp2p/go-libp2p-core/connmgr"
+ "github.com/libp2p/go-libp2p-core/event"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
"github.com/libp2p/go-libp2p-core/protocol"
+ "github.com/libp2p/go-eventbus"
+
logging "github.com/ipfs/go-log"
ma "github.com/multiformats/go-multiaddr"
@@ -24,6 +27,10 @@ type BlankHost struct {
n network.Network
mux *mstream.MultistreamMuxer
cmgr connmgr.ConnManager
+ eventbus event.Bus
+ emitters struct {
+ evtLocalProtocolsUpdated event.Emitter
+ }
}
func NewBlankHost(n network.Network) *BlankHost {
@@ -31,6 +38,12 @@ func NewBlankHost(n network.Network) *BlankHost {
n: n,
cmgr: &connmgr.NullConnMgr{},
mux: mstream.NewMultistreamMuxer(),
+ eventbus: eventbus.NewBus(),
+ }
+
+ var err error
+ if bh.emitters.evtLocalProtocolsUpdated, err = bh.eventbus.Emitter(&event.EvtLocalProtocolsUpdated{}); err != nil {
+ return nil
}
n.SetStreamHandler(bh.newStreamHandler)
@@ -98,8 +111,11 @@ func (bh *BlankHost) NewStream(ctx context.Context, p peer.ID, protos ...protoco
return s, nil
}
-func (bh *BlankHost) RemoveStreamHandler(p protocol.ID) {
- bh.Mux().RemoveHandler(string(p))
+func (bh *BlankHost) RemoveStreamHandler(pid protocol.ID) {
+ bh.Mux().RemoveHandler(string(pid))
+ bh.emitters.evtLocalProtocolsUpdated.Emit(event.EvtLocalProtocolsUpdated{
+ Removed: []protocol.ID{pid},
+ })
}
func (bh *BlankHost) SetStreamHandler(pid protocol.ID, handler network.StreamHandler) {
@@ -109,6 +125,9 @@ func (bh *BlankHost) SetStreamHandler(pid protocol.ID, handler network.StreamHan
handler(is)
return nil
})
+ bh.emitters.evtLocalProtocolsUpdated.Emit(event.EvtLocalProtocolsUpdated{
+ Added: []protocol.ID{pid},
+ })
}
func (bh *BlankHost) SetStreamHandlerMatch(pid protocol.ID, m func(string) bool, handler network.StreamHandler) {
@@ -118,11 +137,13 @@ func (bh *BlankHost) SetStreamHandlerMatch(pid protocol.ID, m func(string) bool,
handler(is)
return nil
})
+ bh.emitters.evtLocalProtocolsUpdated.Emit(event.EvtLocalProtocolsUpdated{
+ Added: []protocol.ID{pid},
+ })
}
// newStreamHandler is the remote-opened stream handler for network.Network
func (bh *BlankHost) newStreamHandler(s network.Stream) {
-
protoID, handle, err := bh.Mux().Negotiate(s)
if err != nil {
log.Warning("protocol mux failed: %s", err)
@@ -148,3 +169,7 @@ func (bh *BlankHost) Network() network.Network {
func (bh *BlankHost) ConnManager() connmgr.ConnManager {
return bh.cmgr
}
+
+func (bh *BlankHost) EventBus() event.Bus {
+ return bh.eventbus
+}
