Repository: goplus/gop
Pull request #1558 (2023-11-26 02:54:55)

jsonrpc: stdio


Patch:
diff --git a/x/fakenet/conn.go b/x/fakenet/conn.go
new file mode 100644
index 00000000..c9cdaf27
--- /dev/null
+++ b/x/fakenet/conn.go
@@ -0,0 +1,129 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package fakenet
+
+import (
+	"io"
+	"net"
+	"sync"
+	"time"
+)
+
+// NewConn returns a net.Conn built on top of the supplied reader and writer.
+// It decouples the read and write on the conn from the underlying stream
+// to enable Close to abort ones that are in progress.
+// It's primary use is to fake a network connection from stdin and stdout.
+func NewConn(name string, in io.ReadCloser, out io.WriteCloser) net.Conn {
+	c := &fakeConn{
+		name:   name,
+		reader: newFeeder(in.Read),
+		writer: newFeeder(out.Write),
+		in:     in,
+		out:    out,
+	}
+	go c.reader.run()
+	go c.writer.run()
+	return c
+}
+
+type fakeConn struct {
+	name   string
+	reader *connFeeder
+	writer *connFeeder
+	in     io.ReadCloser
+	out    io.WriteCloser
+}
+
+type fakeAddr string
+
+// connFeeder serializes calls to the source function (io.Reader.Read or
+// io.Writer.Write) by delegating them to a channel. This also allows calls to
+// be intercepted when the connection is closed, and cancelled early if the
+// connection is closed while the calls are still outstanding.
+type connFeeder struct {
+	source func([]byte) (int, error)
+	input  chan []byte
+	result chan feedResult
+	mu     sync.Mutex
+	closed bool
+	done   chan struct{}
+}
+
+type feedResult struct {
+	n   int
+	err error
+}
+
+func (c *fakeConn) Close() error {
+	c.reader.close()
+	c.writer.close()
+	c.in.Close()
+	c.out.Close()
+	return nil
+}
+
+func (c *fakeConn) Read(b []byte) (n int, err error)   { return c.reader.do(b) }
+func (c *fakeConn) Write(b []byte) (n int, err error)  { return c.writer.do(b) }
+func (c *fakeConn) LocalAddr() net.Addr                { return fakeAddr(c.name) }
+func (c *fakeConn) RemoteAddr() net.Addr               { return fakeAddr(c.name) }
+func (c *fakeConn) SetDeadline(t time.Time) error      { return nil }
+func (c *fakeConn) SetReadDeadline(t time.Time) error  { return nil }
+func (c *fakeConn) SetWriteDeadline(t time.Time) error { return nil }
+func (a fakeAddr) Network() string                     { return "fake" }
+func (a fakeAddr) String() string                      { return string(a) }
+
+func newFeeder(source func([]byte) (int, error)) *connFeeder {
+	return &connFeeder{
+		source: source,
+		input:  make(chan []byte),
+		result: make(chan feedResult),
+		done:   make(chan struct{}),
+	}
+}
+
+func (f *connFeeder) close() {
+	f.mu.Lock()
+	if !f.closed {
+		f.closed = true
+		close(f.done)
+	}
+	f.mu.Unlock()
+}
+
+func (f *connFeeder) do(b []byte) (n int, err error) {
+	// send the request to the worker
+	select {
+	case f.input <- b:
+	case <-f.done:
+		return 0, io.EOF
+	}
+	// get the result from the worker
+	select {
+	case r := <-f.result:
+		return r.n, r.err
+	case <-f.done:
+		return 0, io.EOF
+	}
+}
+
+func (f *connFeeder) run() {
+	var b []byte
+	for {
+		// wait for an input request
+		select {
+		case b = <-f.input:
+		case <-f.done:
+			return
+		}
+		// invoke the underlying method
+		n, err := f.source(b)
+		// send the result back to the requester
+		select {
+		case f.result <- feedResult{n: n, err: err}:
+		case <-f.done:
+			return
+		}
+	}
+}
diff --git a/x/jsonrpc2/stdio/server.go b/x/jsonrpc2/stdio/server.go
index 9e4969dd..8ae33044 100644
--- a/x/jsonrpc2/stdio/server.go
+++ b/x/jsonrpc2/stdio/server.go
@@ -18,162 +18,44 @@ package stdio
 
 import (
 	"context"
-	"errors"
 	"io"
-	"log"
-	"net"
 	"os"
 	"sync/atomic"
-	"time"
 
+	"github.com/goplus/gop/x/fakenet"
 	"github.com/goplus/gop/x/jsonrpc2"
+	"github.com/goplus/gop/x/jsonrpc2/jsonrpc2test"
 )
 
 // -----------------------------------------------------------------------------
 
-var (
-	ErrTooManyConnections = errors.New("too many connections")
-)
-
-var (
-	connCnt int32
-)
-
-type serverConn struct {
-	closed bool
-}
-
-func (p *serverConn) Read(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	n, err = os.Stdin.Read(b)
-	for n == 0 { // retry (to support localDailer)
-		time.Sleep(time.Millisecond)
-		n, err = os.Stdin.Read(b)
-	}
-	return
-}
-
-func (p *serverConn) Write(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	return os.Stdout.Write(b)
-}
-
-func (p *serverConn) Close() error {
-	if jsonrpc2.Verbose {
-		log.Println("==> stdio.conn.Close")
-	}
-	if p.closed {
-		return net.ErrClosed
-	}
-	p.closed = true
-	atomic.AddInt32(&connCnt, -1)
-	return nil
-}
-
-// -----------------------------------------------------------------------------
-
-type newConn struct {
-	in, out *os.File
-	r, w    *os.File
-	oStdin  *os.File
-	oStdout *os.File
-	closed  bool
-}
-
-func (p *newConn) Read(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	return p.in.Read(b)
-}
-
-func (p *newConn) Write(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	return p.out.Write(b)
-}
-
-func (p *newConn) Close() error {
-	if jsonrpc2.Verbose {
-		log.Println("==> stdio.newConn.Close")
-	}
-	if p.closed {
-		return net.ErrClosed
-	}
-	p.closed = true
-	atomic.AddInt32(&dailCnt, -1)
-	os.Stdin, os.Stdout = p.oStdin, p.oStdout
-	p.in.Close()
-	p.w.Close()
-	p.r.Close()
-	p.out.Close()
-	return nil
-}
-
-// -----------------------------------------------------------------------------
-
 type listener struct {
-	closed    bool
-	localDail bool
 }
 
 func (p *listener) Close() error {
-	if jsonrpc2.Verbose {
-		log.Println("==> stdio.listener.Close")
-	}
-	if p.closed {
-		return net.ErrClosed
-	}
-	p.closed = true
 	return nil
 }
 
 // Accept blocks waiting for an incoming connection to the listener.
 func (p *listener) Accept(context.Context) (io.ReadWriteCloser, error) {
-	if atomic.AddInt32(&connCnt, 1) != 1 {
-		atomic.AddInt32(&connCnt, -1)
+	connCnt := &connCnt[server]
+	if atomic.AddInt32(connCnt, 1) != 1 {
+		atomic.AddInt32(connCnt, -1)
 		return nil, ErrTooManyConnections
 	}
-	return new(serverConn), nil
-}
-
-// Dial returns a new communication byte stream to a listening server.
-func (p *listener) Dial(ctx context.Context) (ret io.ReadWriteCloser, err error) {
-	if atomic.AddInt32(&dailCnt, 1) != 1 {
-		atomic.AddInt32(&dailCnt, -1)
-		return nil, ErrTooManyConnections
-	}
-	in, w, err := os.Pipe()
-	if err != nil {
-		return
-	}
-	r, out, err := os.Pipe()
-	if err != nil {
-		in.Close()
-		w.Close()
-		return
-	}
-	oStdin, oStdout := os.Stdin, os.Stdout
-	os.Stdin, os.Stdout = r, w
-	return &newConn{in: in, out: out, r: r, w: w, oStdin: oStdin, oStdout: oStdout}, nil
+	return fakenet.NewConn("stdio", os.Stdin, os.Stdout), nil
 }
 
 func (p *listener) Dialer() jsonrpc2.Dialer {
-	if p.localDail {
-		return p
-	}
 	return nil
 }
 
 // Listener returns a jsonrpc2.Listener based on stdin and stdout.
 func Listener(allowLocalDail bool) jsonrpc2.Listener {
-	l := &listener{localDail: allowLocalDail}
-	return l
+	if allowLocalDail {
+		return jsonrpc2test.NetPipeListener()
+	}
+	return &listener{}
 }
 
 // -----------------------------------------------------------------------------
diff --git a/x/jsonrpc2/stdio/client.go b/x/jsonrpc2/stdio/stdio.go
similarity index 64%
rename from x/jsonrpc2/stdio/client.go
rename to x/jsonrpc2/stdio/stdio.go
index 72b98968..61817e76 100644
--- a/x/jsonrpc2/stdio/client.go
+++ b/x/jsonrpc2/stdio/stdio.go
@@ -18,69 +18,50 @@ package stdio
 
 import (
 	"context"
+	"errors"
 	"io"
-	"log"
-	"net"
 	"sync/atomic"
 
+	"github.com/goplus/gop/x/fakenet"
 	"github.com/goplus/gop/x/jsonrpc2"
 )
 
-// -----------------------------------------------------------------------------
-
-type dialConn struct {
-	in     io.ReadCloser
-	out    io.WriteCloser
-	closed bool
-}
-
-func (p *dialConn) Read(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	return p.in.Read(b)
-}
+var (
+	ErrTooManyConnections = errors.New("too many connections")
+)
 
-func (p *dialConn) Write(b []byte) (n int, err error) {
-	if p.closed {
-		return 0, net.ErrClosed
-	}
-	return p.out.Write(b)
-}
+const (
+	client = iota
+	server
+)
 
 var (
-	dailCnt int32
+	connCnt [2]int32
 )
 
-func (p *dialConn) Close() error {
-	if jsonrpc2.Verbose {
-		log.Println("==> stdio.dialConn.Close")
-	}
-	if p.closed {
-		return net.ErrClosed
-	}
-	p.closed = true
-	atomic.AddInt32(&dailCnt, -1)
-	p.in.Close()
-	p.out.Close()
-	return nil
+// -----------------------------------------------------------------------------
+
+type dialer struct {
+	in  io.ReadCloser
+	out io.WriteCloser
 }
 
 // Dial returns a new communication byte stream to a listening server.
-func (p *dialConn) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
-	if atomic.AddInt32(&dailCnt, 1) != 1 {
-		atomic.AddInt32(&dailCnt, -1)
+func (p *dialer) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
+	dailCnt := &connCnt[client]
+	if atomic.AddInt32(dailCnt, 1) != 1 {
+		atomic.AddInt32(dailCnt, -1)
 		return nil, ErrTooManyConnections
 	}
-	return p, nil
+	return fakenet.NewConn("stdio.dialer", p.in, p.out), nil
 }
 
-// Dialer returns a jsonrpc2.Dialer based on stdin and stdout.
+// Dialer returns a jsonrpc2.Dialer based on in and out.
 func Dialer(in io.ReadCloser, out io.WriteCloser) jsonrpc2.Dialer {
-	return &dialConn{in: in, out: out}
+	return &dialer{in: in, out: out}
 }
 
-// Dial makes a new connection based on stdin and stdout, wraps the returned
+// Dial makes a new connection based on in and out, wraps the returned
 // reader and writer using the framer to make a stream, and then builds a
 // connection on top of that stream using the binder.
 //


Test patch:
diff --git a/x/jsonrpc2/jsonrpc2test/jsonrpc2_test.go b/x/jsonrpc2/jsonrpc2test/jsonrpc2_test.go
index 5093c5ed..72abe2ca 100644
--- a/x/jsonrpc2/jsonrpc2test/jsonrpc2_test.go
+++ b/x/jsonrpc2/jsonrpc2test/jsonrpc2_test.go
@@ -16,9 +16,6 @@ import (
 func TestNetPipe(t *testing.T) {
 	jsonrpc2.SetDebug(jsonrpc2.DbgFlagCall)
 	ctx := context.Background()
-	listener, err := jsonrpc2test.NetPipeListener(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
+	listener := jsonrpc2test.NetPipeListener()
 	cases.Test(t, ctx, listener, jsonrpc2.HeaderFramer(), true)
 }
diff --git a/x/jsonrpc2/jsonrpc2test/pipe.go b/x/jsonrpc2/jsonrpc2test/pipe.go
index 269d208d..f1569e8a 100644
--- a/x/jsonrpc2/jsonrpc2test/pipe.go
+++ b/x/jsonrpc2/jsonrpc2test/pipe.go
@@ -32,11 +32,11 @@ import (
 // It is only possibly to connect to it using the Dialer returned by the
 // Dialer method, each call to that method will generate a new pipe the other
 // side of which will be returned from the Accept call.
-func NetPipeListener(ctx context.Context) (jsonrpc2.Listener, error) {
+func NetPipeListener() jsonrpc2.Listener {
 	return &netPiper{
 		done:   make(chan struct{}),
 		dialed: make(chan io.ReadWriteCloser),
-	}, nil
+	}
 }
 
 // netPiper is the implementation of Listener build on top of net.Pipes.
diff --git a/x/jsonrpc2/stdio/jsonrpc2_test.go b/x/jsonrpc2/stdio/jsonrpc2_test.go
deleted file mode 100644
index 1cc9621c..00000000
--- a/x/jsonrpc2/stdio/jsonrpc2_test.go
+++ /dev/null
@@ -1,32 +0,0 @@
-// Copyright 2018 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package stdio_test
-
-import (
-	"context"
-	"io"
-	"testing"
-
-	"github.com/goplus/gop/x/jsonrpc2"
-	"github.com/goplus/gop/x/jsonrpc2/jsonrpc2test/cases"
-	"github.com/goplus/gop/x/jsonrpc2/stdio"
-)
-
-func TestStdio(t *testing.T) {
-	ctx := context.Background()
-	listener := stdio.Listener(true)
-	cases.Test(t, ctx, listener, nil, false)
-}
-
-func TestDial(t *testing.T) {
-	r, w := io.Pipe()
-	stdio.Dial(r, w, jsonrpc2.BinderFunc(
-		func(ctx context.Context, c *jsonrpc2.Connection) jsonrpc2.ConnectionOptions {
-			return jsonrpc2.ConnectionOptions{}
-		}), nil)
-	if _, err := stdio.Dialer(r, w).Dial(context.Background()); err != stdio.ErrTooManyConnections {
-		t.Fatal("stdio.Dialer:", err)
-	}
-}
diff --git a/x/jsonrpc2/stdio/stdio_test.go b/x/jsonrpc2/stdio/stdio_test.go
deleted file mode 100644
index b5bc4f72..00000000
--- a/x/jsonrpc2/stdio/stdio_test.go
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Copyright (c) 2023 The GoPlus Authors (goplus.org). All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package stdio
-
-import (
-	"context"
-	"io"
-	"net"
-	"sync/atomic"
-	"testing"
-
-	"github.com/goplus/gop/x/jsonrpc2"
-)
-
-func init() {
-	jsonrpc2.SetDebug(jsonrpc2.DbgFlagAll)
-}
-
-func TestServerConn(t *testing.T) {
-	c := &serverConn{}
-	atomic.AddInt32(&connCnt, 1)
-	if err := c.Close(); err != nil {
-		t.Fatal("Close failed:", err)
-	}
-	if err := c.Close(); err != net.ErrClosed {
-		t.Fatal("Close:", err)
-	}
-	if _, err := c.Read(nil); err != net.ErrClosed {
-		t.Fatal("Read:", err)
-	}
-	if _, err := c.Write(nil); err != net.ErrClosed {
-		t.Fatal("Write:", err)
-	}
-}
-
-func TestDialConn(t *testing.T) {
-	r, w := io.Pipe()
-	c := &dialConn{in: r, out: w}
-	c.Dial(context.Background())
-	if err := c.Close(); err != nil {
-		t.Fatal("Close failed:", err)
-	}
-	if err := c.Close(); err != net.ErrClosed {
-		t.Fatal("Close:", err)
-	}
-	if _, err := c.Read(nil); err != net.ErrClosed {
-		t.Fatal("Read:", err)
-	}
-	if _, err := c.Write(nil); err != net.ErrClosed {
-		t.Fatal("Write:", err)
-	}
-	c.closed = false
-	c.Write(nil)
-}
-
-func TestListener(t *testing.T) {
-	l := new(listener)
-	if l.Dialer() != nil {
-		t.Fatal("Dialer: not nil?")
-	}
-	c, err := l.Dial(context.Background())
-	if err != nil {
-		t.Fatal("Dial failed:", err)
-	}
-	if _, err := l.Dial(context.Background()); err != ErrTooManyConnections {
-		t.Fatal("Dial:", err)
-	}
-	if err := c.Close(); err != nil {
-		t.Fatal("Close failed:", err)
-	}
-	if err := c.Close(); err != net.ErrClosed {
-		t.Fatal("Close:", err)
-	}
-	if _, err := c.Read(nil); err != net.ErrClosed {
-		t.Fatal("Read:", err)
-	}
-	if _, err := c.Write(nil); err != net.ErrClosed {
-		t.Fatal("Write:", err)
-	}
-	l.Close()
-	if err := l.Close(); err != net.ErrClosed {
-		t.Fatal("l.Close:", err)
-	}
-}