fix(stream): properly handle remote stream scheme IPv4/6

internal/route/route.go

@@ -269,9 +269,9 @@ func (r *Route) validate() gperr.Error {
 
 	switch r.Scheme {
 	case route.SchemeFileServer:
-		r.ProxyURL = gperr.Collect(&errs, nettypes.ParseURL, "file://"+r.Root)
 		r.Host = ""
 		r.Port.Proxy = 0
+		r.ProxyURL = gperr.Collect(&errs, nettypes.ParseURL, "file://"+r.Root)
 	case route.SchemeHTTP, route.SchemeHTTPS, route.SchemeH2C:
 		if r.Port.Listening != 0 {
 			errs.Addf("unexpected listening port for %s scheme", r.Scheme)
@@ -286,22 +286,30 @@ func (r *Route) validate() gperr.Error {
 			if bindIP == nil {
 				return gperr.Errorf("invalid bind address %s", r.Bind)
 			}
-			var scheme string
-			if bindIP.To4() == nil { // IPv6
-				if r.Scheme == route.SchemeTCP {
-					scheme = "tcp6"
-				} else {
-					scheme = "udp6"
-				}
-			} else {
-				if r.Scheme == route.SchemeTCP {
-					scheme = "tcp4"
-				} else {
-					scheme = "udp4"
-				}
+			remoteIP := net.ParseIP(r.Host)
+			if remoteIP == nil {
+				return gperr.Errorf("invalid remote address %s", r.Host)
 			}
-			r.LisURL = gperr.Collect(&errs, nettypes.ParseURL, fmt.Sprintf("%s://%s:%d", scheme, r.Bind, r.Port.Listening))
+			toNetwork := func(ip net.IP, scheme route.Scheme) string {
+				if ip.To4() == nil {
+					if scheme == route.SchemeTCP {
+						return "tcp6"
+					}
+					return "udp6"
+				}
+				if scheme == route.SchemeTCP {
+					return "tcp4"
+				}
+				return "udp4"
+			}
+			lScheme := toNetwork(bindIP, r.Scheme)
+			rScheme := toNetwork(remoteIP, r.Scheme)
+
+			r.LisURL = gperr.Collect(&errs, nettypes.ParseURL, fmt.Sprintf("%s://%s:%d", lScheme, r.Bind, r.Port.Listening))
+			r.ProxyURL = gperr.Collect(&errs, nettypes.ParseURL, fmt.Sprintf("%s://%s:%d", rScheme, r.Host, r.Port.Proxy))
 		}
+
+		// should exclude, we don't care the scheme here.
 		r.ProxyURL = gperr.Collect(&errs, nettypes.ParseURL, fmt.Sprintf("%s://%s:%d", r.Scheme, r.Host, r.Port.Proxy))
 	}
 

internal/route/stream.go

@@ -70,7 +70,7 @@ func (r *StreamRoute) Start(parent task.Parent) gperr.Error {
 
 	r.ListenAndServe(r.task.Context(), nil, nil)
 	r.l = log.With().
-		Str("type", r.LisURL.Scheme).
+		Str("type", r.LisURL.Scheme+"->"+r.ProxyURL.Scheme).
 		Str("name", r.Name()).
 		Stringer("rurl", r.ProxyURL).
 		Stringer("laddr", r.LocalAddr()).Logger()
@@ -102,9 +102,8 @@ func (r *StreamRoute) LocalAddr() net.Addr {
 
 func (r *StreamRoute) initStream() (nettypes.Stream, error) {
 	lurl, rurl := r.LisURL, r.ProxyURL
-	// lurl scheme is either tcp4/tcp6 -> tcp, udp4/udp6 -> udp
-	// rurl scheme does not have the trailing 4/6
-	if strings.TrimRight(lurl.Scheme, "46") != rurl.Scheme {
+	// tcp4/tcp6 -> tcp, udp4/udp6 -> udp
+	if strings.TrimRight(lurl.Scheme, "46") != strings.TrimRight(rurl.Scheme, "46") {
 		return nil, fmt.Errorf("incoherent scheme is not yet supported: %s != %s", lurl.Scheme, rurl.Scheme)
 	}
 

internal/route/stream/tcp_tcp.go

@@ -136,7 +136,7 @@ func (s *TCPTCPStream) handle(ctx context.Context, conn net.Conn) {
 	if s.agent != nil {
 		dstConn, err = s.agent.NewTCPClient(s.dst.String())
 	} else {
-		dstConn, err = net.DialTCP("tcp", nil, s.dst)
+		dstConn, err = net.DialTCP(s.dst.Network(), nil, s.dst)
 	}
 	if err != nil {
 		if !s.closed.Load() {

internal/route/stream/udp_udp.go

@@ -198,7 +198,7 @@ func (s *UDPUDPStream) createConnection(ctx context.Context, srcAddr *net.UDPAdd
 	if s.agent != nil {
 		dstConn, err = s.agent.NewUDPClient(s.dst.String())
 	} else {
-		dstConn, err = net.DialUDP("udp", nil, s.dst)
+		dstConn, err = net.DialUDP(s.dst.Network(), nil, s.dst)
 	}
 	if err != nil {
 		logErr(s, err, "failed to dial dst")