Merge branch 'release/0.6.0'

CHANGELOG.md

@@ -5,9 +5,14 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 
 ## [Unreleased]
 
+## [0.6.0] - 2020-12-26
+### Changed
+* (internal) Improve performance when showing the progress bar
+* (internal) Improve performance when using --yamux, reducing unnecessary buffers
+
 ## [0.5.0] - 2020-12-26
 ### Added
-* add "socks" subcommand for SOCKS5 proxy
+* Add "socks" subcommand for SOCKS5 proxy
 
 ## [0.4.2] - 2020-12-11
 ### Changed
@@ -59,7 +64,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 ### Added
 * Initial release
 
-[Unreleased]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.5.0...HEAD
+[Unreleased]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.6.0...HEAD
+[0.6.0]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.5.0...v0.6.0
 [0.5.0]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.4.2...v0.5.0
 [0.4.2]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.4.1...v0.4.2
 [0.4.1]: https://github.com/nwtgck/go-piping-tunnel/compare/v0.4.0...v0.4.1

cmd/client.go

@@ -8,6 +8,7 @@ import (
 	"github.com/nwtgck/go-piping-tunnel/util"
 	"github.com/spf13/cobra"
 	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -71,8 +72,7 @@ var clientCmd = &cobra.Command{
 		ln.Close()
 		var progress *io_progress.IOProgress = nil
 		if showProgress {
-			p := io_progress.NewIOProgress(conn, os.Stderr, makeProgressMessage)
-			progress = &p
+			progress = io_progress.NewIOProgress(conn, ioutil.Discard, os.Stderr, makeProgressMessage)
 		}
 		var reader io.Reader = conn
 		if progress != nil {
@@ -147,7 +147,7 @@ func clientHandleWithYamux(ln net.Listener, httpClient *http.Client, headers []p
 	}
 	var readWriteCloser io.ReadWriteCloser = duplex
 	if showProgress {
-		readWriteCloser = util.NewIOProgressReadWriteCloser(duplex, os.Stderr, makeProgressMessage)
+		readWriteCloser = io_progress.NewIOProgress(duplex, duplex, os.Stderr, makeProgressMessage)
 	}
 	yamuxSession, err := yamux.Client(readWriteCloser, nil)
 	if err != nil {

cmd/server.go

@@ -8,6 +8,7 @@ import (
 	"github.com/nwtgck/go-piping-tunnel/util"
 	"github.com/spf13/cobra"
 	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -67,8 +68,7 @@ var serverCmd = &cobra.Command{
 		defer conn.Close()
 		var progress *io_progress.IOProgress = nil
 		if showProgress {
-			p := io_progress.NewIOProgress(conn, os.Stderr, makeProgressMessage)
-			progress = &p
+			progress = io_progress.NewIOProgress(conn, ioutil.Discard, os.Stderr, makeProgressMessage)
 		}
 		var reader io.Reader = conn
 		if progress != nil {
@@ -144,7 +144,7 @@ func serverHandleWithYamux(httpClient *http.Client, headers []piping_tunnel_util
 	}
 	var readWriteCloser io.ReadWriteCloser = duplex
 	if showProgress {
-		readWriteCloser = util.NewIOProgressReadWriteCloser(duplex, os.Stderr, makeProgressMessage)
+		readWriteCloser = io_progress.NewIOProgress(duplex, duplex, os.Stderr, makeProgressMessage)
 	}
 	yamuxSession, err := yamux.Server(readWriteCloser, nil)
 	if err != nil {

cmd/socks.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/armon/go-socks5"
 	"github.com/hashicorp/yamux"
+	"github.com/nwtgck/go-piping-tunnel/io_progress"
 	piping_tunnel_util "github.com/nwtgck/go-piping-tunnel/piping-tunnel-util"
 	"github.com/nwtgck/go-piping-tunnel/util"
 	"github.com/spf13/cobra"
@@ -93,7 +94,7 @@ func socksHandleWithYamux(socks5Server *socks5.Server, httpClient *http.Client,
 	}
 	var readWriteCloser io.ReadWriteCloser = duplex
 	if showProgress {
-		readWriteCloser = util.NewIOProgressReadWriteCloser(duplex, os.Stderr, makeProgressMessage)
+		readWriteCloser = io_progress.NewIOProgress(duplex, duplex, os.Stderr, makeProgressMessage)
 	}
 	yamuxSession, err := yamux.Server(readWriteCloser, nil)
 	if err != nil {

io_progress/io_progress.go

@@ -8,9 +8,10 @@ import (
 )
 
 type IOProgress struct {
-	io.Reader
+	io.ReadWriteCloser
 	CurrReadBytes   uint64
 	reader          io.Reader
+	writer          io.Writer
 	CurrWriteBytes  uint64
 	StartTime       time.Time
 	messageWriter   io.Writer
@@ -19,8 +20,8 @@ type IOProgress struct {
 	lastDisplayTime time.Time
 }
 
-func NewIOProgress(reader io.Reader, messageWriter io.Writer, makeMessage func(progress *IOProgress) string) IOProgress {
-	return IOProgress{reader: reader, messageWriter: messageWriter, StartTime: time.Now(), makeMessage: makeMessage}
+func NewIOProgress(reader io.Reader, writer io.Writer, messageWriter io.Writer, makeMessage func(progress *IOProgress) string) *IOProgress {
+	return &IOProgress{reader: reader, writer: writer, messageWriter: messageWriter, StartTime: time.Now(), makeMessage: makeMessage}
 }
 
 func (progress *IOProgress) Read(p []byte) (int, error) {
@@ -34,10 +35,25 @@ func (progress *IOProgress) Read(p []byte) (int, error) {
 }
 
 func (progress *IOProgress) Write(p []byte) (int, error) {
-	l := len(p)
-	progress.CurrWriteBytes += uint64(l)
+	n, err := progress.writer.Write(p)
+	if err != nil {
+		return n, err
+	}
+	progress.CurrWriteBytes += uint64(n)
 	progress.displayIfShould()
-	return l, nil
+	return n, nil
+}
+
+func (progress *IOProgress) Close() error {
+	if r, ok := progress.reader.(io.ReadCloser); ok {
+		if err := r.Close(); err != nil {
+			return err
+		}
+	}
+	if w, ok := progress.writer.(io.WriteCloser); ok {
+		return w.Close()
+	}
+	return nil
 }
 
 func (progress *IOProgress) displayIfShould() {

piping-tunnel-util/duplex.go

@@ -6,8 +6,9 @@ import (
 )
 
 type PipingDuplex struct {
-	uploadWriter   *io.PipeWriter
-	downloadReader *io.PipeReader
+	downloadReaderChan chan interface{} // io.ReadCloser or error
+	uploadWriter       *io.PipeWriter
+	downloadReader     io.ReadCloser
 }
 
 func NewPipingDuplex(httpClient *http.Client, headers []KeyValue, uploadPath, downloadPath string) (*PipingDuplex, error) {
@@ -24,29 +25,42 @@ func NewPipingDuplex(httpClient *http.Client, headers []KeyValue, uploadPath, do
 	if err != nil {
 		return nil, err
 	}
-	req, err = http.NewRequest("GET", downloadPath, nil)
-	if err != nil {
-		return nil, err
-	}
-	for _, kv := range headers {
-		req.Header.Set(kv.Key, kv.Value)
-	}
-	downloadPr, downloadPw := io.Pipe()
 
+	downloadReaderChan := make(chan interface{}, 1)
 	go func() {
-		res, _ := httpClient.Do(req)
-		// TODO: hard code
-		var buf = make([]byte, 16)
-		io.CopyBuffer(downloadPw, res.Body, buf)
+		req, err = http.NewRequest("GET", downloadPath, nil)
+		if err != nil {
+			downloadReaderChan <- err
+			return
+		}
+		for _, kv := range headers {
+			req.Header.Set(kv.Key, kv.Value)
+		}
+		res, err := httpClient.Do(req)
+		if err != nil {
+			downloadReaderChan <- err
+			return
+		}
+		downloadReaderChan <- res.Body
 	}()
 
 	return &PipingDuplex{
-		uploadWriter:   uploadPw,
-		downloadReader: downloadPr,
+		downloadReaderChan: downloadReaderChan,
+		uploadWriter:       uploadPw,
 	}, nil
 }
 
 func (pd *PipingDuplex) Read(b []byte) (n int, err error) {
+	if pd.downloadReaderChan != nil {
+		// Get io.ReaderCloser or error
+		result := <-pd.downloadReaderChan
+		// If result is error
+		if err, ok := result.(error); ok {
+			return 0, err
+		}
+		pd.downloadReader = result.(io.ReadCloser)
+		pd.downloadReaderChan = nil
+	}
 	return pd.downloadReader.Read(b)
 }
 

util/util.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"github.com/nwtgck/go-piping-tunnel/io_progress"
-	"io"
 	"math"
 	"net"
 	"net/http"
@@ -79,44 +77,3 @@ func HumanizeBytes(s float64) string {
 func logn(n, b float64) float64 {
 	return math.Log(n) / math.Log(b)
 }
-
-type IOProgressReadWriteCloser struct {
-	pw         *io.PipeWriter
-	progress   *io_progress.IOProgress
-	baseDuplex io.ReadWriteCloser
-}
-
-func NewIOProgressReadWriteCloser(duplex io.ReadWriteCloser, messageWriter io.Writer, makeMessage func(progress *io_progress.IOProgress) string) IOProgressReadWriteCloser {
-	pr, pw := io.Pipe()
-	p := io_progress.NewIOProgress(pr, messageWriter, makeMessage)
-	go func() {
-		// TODO: hard code
-		var buf = make([]byte, 16)
-		io.CopyBuffer(duplex, &p, buf)
-	}()
-	return IOProgressReadWriteCloser{
-		pw:         pw,
-		progress:   &p,
-		baseDuplex: duplex,
-	}
-}
-
-func (p IOProgressReadWriteCloser) Read(b []byte) (n int, err error) {
-	n, err = p.baseDuplex.Read(b)
-	if err != nil {
-		return n, err
-	}
-	_, err = p.progress.Write(b[:n])
-	if err != nil {
-		return n, err
-	}
-	return n, nil
-}
-
-func (p IOProgressReadWriteCloser) Write(b []byte) (n int, err error) {
-	return p.pw.Write(b)
-}
-
-func (p IOProgressReadWriteCloser) Close() error {
-	return p.baseDuplex.Close()
-}

version/version.go

@@ -1,3 +1,3 @@
 package version
 
-const Version = "0.5.0"
+const Version = "0.6.0"