rename CloseError() to Interrupt()

app/commander/outbound.go

@@ -8,7 +8,6 @@ import (
 	"v2ray.com/core/common/net"
 	"v2ray.com/core/common/signal/done"
 	"v2ray.com/core/transport"
-	"v2ray.com/core/transport/pipe"
 )
 
 // OutboundListener is a net.Listener for listening gRPC connections.
@@ -73,8 +72,8 @@ func (co *Outbound) Dispatch(ctx context.Context, link *transport.Link) {
 	co.access.RLock()
 
 	if co.closed {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		co.access.RUnlock()
 		return
 	}

app/dispatcher/default.go

@@ -77,13 +77,13 @@ func (r *cachedReader) ReadMultiBufferTimeout(timeout time.Duration) (buf.MultiB
 	return r.reader.ReadMultiBufferTimeout(timeout)
 }
 
-func (r *cachedReader) CloseError() {
+func (r *cachedReader) Interrupt() {
 	r.Lock()
 	if r.cache != nil {
 		r.cache = buf.ReleaseMulti(r.cache)
 	}
 	r.Unlock()
-	r.reader.CloseError()
+	r.reader.Interrupt()
 }
 
 // DefaultDispatcher is a default implementation of Dispatcher.
@@ -267,7 +267,7 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.
 	if dispatcher == nil {
 		newError("default outbound handler not exist").WriteToLog(session.ExportIDToError(ctx))
 		common.Close(link.Writer)
-		pipe.CloseError(link.Reader)
+		common.Interrupt(link.Reader)
 		return
 	}
 

app/dispatcher/stats.go

@@ -4,7 +4,6 @@ import (
 	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 	"v2ray.com/core/features/stats"
-	"v2ray.com/core/transport/pipe"
 )
 
 type SizeStatWriter struct {
@@ -21,6 +20,6 @@ func (w *SizeStatWriter) Close() error {
 	return common.Close(w.Writer)
 }
 
-func (w *SizeStatWriter) CloseError() {
-	pipe.CloseError(w.Writer)
+func (w *SizeStatWriter) Interrupt() {
+	common.Interrupt(w.Writer)
 }

app/proxyman/outbound/handler.go

@@ -100,17 +100,17 @@ func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) {
 	if h.mux != nil {
 		if err := h.mux.Dispatch(ctx, link); err != nil {
 			newError("failed to process mux outbound traffic").Base(err).WriteToLog(session.ExportIDToError(ctx))
-			pipe.CloseError(link.Writer)
+			common.Interrupt(link.Writer)
 		}
 	} else {
 		if err := h.proxy.Process(ctx, link, h); err != nil {
 			// Ensure outbound ray is properly closed.
 			newError("failed to process outbound traffic").Base(err).WriteToLog(session.ExportIDToError(ctx))
-			pipe.CloseError(link.Writer)
+			common.Interrupt(link.Writer)
 		} else {
 			common.Must(common.Close(link.Writer))
 		}
-		pipe.CloseError(link.Reader)
+		common.Interrupt(link.Reader)
 	}
 }
 

app/reverse/portal.go

@@ -97,7 +97,7 @@ func (o *Outbound) Tag() string {
 func (o *Outbound) Dispatch(ctx context.Context, link *transport.Link) {
 	if err := o.portal.HandleConnection(ctx, link); err != nil {
 		newError("failed to process reverse connection").Base(err).WriteToLog(session.ExportIDToError(ctx))
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Writer)
 	}
 }
 
@@ -244,7 +244,7 @@ func (w *PortalWorker) heartbeat() error {
 
 		defer func() {
 			common.Close(w.writer)
-			pipe.CloseError(w.reader)
+			common.Interrupt(w.reader)
 			w.writer = nil
 		}()
 	}

common/interfaces.go

@@ -6,6 +6,11 @@ type Closable interface {
 	Close() error
 }
 
+// Interruptible is an interface for objects that can be stopped before its completion.
+type Interruptible interface {
+	Interrupt()
+}
+
 // Close closes the obj if it is a Closable.
 func Close(obj interface{}) error {
 	if c, ok := obj.(Closable); ok {
@@ -14,6 +19,15 @@ func Close(obj interface{}) error {
 	return nil
 }
 
+// Interrupt calls Interrupt() if object implements Interruptible interface, or Close() if the object implements Closable interface.
+func Interrupt(obj interface{}) error {
+	if c, ok := obj.(Interruptible); ok {
+		c.Interrupt()
+		return nil
+	}
+	return Close(obj)
+}
+
 // Runnable is the interface for objects that can start to work and stop on demand.
 type Runnable interface {
 	// Start starts the runnable object. Upon the method returning nil, the object begins to function properly.

common/mux/client.go

@@ -213,8 +213,8 @@ func (m *ClientWorker) monitor() {
 		select {
 		case <-m.done.Wait():
 			m.sessionManager.Close()
-			common.Close(m.link.Writer)    // nolint: errcheck
-			pipe.CloseError(m.link.Reader) // nolint: errcheck
+			common.Close(m.link.Writer)     // nolint: errcheck
+			common.Interrupt(m.link.Reader) // nolint: errcheck
 			return
 		case <-timer.C:
 			size := m.sessionManager.Size()
@@ -253,14 +253,14 @@ func fetchInput(ctx context.Context, s *Session, output buf.Writer) {
 	if err := writeFirstPayload(s.input, writer); err != nil {
 		newError("failed to write first payload").Base(err).WriteToLog(session.ExportIDToError(ctx))
 		writer.hasError = true
-		pipe.CloseError(s.input)
+		common.Interrupt(s.input)
 		return
 	}
 
 	if err := buf.Copy(s.input, writer); err != nil {
 		newError("failed to fetch all input").Base(err).WriteToLog(session.ExportIDToError(ctx))
 		writer.hasError = true
-		pipe.CloseError(s.input)
+		common.Interrupt(s.input)
 		return
 	}
 }
@@ -339,7 +339,7 @@ func (m *ClientWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 		closingWriter.Close()
 
 		drainErr := buf.Copy(rr, buf.Discard)
-		pipe.CloseError(s.input)
+		common.Interrupt(s.input)
 		s.Close()
 		return drainErr
 	}
@@ -350,8 +350,8 @@ func (m *ClientWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 func (m *ClientWorker) handleStatusEnd(meta *FrameMetadata, reader *buf.BufferedReader) error {
 	if s, found := m.sessionManager.Get(meta.SessionID); found {
 		if meta.Option.Has(OptionError) {
-			pipe.CloseError(s.input)
-			pipe.CloseError(s.output)
+			common.Interrupt(s.input)
+			common.Interrupt(s.output)
 		}
 		s.Close()
 	}

common/mux/server.go

@@ -5,6 +5,7 @@ import (
 	"io"
 
 	"v2ray.com/core"
+	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 	"v2ray.com/core/common/errors"
 	"v2ray.com/core/common/log"
@@ -146,7 +147,7 @@ func (w *ServerWorker) handleStatusNew(ctx context.Context, meta *FrameMetadata,
 	rr := s.NewReader(reader)
 	if err := buf.Copy(rr, s.output); err != nil {
 		buf.Copy(rr, buf.Discard)
-		pipe.CloseError(s.input)
+		common.Interrupt(s.input)
 		return s.Close()
 	}
 	return nil
@@ -177,7 +178,7 @@ func (w *ServerWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 		closingWriter.Close()
 
 		drainErr := buf.Copy(rr, buf.Discard)
-		pipe.CloseError(s.input)
+		common.Interrupt(s.input)
 		s.Close()
 		return drainErr
 	}
@@ -188,8 +189,8 @@ func (w *ServerWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 func (w *ServerWorker) handleStatusEnd(meta *FrameMetadata, reader *buf.BufferedReader) error {
 	if s, found := w.sessionManager.Get(meta.SessionID); found {
 		if meta.Option.Has(OptionError) {
-			pipe.CloseError(s.input)
-			pipe.CloseError(s.output)
+			common.Interrupt(s.input)
+			common.Interrupt(s.output)
 		}
 		s.Close()
 	}
@@ -241,7 +242,7 @@ func (w *ServerWorker) run(ctx context.Context) {
 			if err != nil {
 				if errors.Cause(err) != io.EOF {
 					newError("unexpected EOF").Base(err).WriteToLog(session.ExportIDToError(ctx))
-					pipe.CloseError(input)
+					common.Interrupt(input)
 				}
 				return
 			}

proxy/blackhole/blackhole.go

@@ -10,7 +10,6 @@ import (
 	"v2ray.com/core/common"
 	"v2ray.com/core/transport"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/pipe"
 )
 
 // Handler is an outbound connection that silently swallow the entire payload.
@@ -36,7 +35,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 		// Sleep a little here to make sure the response is sent to client.
 		time.Sleep(time.Second)
 	}
-	pipe.CloseError(link.Writer)
+	common.Interrupt(link.Writer)
 	return nil
 }
 

proxy/dokodemo/dokodemo.go

@@ -16,7 +16,6 @@ import (
 	"v2ray.com/core/features/policy"
 	"v2ray.com/core/features/routing"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/pipe"
 )
 
 func init() {
@@ -148,8 +147,8 @@ func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn in
 	}
 
 	if err := task.Run(ctx, task.OnSuccess(requestDone, task.Close(link.Writer)), responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

proxy/http/server.go

@@ -22,7 +22,6 @@ import (
 	"v2ray.com/core/features/policy"
 	"v2ray.com/core/features/routing"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/pipe"
 )
 
 // Server is an HTTP proxy server.
@@ -51,6 +50,7 @@ func (s *Server) policy() policy.Session {
 	return p
 }
 
+// Network implements proxy.Inbound.
 func (*Server) Network() []net.Network {
 	return []net.Network{net.Network_TCP}
 }
@@ -191,8 +191,8 @@ func (s *Server) handleConnect(ctx context.Context, request *http.Request, reade
 
 	var closeWriter = task.OnSuccess(requestDone, task.Close(link.Writer))
 	if err := task.Run(ctx, closeWriter, responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 
@@ -287,8 +287,8 @@ func (s *Server) handlePlainHTTP(ctx context.Context, request *http.Request, wri
 	}
 
 	if err := task.Run(ctx, requestDone, responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

proxy/mtproto/server.go

@@ -17,7 +17,6 @@ import (
 	"v2ray.com/core/features/policy"
 	"v2ray.com/core/features/routing"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/pipe"
 )
 
 var (
@@ -143,8 +142,8 @@ func (s *Server) Process(ctx context.Context, network net.Network, conn internet
 
 	var responseDoneAndCloseWriter = task.OnSuccess(response, task.Close(link.Writer))
 	if err := task.Run(ctx, request, responseDoneAndCloseWriter); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

proxy/shadowsocks/server.go

@@ -17,7 +17,6 @@ import (
 	"v2ray.com/core/features/routing"
 	"v2ray.com/core/transport/internet"
 	"v2ray.com/core/transport/internet/udp"
-	"v2ray.com/core/transport/pipe"
 )
 
 type Server struct {
@@ -231,8 +230,8 @@ func (s *Server) handleConnection(ctx context.Context, conn internet.Connection,
 
 	var requestDoneAndCloseWriter = task.OnSuccess(requestDone, task.Close(link.Writer))
 	if err := task.Run(ctx, requestDoneAndCloseWriter, responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

proxy/socks/server.go

@@ -19,7 +19,6 @@ import (
 	"v2ray.com/core/features/routing"
 	"v2ray.com/core/transport/internet"
 	"v2ray.com/core/transport/internet/udp"
-	"v2ray.com/core/transport/pipe"
 )
 
 // Server is a SOCKS 5 proxy server
@@ -166,8 +165,8 @@ func (s *Server) transport(ctx context.Context, reader io.Reader, writer io.Writ
 
 	var requestDonePost = task.OnSuccess(requestDone, task.Close(link.Writer))
 	if err := task.Run(ctx, requestDonePost, responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

proxy/vmess/inbound/inbound.go

@@ -26,7 +26,6 @@ import (
 	"v2ray.com/core/proxy/vmess"
 	"v2ray.com/core/proxy/vmess/encoding"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/pipe"
 )
 
 type userByEmail struct {
@@ -304,8 +303,8 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection i
 
 	var requestDonePost = task.OnSuccess(requestDone, task.Close(link.Writer))
 	if err := task.Run(ctx, requestDonePost, responseDone); err != nil {
-		pipe.CloseError(link.Reader)
-		pipe.CloseError(link.Writer)
+		common.Interrupt(link.Reader)
+		common.Interrupt(link.Writer)
 		return newError("connection ends").Base(err)
 	}
 

testing/servers/tcp/tcp.go

@@ -81,7 +81,7 @@ func (server *Server) handleConnection(conn net.Conn) {
 			}
 		}
 	}, func() error {
-		defer pReader.CloseError()
+		defer pReader.Interrupt()
 
 		w := buf.NewWriter(conn)
 		for {

transport/pipe/impl.go

@@ -182,7 +182,8 @@ func (p *pipe) Close() error {
 	return nil
 }
 
-func (p *pipe) CloseError() {
+// Interrupt implements common.Interruptible.
+func (p *pipe) Interrupt() {
 	p.Lock()
 	defer p.Unlock()
 

transport/pipe/pipe.go

@@ -67,14 +67,3 @@ func New(opts ...Option) (*Reader, *Writer) {
 			pipe: p,
 		}
 }
-
-type closeError interface {
-	CloseError()
-}
-
-// CloseError invokes CloseError() method if the object is either Reader or Writer.
-func CloseError(v interface{}) {
-	if c, ok := v.(closeError); ok {
-		c.CloseError()
-	}
-}

transport/pipe/pipe_test.go

@@ -32,7 +32,7 @@ func TestPipeReadWrite(t *testing.T) {
 	assert(rb.String(), Equals, "abcdefg")
 }
 
-func TestPipeCloseError(t *testing.T) {
+func TestPipeInterrupt(t *testing.T) {
 	assert := With(t)
 
 	pReader, pWriter := New(WithSizeLimit(1024))
@@ -40,7 +40,7 @@ func TestPipeCloseError(t *testing.T) {
 	b := buf.New()
 	b.Write(payload)
 	assert(pWriter.WriteMultiBuffer(buf.MultiBuffer{b}), IsNil)
-	pWriter.CloseError()
+	pWriter.Interrupt()
 
 	rb, err := pReader.ReadMultiBuffer()
 	assert(err, Equals, io.ErrClosedPipe)

transport/pipe/reader.go

@@ -21,7 +21,7 @@ func (r *Reader) ReadMultiBufferTimeout(d time.Duration) (buf.MultiBuffer, error
 	return r.pipe.ReadMultiBufferTimeout(d)
 }
 
-// CloseError sets the pipe to error state. Both reading and writing from/to the pipe will return io.ErrClosedPipe.
-func (r *Reader) CloseError() {
-	r.pipe.CloseError()
+// Interrupt implements common.Interruptible.
+func (r *Reader) Interrupt() {
+	r.pipe.Interrupt()
 }

transport/pipe/writer.go

@@ -19,7 +19,7 @@ func (w *Writer) Close() error {
 	return w.pipe.Close()
 }
 
-// CloseError sets the pipe to error state. Both reading and writing from/to the pipe will return io.ErrClosedPipe.
-func (w *Writer) CloseError() {
-	w.pipe.CloseError()
+// Interrupt implements common.Interruptible.
+func (w *Writer) Interrupt() {
+	w.pipe.Interrupt()
 }