fix(ftp): connect fix

octohelm · Apr 22, 2024 · b9b0fa3 · b9b0fa3
1 parent 575423f
commit b9b0fa3
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 36 deletions.
diff --git a/pkg/filesystem/ftp/client.go b/pkg/filesystem/ftp/client.go
@@ -88,7 +88,6 @@ func (p *Pool) Conn(ctx context.Context, args ...any) (Conn, error) {
 		}
 		p.p = pool
 	})
-
 	if p.err != nil {
 		return nil, p.err
 	}

diff --git a/pkg/filesystem/ftp/config.go b/pkg/filesystem/ftp/config.go
@@ -4,7 +4,7 @@ import (
 	"context"
 	"net/url"
 	"strconv"
-	"sync/atomic"
+	"sync"
 	"time"
 
 	"github.com/octohelm/unifs/pkg/strfmt"
@@ -13,45 +13,47 @@ import (
 type Config struct {
 	Endpoint strfmt.Endpoint `flag:",upstream"`
 
-	p atomic.Pointer[Pool]
+	p  *Pool
+	mu sync.Mutex
 }
 
 func (c *Config) BasePath() string {
 	return c.Endpoint.Path
 }
 
 func (c *Config) Conn(ctx context.Context, args ...any) (Conn, error) {
-	if p := c.p.Load(); p != nil {
-		return p.Conn(ctx, args...)
-	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
 
-	p := &Pool{
-		Attr: c.Endpoint.Host(),
-	}
+	if c.p == nil {
+		p := &Pool{
+			Attr: c.Endpoint.Host(),
+		}
 
-	if c.Endpoint.Username != "" {
-		p.Auth = url.UserPassword(c.Endpoint.Username, c.Endpoint.Password)
-	}
+		if c.Endpoint.Username != "" {
+			p.Auth = url.UserPassword(c.Endpoint.Username, c.Endpoint.Password)
+		}
 
-	p.ConnectTimeout = 5 * time.Second
+		p.ConnectTimeout = 5 * time.Second
 
-	if t := c.Endpoint.Extra.Get("timeout"); t != "" {
-		d, err := time.ParseDuration(t)
-		if err != nil {
-			return nil, err
+		if t := c.Endpoint.Extra.Get("timeout"); t != "" {
+			d, err := time.ParseDuration(t)
+			if err != nil {
+				return nil, err
+			}
+			p.ConnectTimeout = d
 		}
-		p.ConnectTimeout = d
-	}
 
-	if t := c.Endpoint.Extra.Get("maxConnections"); t != "" {
-		d, err := strconv.ParseInt(t, 10, 64)
-		if err != nil {
-			return nil, err
+		if t := c.Endpoint.Extra.Get("maxConnections"); t != "" {
+			d, err := strconv.ParseInt(t, 10, 64)
+			if err != nil {
+				return nil, err
+			}
+			p.MaxConnections = int32(d)
 		}
-		p.MaxConnections = int32(d)
-	}
 
-	c.p.Store(p)
+		c.p = p
+	}
 
-	return p.Conn(ctx, args...)
+	return c.p.Conn(ctx, args...)
 }
diff --git a/pkg/filesystem/ftp/file.go b/pkg/filesystem/ftp/file.go
@@ -70,19 +70,19 @@ func (f *file) Read(p []byte) (n int, err error) {
 	}
 
 	f.once.Do(func() {
-		c, err := f.client.Conn(f.ctx)
+		conn, err := f.client.Conn(f.ctx)
 		if err != nil {
 			f.err = normalizeError("read", f.entry.Name, err)
 			return
 		}
 
-		resp, err := c.RetrFrom(f.entry.Name, f.offset)
+		resp, err := conn.RetrFrom(f.entry.Name, f.offset)
 		if err != nil {
 			f.err = normalizeError("read", f.entry.Name, err)
 			return
 		}
 
-		f.readCloser = &readCloser{Response: resp, conn: c}
+		f.readCloser = &readCloser{Response: resp, conn: conn}
 	})
 
 	return f.readCloser.Read(p)
@@ -110,7 +110,7 @@ func (f *file) Write(p []byte) (n int, err error) {
 	}
 
 	f.once.Do(func() {
-		c, err := f.client.Conn(f.ctx, "write", f.entry.Name)
+		conn, err := f.client.Conn(f.ctx, "write", f.entry.Name)
 		if err != nil {
 			f.err = normalizeError("write", f.entry.Name, err)
 			return
@@ -123,12 +123,12 @@ func (f *file) Write(p []byte) (n int, err error) {
 
 		go func() {
 			defer func() {
+				_ = conn.Close()
 				_ = r.Close()
-				_ = c.Close()
 				ww.wg.Done()
 			}()
 
-			if err := c.StorFrom(f.entry.Name, r, f.offset); err != nil {
+			if err := conn.StorFrom(f.entry.Name, r, f.offset); err != nil {
 				f.err = normalizeError("write", f.entry.Name, err)
 			}
 		}()
@@ -155,13 +155,13 @@ func (c *writeCloser) Close() error {
 }
 
 func (f *file) Readdir(count int) ([]os.FileInfo, error) {
-	c, err := f.client.Conn(f.ctx)
+	conn, err := f.client.Conn(f.ctx)
 	if err != nil {
 		return nil, normalizeError("write", f.entry.Name, err)
 	}
-	defer c.Close()
+	defer conn.Close()
 
-	entries, err := c.List(f.entry.Name)
+	entries, err := conn.List(f.entry.Name)
 	if err != nil {
 		return nil, normalizeError("readdir", f.entry.Name, err)
 	}