Resolve addresses when creating a new stream (#1342)

* Resolve addresses when creating a new stream

BasicHost.NewStream will try to establish a connection if one doesn't already exist.  This will fail if the hosts addresses have not yet been resolved.  This PR resolves the hosts addresses before creating the stream and possible new connection.

Fixes #1302

* Changes from review comments
This commit is contained in:
Andrew Gillis 2022-03-31 02:45:20 -07:00 committed by GitHub
parent 72fe013d15
commit e8d3df92f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 57 additions and 0 deletions

View File

@ -605,6 +605,17 @@ func (h *BasicHost) RemoveStreamHandler(pid protocol.ID) {
// to create one. If ProtocolID is "", writes no header. // to create one. If ProtocolID is "", writes no header.
// (Threadsafe) // (Threadsafe)
func (h *BasicHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error) { func (h *BasicHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error) {
// Ensure we have a connection, with peer addresses resolved by the routing system (#207)
// It is not sufficient to let the underlying host connect, it will most likely not have
// any addresses for the peer without any prior connections.
// If the caller wants to prevent the host from dialing, it should use the NoDial option.
if nodial, _ := network.GetNoDial(ctx); !nodial {
err := h.Connect(ctx, peer.AddrInfo{ID: p})
if err != nil {
return nil, err
}
}
s, err := h.Network().NewStream(ctx, p) s, err := h.Network().NewStream(ctx, p)
if err != nil { if err != nil {
return nil, err return nil, err

View File

@ -6,6 +6,7 @@ import (
"io" "io"
"io/ioutil" "io/ioutil"
"reflect" "reflect"
"strings"
"sync" "sync"
"testing" "testing"
"time" "time"
@ -439,6 +440,51 @@ func TestNewDialOld(t *testing.T) {
require.Equal(t, s.Protocol(), protocol.ID("/testing"), "should have gotten /testing") require.Equal(t, s.Protocol(), protocol.ID("/testing"), "should have gotten /testing")
} }
func TestNewStreamResolve(t *testing.T) {
h1, err := NewHost(swarmt.GenSwarm(t), nil)
require.NoError(t, err)
h2, err := NewHost(swarmt.GenSwarm(t), nil)
require.NoError(t, err)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
// Get the tcp port that h2 is listening on.
h2pi := h2.Peerstore().PeerInfo(h2.ID())
var dialAddr string
const tcpPrefix = "/ip4/127.0.0.1/tcp/"
for _, addr := range h2pi.Addrs {
addrStr := addr.String()
if strings.HasPrefix(addrStr, tcpPrefix) {
port := addrStr[len(tcpPrefix):]
dialAddr = "/dns4/localhost/tcp/" + port
break
}
}
assert.NotEqual(t, dialAddr, "")
// Add the DNS multiaddr to h1's peerstore.
maddr, err := ma.NewMultiaddr(dialAddr)
require.NoError(t, err)
h1.Peerstore().AddAddr(h2.ID(), maddr, time.Second)
connectedOn := make(chan protocol.ID)
h2.SetStreamHandler("/testing", func(s network.Stream) {
connectedOn <- s.Protocol()
s.Close()
})
// NewStream will make a new connection using the DNS address in h1's
// peerstore.
s, err := h1.NewStream(ctx, h2.ID(), "/testing/1.0.0", "/testing")
require.NoError(t, err)
// force the lazy negotiation to complete
_, err = s.Write(nil)
require.NoError(t, err)
assertWait(t, connectedOn, "/testing")
}
func TestProtoDowngrade(t *testing.T) { func TestProtoDowngrade(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()