X pipenet draftly works

go/xcommon/pipenet/pipenet.go

@@ -19,8 +19,8 @@
 //
 // TODO describe addressing scheme
 //
-// it is useful for testing networked applications interaction in 1 process
-// without going to OS networking stack. XXX
+// it might be handy for testing interaction in networked applications in 1
+// process without going to OS networking stack.
 package pipenet
 
 import (
@@ -52,7 +52,7 @@ type Addr struct {
 // Network implements synchronous in-memory network of pipes
 // It can be worked with the same way a regular TCP network is handled with Dial/Listen/Accept/...
 //
-// Network must be created with New
+// Network must be created with New	XXX is it really ok to have global state ?
 type Network struct {
 	// name of this network under "pipe" namespace -> e.g. ""
 	// full network name will be reported as "pipe"+Name		XXX -> just full name ?
@@ -226,8 +226,8 @@ func (l *listener) Accept() (net.Conn, error) {
 
 		n.mu.Unlock()
 
-		resp <- pc
-		return ps, nil
+		resp <- e.pipev[0]
+		return e.pipev[1], nil
 	}
 }
 
@@ -346,14 +346,14 @@ func New(name string) *Network {
 // lookupNet lookups Network by name
 // name is full network name, e.g. "pipe"
 func lookupNet(name string) (*Network, error) {
-	if !strings.HasPrefix(NetPrefix, name) {
+	if !strings.HasPrefix(name, NetPrefix) {
 		return nil, errBadNetwork
 	}
 
 	netMu.Lock()
 	defer netMu.Unlock()
 
-	n := networks[strings.TrimPrefix(NetPrefix, name)]
+	n := networks[strings.TrimPrefix(name, NetPrefix)]
 	if n == nil {
 		return nil, errNetNotFound
 	}

go/xcommon/pipenet/pipenet_test.go

@@ -17,4 +17,113 @@
 
 package pipenet
 
-// TODO
+import (
+	"fmt"
+	"io"
+	"net"
+	"reflect"
+	"testing"
+
+	"golang.org/x/sync/errgroup"
+
+	"lab.nexedi.com/kirr/go123/exc"
+)
+
+// we assume net.Pipe works ok; here we only test Listen/Accept/Dial routing
+// XXX tests are ugly, non-robust and small coverage
+
+func xlisten(network, laddr string) net.Listener {
+	l, err := Listen(network, laddr)
+	exc.Raiseif(err)
+	return l
+}
+
+func xaccept(l net.Listener) net.Conn {
+	c, err := l.Accept()
+	exc.Raiseif(err)
+	return c
+}
+
+func xdial(network, addr string) net.Conn {
+	c, err := Dial(network, addr)
+	exc.Raiseif(err)
+	return c
+}
+
+func xread(r io.Reader) string {
+	buf := make([]byte, 4096)
+	n, err := r.Read(buf)
+	exc.Raiseif(err)
+	return string(buf[:n])
+}
+
+func xwrite(w io.Writer, data string) {
+	_, err := w.Write([]byte(data))
+	exc.Raiseif(err)
+}
+
+func xwait(w interface { Wait() error }) {
+	err := w.Wait()
+	exc.Raiseif(err)
+}
+
+func assertEq(t *testing.T, a, b interface{}) {
+	if !reflect.DeepEqual(a, b) {
+		fmt.Printf("not equal:\nhave: %v\nwant: %v\n", a, b)
+		t.Errorf("not equal:\nhave: %v\nwant: %v", a, b)
+		exc.Raise(0)
+	}
+}
+
+
+func TestPipeNet(t *testing.T) {
+	New("α")
+
+	_, err := Dial("α", "0")
+	assertEq(t, err, &net.OpError{Op: "dial", Net: "α", Addr: &Addr{"α", "0"}, Err: errBadNetwork})
+
+	_, err = Dial("pipeα", "0")
+	assertEq(t, err, &net.OpError{Op: "dial", Net: "pipeα", Addr: &Addr{"pipeα", "0"}, Err: errConnRefused})
+
+	l1 := xlisten("pipeα", "")
+	assertEq(t, l1.Addr(), &Addr{"pipeα", "0"})
+
+	// XXX -> use workGroup (in connection_test.go)
+	wg := &errgroup.Group{}
+	wg.Go(func() error {
+		return exc.Runx(func() {
+			c1s := xaccept(l1)
+			assertEq(t, c1s.LocalAddr(), &Addr{"pipeα", "1s"})
+			assertEq(t, c1s.RemoteAddr(), &Addr{"pipeα", "1c"})
+
+			assertEq(t, xread(c1s), "ping")
+			xwrite(c1s, "pong")
+
+			c2s := xaccept(l1)
+			assertEq(t, c2s.LocalAddr(), &Addr{"pipeα", "2s"})
+			assertEq(t, c2s.RemoteAddr(), &Addr{"pipeα", "2c"})
+
+			assertEq(t, xread(c2s), "hello")
+			xwrite(c2s, "world")
+		})
+	})
+
+	c1c := xdial("pipeα", "0")
+	assertEq(t, c1c.LocalAddr(), &Addr{"pipeα", "1c"})
+	assertEq(t, c1c.RemoteAddr(), &Addr{"pipeα", "1s"})
+
+	xwrite(c1c, "ping")
+	assertEq(t, xread(c1c), "pong")
+
+	c2c := xdial("pipeα", "0")
+	assertEq(t, c2c.LocalAddr(), &Addr{"pipeα", "2c"})
+	assertEq(t, c2c.RemoteAddr(), &Addr{"pipeα", "2s"})
+
+	xwrite(c2c, "hello")
+	assertEq(t, xread(c2c), "world")
+
+	xwait(wg)
+
+	l2 := xlisten("pipeα", "")
+	assertEq(t, l2.Addr(), &Addr{"pipeα", "3"})
+}