file copy to also use modern streaming

Author: sawka

cmd/server/main-server.go

@@ -384,7 +384,6 @@ func shutdownActivityUpdate() {
 
 func createMainWshClient() {
 	rpc := wshserver.GetMainRpcClient()
-	wshfs.RpcClient = rpc
 	wshutil.DefaultRouter.RegisterTrustedLeaf(rpc, wshutil.DefaultRoute)
 	wps.Broker.SetClient(wshutil.DefaultRouter)
 	localInitialEnv := envutil.PruneInitialEnv(envutil.SliceToMap(os.Environ()))
@@ -393,6 +392,8 @@ func createMainWshClient() {
 	localConnWsh := wshutil.MakeWshRpc(wshrpc.RpcContext{Conn: wshrpc.LocalConnName}, remoteImpl, "conn:local")
 	go wshremote.RunSysInfoLoop(localConnWsh, wshrpc.LocalConnName)
 	wshutil.DefaultRouter.RegisterTrustedLeaf(localConnWsh, wshutil.MakeConnectionRouteId(wshrpc.LocalConnName))
+	wshfs.RpcClient = localConnWsh
+	wshfs.RpcClientRouteId = wshutil.MakeConnectionRouteId(wshrpc.LocalConnName)
 }
 
 func grabAndRemoveEnvVars() error {

cmd/wsh/cmd/wshcmd-connserver.go

@@ -183,7 +183,7 @@ func runListener(listener net.Listener, router *wshutil.WshRouter) {
 	}
 }
 
-func setupConnServerRpcClientWithRouter(router *wshutil.WshRouter, sockName string) (*wshutil.WshRpc, error) {
+func setupConnServerRpcClientWithRouter(router *wshutil.WshRouter, sockName string) (*wshutil.WshRpc, string, error) {
 	routeId := wshutil.MakeConnectionRouteId(connServerConnName)
 	rpcCtx := wshrpc.RpcContext{
 		RouteId: routeId,
@@ -196,7 +196,7 @@ func setupConnServerRpcClientWithRouter(router *wshutil.WshRouter, sockName stri
 
 	connServerClient := wshutil.MakeWshRpc(rpcCtx, wshremote.MakeRemoteRpcServerImpl(os.Stdout, router, bareClient, false, connServerInitialEnv, sockName), routeId)
 	router.RegisterTrustedLeaf(connServerClient, routeId)
-	return connServerClient, nil
+	return connServerClient, routeId, nil
 }
 
 func serverRunRouter() error {
@@ -236,11 +236,12 @@ func serverRunRouter() error {
 	sockName := getRemoteDomainSocketName()
 
 	// setup the connserver rpc client first
-	client, err := setupConnServerRpcClientWithRouter(router, sockName)
+	client, bareRouteId, err := setupConnServerRpcClientWithRouter(router, sockName)
 	if err != nil {
 		return fmt.Errorf("error setting up connserver rpc client: %v", err)
 	}
 	wshfs.RpcClient = client
+	wshfs.RpcClientRouteId = bareRouteId
 
 	log.Printf("trying to get JWT public key")
 
@@ -360,11 +361,12 @@ func serverRunRouterDomainSocket(jwtToken string) error {
 	log.Printf("got JWT public key")
 
 	// now setup the connserver rpc client
-	client, err := setupConnServerRpcClientWithRouter(router, sockName)
+	client, bareRouteId, err := setupConnServerRpcClientWithRouter(router, sockName)
 	if err != nil {
 		return fmt.Errorf("error setting up connserver rpc client: %v", err)
 	}
 	wshfs.RpcClient = client
+	wshfs.RpcClientRouteId = bareRouteId
 
 	// set up the local domain socket listener for local wsh commands
 	unixListener, err := MakeRemoteUnixListener()
@@ -402,6 +404,7 @@ func serverRunNormal(jwtToken string) error {
 		return err
 	}
 	wshfs.RpcClient = RpcClient
+	wshfs.RpcClientRouteId = RpcClientRouteId
 	WriteStdout("running wsh connserver (%s)\n", RpcContext.Conn)
 	go func() {
 		defer func() {

pkg/remote/fileshare/wshfs/wshfs.go

@@ -30,6 +30,7 @@ const (
 
 // This needs to be set by whoever initializes the client, either main-server or wshcmd-connserver
 var RpcClient *wshutil.WshRpc
+var RpcClientRouteId string
 
 func parseConnection(ctx context.Context, path string) (*connparse.Connection, error) {
 	conn, err := connparse.ParseURIAndReplaceCurrentHost(ctx, path)

pkg/wshrpc/wshremote/wshremote_file.go

@@ -18,7 +18,6 @@ import (
 
 	"github.com/wavetermdev/waveterm/pkg/panichandler"
 	"github.com/wavetermdev/waveterm/pkg/remote/connparse"
-	"github.com/wavetermdev/waveterm/pkg/remote/fileshare/fsutil"
 	"github.com/wavetermdev/waveterm/pkg/remote/fileshare/wshfs"
 	"github.com/wavetermdev/waveterm/pkg/util/fileutil"
 	"github.com/wavetermdev/waveterm/pkg/util/utilfn"
@@ -32,7 +31,6 @@ const RemoteFileTransferSizeLimit = 32 * 1024 * 1024
 
 var DisableRecursiveFileOpts = true
 
-
 func (impl *ServerImpl) remoteStreamFileDir(ctx context.Context, path string, byteRange fileutil.ByteRangeType, dataCallback func(fileInfo []*wshrpc.FileInfo, data []byte, byteRange fileutil.ByteRangeType)) error {
 	innerFilesEntries, err := os.ReadDir(path)
 	if err != nil {
@@ -315,8 +313,25 @@ func (impl *ServerImpl) RemoteFileCopyCommand(ctx context.Context, data wshrpc.C
 	}
 	defer destFile.Close()
 
-	streamChan := wshclient.RemoteStreamFileCommand(wshfs.RpcClient, wshrpc.CommandRemoteStreamFileData{Path: srcConn.Path}, &wshrpc.RpcOpts{Timeout: opts.Timeout, Route: wshutil.MakeConnectionRouteId(srcConn.Host)})
-	if err = fsutil.ReadFileStreamToWriter(readCtx, streamChan, destFile); err != nil {
+	if wshfs.RpcClientRouteId == "" {
+		return false, fmt.Errorf("stream broker route id not available for file copy")
+	}
+	writerRouteId := wshutil.MakeConnectionRouteId(srcConn.Host)
+	reader, streamMeta := wshfs.RpcClient.StreamBroker.CreateStreamReader(wshfs.RpcClientRouteId, writerRouteId, 256*1024)
+	log.Printf("RemoteFileCopyCommand: readroute=%s writeroute=%s", streamMeta.ReaderRouteId, streamMeta.WriterRouteId)
+	defer reader.Close()
+	go func() {
+		<-readCtx.Done()
+		reader.Close()
+	}()
+	streamData := wshrpc.CommandRemoteFileStreamData{
+		Path:       srcConn.Path,
+		StreamMeta: *streamMeta,
+	}
+	if _, err = wshclient.RemoteFileStreamCommand(wshfs.RpcClient, streamData, &wshrpc.RpcOpts{Route: writerRouteId}); err != nil {
+		return false, fmt.Errorf("error starting file stream for %q: %w", data.SrcUri, err)
+	}
+	if _, err = io.Copy(destFile, reader); err != nil {
 		return false, fmt.Errorf("error copying file %q to %q: %w", data.SrcUri, data.DestUri, err)
 	}