merge http.Connection into net.Connection

app/commander/outbound.go

@@ -80,7 +80,7 @@ func (co *Outbound) Dispatch(ctx context.Context, link *core.Link) {
 	}
 
 	closeSignal := signal.NewNotifier()
-	c := net.NewConnection(net.ConnectionInputMulti(link.Writer), net.ConnectionOutputMulti(link.Reader), net.ConnectionOnClose(closeSignal))
+	c := net.NewConnection(net.ConnectionInputMulti(link.Writer), net.ConnectionOutputMulti(link.Reader), net.ConnectionOnClose(signal.NotifyClose(closeSignal)))
 	co.listener.add(c)
 	co.access.RUnlock()
 	<-closeSignal.Wait()

common/net/connection.go

@@ -48,15 +48,23 @@ func ConnectionOutputMulti(reader buf.Reader) ConnectionOption {
 	}
 }
 
-func ConnectionOnClose(s *signal.Notifier) ConnectionOption {
+func ConnectionOnClose(n io.Closer) ConnectionOption {
 	return func(c *connection) {
-		c.onClose = s
+		c.onClose = n
 	}
 }
 
 func NewConnection(opts ...ConnectionOption) net.Conn {
 	c := &connection{
 		done: signal.NewDone(),
+		local: &net.TCPAddr{
+			IP:   []byte{0, 0, 0, 0},
+			Port: 0,
+		},
+		remote: &net.TCPAddr{
+			IP:   []byte{0, 0, 0, 0},
+			Port: 0,
+		},
 	}
 
 	for _, opt := range opts {
@@ -70,7 +78,7 @@ type connection struct {
 	reader  *buf.BufferedReader
 	writer  buf.Writer
 	done    *signal.Done
-	onClose *signal.Notifier
+	onClose io.Closer
 	local   Addr
 	remote  Addr
 }
@@ -110,7 +118,7 @@ func (c *connection) Close() error {
 	common.Close(c.reader)
 	common.Close(c.writer)
 	if c.onClose != nil {
-		c.onClose.Signal()
+		return c.onClose.Close()
 	}
 
 	return nil

common/signal/notifier.go

@@ -1,5 +1,7 @@
 package signal
 
+import "io"
+
 // Notifier is a utility for notifying changes. The change producer may notify changes multiple time, and the consumer may get notified asynchronously.
 type Notifier struct {
 	c chan struct{}
@@ -24,3 +26,18 @@ func (n *Notifier) Signal() {
 func (n *Notifier) Wait() <-chan struct{} {
 	return n.c
 }
+
+type nCloser struct {
+	n *Notifier
+}
+
+func (c *nCloser) Close() error {
+	c.n.Signal()
+	return nil
+}
+
+func NotifyClose(n *Notifier) io.Closer {
+	return &nCloser{
+		n: n,
+	}
+}

transport/internet/http/connection.go

@@ -1,49 +0,0 @@
-package http
-
-import (
-	"io"
-	"time"
-
-	"v2ray.com/core/common"
-	"v2ray.com/core/common/net"
-)
-
-type Connection struct {
-	Reader io.Reader
-	Writer io.Writer
-	Closer common.Closable
-	Local  net.Addr
-	Remote net.Addr
-}
-
-func (c *Connection) Read(b []byte) (int, error) {
-	return c.Reader.Read(b)
-}
-
-func (c *Connection) Write(b []byte) (int, error) {
-	return c.Writer.Write(b)
-}
-
-func (c *Connection) Close() error {
-	return c.Closer.Close()
-}
-
-func (c *Connection) LocalAddr() net.Addr {
-	return c.Local
-}
-
-func (c *Connection) RemoteAddr() net.Addr {
-	return c.Remote
-}
-
-func (c *Connection) SetDeadline(t time.Time) error {
-	return nil
-}
-
-func (c *Connection) SetReadDeadline(t time.Time) error {
-	return nil
-}
-
-func (c *Connection) SetWriteDeadline(t time.Time) error {
-	return nil
-}

transport/internet/http/dialer.go

@@ -112,19 +112,11 @@ func Dial(ctx context.Context, dest net.Destination) (internet.Connection, error
 
 	bwriter := buf.NewBufferedWriter(pwriter)
 	common.Must(bwriter.SetBuffered(false))
-	return &Connection{
+	return net.NewConnection(
-		Reader: response.Body,
+		net.ConnectionOutput(response.Body),
-		Writer: bwriter,
+		net.ConnectionInput(bwriter),
-		Closer: common.NewChainedClosable(breader, bwriter, response.Body),
+		net.ConnectionOnClose(common.NewChainedClosable(breader, bwriter, response.Body)),
-		Local: &net.TCPAddr{
+	), nil
-			IP:   []byte{0, 0, 0, 0},
-			Port: 0,
-		},
-		Remote: &net.TCPAddr{
-			IP:   []byte{0, 0, 0, 0},
-			Port: 0,
-		},
-	}, nil
 }
 
 func init() {

transport/internet/http/hub.go

@@ -64,13 +64,14 @@ func (l *Listener) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 		f.Flush()
 	}
 	done := signal.NewDone()
-	l.handler(&Connection{
+	conn := net.NewConnection(
-		Reader: request.Body,
+		net.ConnectionOutput(request.Body),
-		Writer: flushWriter{w: writer, d: done},
+		net.ConnectionInput(flushWriter{w: writer, d: done}),
-		Closer: common.NewChainedClosable(done, request.Body),
+		net.ConnectionOnClose(common.NewChainedClosable(done, request.Body)),
-		Local:  l.Addr(),
+		net.ConnectionLocalAddr(l.Addr()),
-		Remote: l.Addr(),
+		net.ConnectionRemoteAddr(l.Addr()),
-	})
+	)
+	l.handler(conn)
 	<-done.Wait()
 }