identify: refactor sending of Identify pushes

p2p/host/basic/basic_host.go

@@ -181,7 +181,7 @@ func NewHost(n network.Network, opts *HostOpts) (*BasicHost, error) {
 
  h.updateLocalIpAddr()
 
- if h.emitters.evtLocalProtocolsUpdated, err = h.eventbus.Emitter(&event.EvtLocalProtocolsUpdated{}); err != nil {
+ if h.emitters.evtLocalProtocolsUpdated, err = h.eventbus.Emitter(&event.EvtLocalProtocolsUpdated{}, eventbus.Stateful); err != nil {
  return nil, err
  }
  if h.emitters.evtLocalAddrsUpdated, err = h.eventbus.Emitter(&event.EvtLocalAddressesUpdated{}, eventbus.Stateful); err != nil {
@@ -367,6 +367,7 @@ func (h *BasicHost) updateLocalIpAddr() {
 func (h *BasicHost) Start() {
  h.psManager.Start()
  h.refCount.Add(1)
+ h.ids.Start()
  go h.background()
 }
 

p2p/host/basic/basic_host_test.go

@@ -176,7 +176,7 @@ func TestHostAddrsFactory(t *testing.T) {
 
  addrs := h.Addrs()
  if len(addrs) != 1 {
- t.Fatalf("expected 1 addr, got %d", len(addrs))
+ t.Fatalf("expected 1 addr, got %+v", addrs)
  }
  if !addrs[0].Equal(maddr) {
  t.Fatalf("expected %s, got %s", maddr.String(), addrs[0].String())
@@ -245,8 +245,10 @@ func getHostPair(t *testing.T) (host.Host, host.Host) {
 
  h1, err := NewHost(swarmt.GenSwarm(t), nil)
  require.NoError(t, err)
+ h1.Start()
  h2, err := NewHost(swarmt.GenSwarm(t), nil)
  require.NoError(t, err)
+ h2.Start()
 
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  defer cancel()
@@ -342,14 +344,12 @@ func TestHostProtoMismatch(t *testing.T) {
 }
 
 func TestHostProtoPreknowledge(t *testing.T) {
- ctx, cancel := context.WithCancel(context.Background())
- defer cancel()
-
  h1, err := NewHost(swarmt.GenSwarm(t), nil)
  require.NoError(t, err)
+ defer h1.Close()
+
  h2, err := NewHost(swarmt.GenSwarm(t), nil)
  require.NoError(t, err)
- defer h1.Close()
  defer h2.Close()
 
  conn := make(chan protocol.ID)
@@ -362,8 +362,11 @@ func TestHostProtoPreknowledge(t *testing.T) {
  // Prevent pushing identify information so this test actually _uses_ the super protocol.
  h1.RemoveStreamHandler(identify.IDPush)
 
+ h1.Start()
+ h2.Start()
+
  h2pi := h2.Peerstore().PeerInfo(h2.ID())
- require.NoError(t, h1.Connect(ctx, h2pi))
+ require.NoError(t, h1.Connect(context.Background(), h2pi))
 
  // wait for identify handshake to finish completely
  select {
@@ -380,12 +383,12 @@ func TestHostProtoPreknowledge(t *testing.T) {
 
  h2.SetStreamHandler("/foo", handler)
 
- s, err := h1.NewStream(ctx, h2.ID(), "/foo", "/bar", "/super")
+ s, err := h1.NewStream(context.Background(), h2.ID(), "/foo", "/bar", "/super")
  require.NoError(t, err)
 
  select {
  case p := <-conn:
- t.Fatal("shouldnt have gotten connection yet, we should have a lazy stream: ", p)
+ t.Fatal("shouldn't have gotten connection yet, we should have a lazy stream: ", p)
  case <-time.After(time.Millisecond * 50):
  }
 
@@ -532,7 +535,6 @@ func TestAddrChangeImmediatelyIfAddressNonEmpty(t *testing.T) {
  return taddrs
  }})
  require.NoError(t, err)
- h.Start()
  defer h.Close()
 
  sub, err := h.EventBus().Subscribe(&event.EvtLocalAddressesUpdated{})
@@ -541,6 +543,7 @@ func TestAddrChangeImmediatelyIfAddressNonEmpty(t *testing.T) {
  t.Error(err)
  }
  defer sub.Close()
+ h.Start()
 
  expected := event.EvtLocalAddressesUpdated{
  Diffs: true,

p2p/net/mock/mock_test.go

@@ -528,7 +528,7 @@ func TestLimitedStreams(t *testing.T) {
  }
 
  wg.Wait()
- if !within(time.Since(before), time.Second*2, time.Second) {
+ if !within(time.Since(before), time.Second*5/2, time.Second) {
  t.Fatal("Expected 2ish seconds but got ", time.Since(before))
  }
 }

p2p/protocol/holepunch/holepunch_test.go

@@ -7,6 +7,10 @@ import (
  "testing"
  "time"
 
+ "github.com/libp2p/go-libp2p/p2p/transport/tcp"
+
+ "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto"
+
  "github.com/libp2p/go-libp2p"
  "github.com/libp2p/go-libp2p-testing/race"
  "github.com/libp2p/go-libp2p/core/host"
@@ -68,6 +72,8 @@ var _ identify.IDService = &mockIDService{}
 func newMockIDService(t *testing.T, h host.Host) identify.IDService {
  ids, err := identify.NewIDService(h)
  require.NoError(t, err)
+ ids.Start()
+ t.Cleanup(func() { ids.Close() })
  return &mockIDService{IDService: ids}
 }
 
@@ -448,10 +454,23 @@ func makeRelayedHosts(t *testing.T, h1opt, h2opt []holepunch.Option, addHolePunc
  libp2p.ResourceManager(&network.NullResourceManager{}),
  )
  require.NoError(t, err)
-
  _, err = relayv1.NewRelay(relay)
  require.NoError(t, err)
 
+ // make sure the relay service is started and advertised by Identify
+ h, err := libp2p.New(
+ libp2p.NoListenAddrs,
+ libp2p.Transport(tcp.NewTCPTransport),
+ libp2p.DisableRelay(),
+ )
+ require.NoError(t, err)
+ defer h.Close()
+ require.NoError(t, h.Connect(context.Background(), peer.AddrInfo{ID: relay.ID(), Addrs: relay.Addrs()}))
+ require.Eventually(t, func() bool {
+ supported, err := h.Peerstore().SupportsProtocols(relay.ID(), proto.ProtoIDv2Hop, relayv1.ProtoID)
+ return err == nil && len(supported) > 0
+ }, 3*time.Second, 100*time.Millisecond)
+
  h2 = mkHostWithStaticAutoRelay(t, relay)
  if addHolePuncher {
  hps = addHolePunchService(t, h2, h2opt...)