refactor: expose port pipe

ext/pkg/network/http_test.go

@@ -3,14 +3,15 @@ package network
 import (
 	"context"
 	"crypto/tls"
-	"github.com/go-faker/faker/v4"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"testing"
 	"time"
 
+	"github.com/go-faker/faker/v4"
+
 	"golang.org/x/net/http2"
 
 	"github.com/siyul-park/uniflow/pkg/node"

pkg/port/pipe.go

@@ -0,0 +1,39 @@
+package port
+
+import (
+	"github.com/siyul-park/uniflow/pkg/packet"
+	"github.com/siyul-park/uniflow/pkg/process"
+)
+
+// Pipe is a connection between two ports.
+func Pipe() (*InPort, *OutPort) {
+	inPort, outPort := NewIn(), NewOut()
+
+	inPort.AddListener(ListenFunc(func(proc *process.Process) {
+		reader := inPort.Open(proc)
+		var writer *packet.Writer
+
+		for inPck := range reader.Read() {
+			if writer == nil {
+				writer = outPort.Open(proc)
+			}
+			if writer.Write(inPck) == 0 {
+				reader.Receive(inPck)
+			}
+		}
+	}))
+
+	outPort.AddListener(ListenFunc(func(proc *process.Process) {
+		var reader *packet.Reader
+		writer := outPort.Open(proc)
+
+		for backPck := range writer.Receive() {
+			if reader == nil {
+				reader = inPort.Open(proc)
+			}
+			reader.Receive(backPck)
+		}
+	}))
+
+	return inPort, outPort
+}

pkg/port/pipe_test.go

@@ -0,0 +1,45 @@
+package port
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/go-faker/faker/v4"
+	"github.com/siyul-park/uniflow/pkg/packet"
+	"github.com/siyul-park/uniflow/pkg/process"
+	"github.com/siyul-park/uniflow/pkg/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPipe(t *testing.T) {
+	inPort0, outPort0 := Pipe()
+	defer inPort0.Close()
+	defer outPort0.Close()
+
+	inPort1, outPort1 := NewIn(), NewOut()
+	defer inPort1.Close()
+	defer outPort1.Close()
+
+	outPort0.Link(inPort1)
+	outPort1.Link(inPort0)
+
+	proc := process.New()
+	defer proc.Exit(nil)
+
+	inWriter := outPort1.Open(proc)
+	outReader := inPort1.Open(proc)
+
+	ctx, cancel := context.WithTimeout(context.TODO(), time.Second)
+	defer cancel()
+
+	inPck := packet.New(types.NewString(faker.UUIDHyphenated()))
+	inWriter.Write(inPck)
+
+	select {
+	case outPck := <-outReader.Read():
+		assert.Equal(t, inPck.Payload(), outPck.Payload())
+	case <-ctx.Done():
+		assert.NoError(t, ctx.Err())
+	}
+}

pkg/symbol/cluster.go

@@ -4,9 +4,7 @@ import (
 	"sync"
 
 	"github.com/siyul-park/uniflow/pkg/node"
-	"github.com/siyul-park/uniflow/pkg/packet"
 	"github.com/siyul-park/uniflow/pkg/port"
-	"github.com/siyul-park/uniflow/pkg/process"
 	"github.com/siyul-park/uniflow/pkg/spec"
 )
 
@@ -58,25 +56,15 @@ func (n *Cluster) Inbound(source string, target spec.Port) bool {
 		return false
 	}
 
-	inPort, ok1 := n.inPorts[source]
-	if !ok1 {
-		inPort = port.NewIn()
+	_, ok1 := n._inPorts[source]
+	outPort, ok2 := n.outPorts[source]
+	if !ok1 || !ok2 {
+		var inPort *port.InPort
+		inPort, outPort = port.Pipe()
 		n.inPorts[source] = inPort
-	}
-
-	outPort, ok2 := n._outPorts[source]
-	if !ok2 {
-		outPort = port.NewOut()
 		n._outPorts[source] = outPort
 	}
 
-	if !ok1 {
-		inPort.AddListener(n.inbound(inPort, outPort))
-	}
-	if !ok2 {
-		outPort.AddListener(n.outbound(inPort, outPort))
-	}
-
 	outPort.Link(prt)
 	return true
 }
@@ -103,24 +91,14 @@ func (n *Cluster) Outbound(source string, target spec.Port) bool {
 	}
 
 	inPort, ok1 := n._inPorts[source]
-	if !ok1 {
-		inPort = port.NewIn()
+	_, ok2 := n.outPorts[source]
+	if !ok1 || !ok2 {
+		var outPort *port.OutPort
+		inPort, outPort = port.Pipe()
 		n._inPorts[source] = inPort
-	}
-
-	outPort, ok2 := n.outPorts[source]
-	if !ok2 {
-		outPort = port.NewOut()
 		n.outPorts[source] = outPort
 	}
 
-	if !ok1 {
-		inPort.AddListener(n.inbound(inPort, outPort))
-	}
-	if !ok2 {
-		outPort.AddListener(n.outbound(inPort, outPort))
-	}
-
 	prt.Link(inPort)
 	return true
 }
@@ -205,33 +183,3 @@ func (n *Cluster) Close() error {
 	}
 	return nil
 }
-
-func (n *Cluster) inbound(inPort *port.InPort, outPort *port.OutPort) port.Listener {
-	return port.ListenFunc(func(proc *process.Process) {
-		reader := inPort.Open(proc)
-		var writer *packet.Writer
-
-		for inPck := range reader.Read() {
-			if writer == nil {
-				writer = outPort.Open(proc)
-			}
-			if writer.Write(inPck) == 0 {
-				reader.Receive(inPck)
-			}
-		}
-	})
-}
-
-func (n *Cluster) outbound(inPort *port.InPort, outPort *port.OutPort) port.Listener {
-	return port.ListenFunc(func(proc *process.Process) {
-		var reader *packet.Reader
-		writer := outPort.Open(proc)
-
-		for backPck := range writer.Receive() {
-			if reader == nil {
-				reader = inPort.Open(proc)
-			}
-			reader.Receive(backPck)
-		}
-	})
-}