test: add proxy test

pkg/plugin/controllx/snippet_test.go

@@ -20,7 +20,7 @@ func TestNewSnippetNode(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, n)
 
-	_ = n.Close()
+	assert.NoError(t, n.Close())
 }
 
 func TestSnippetNode_Send(t *testing.T) {

pkg/plugin/controllx/switch_test.go

@@ -18,7 +18,7 @@ func TestNewSwitchNode(t *testing.T) {
 	n := NewSwitchNode()
 	assert.NotNil(t, n)
 
-	_ = n.Close()
+	assert.NoError(t, n.Close())
 }
 
 func TestSwitchNode_Send(t *testing.T) {

pkg/plugin/networkx/proxy_test.go

@@ -0,0 +1,98 @@
+package networkx
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/go-faker/faker/v4"
+	"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/primitive"
+	"github.com/siyul-park/uniflow/pkg/process"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewProxyNode(t *testing.T) {
+	n, err := NewProxyNode(faker.URL())
+	assert.NoError(t, err)
+	assert.NotNil(t, n)
+
+	assert.NoError(t, n.Close())
+}
+
+func TestProxyNode_Send(t *testing.T) {
+	called := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called++
+	}))
+	defer server.Close()
+
+	proxy := NewProxyNode(server.URL)
+	defer proxy.Close()
+
+	io := port.New()
+	ioPort, _ := proxy.Port(node.PortIO)
+	ioPort.Link(io)
+
+	proc := process.New()
+	defer proc.Exit(nil)
+
+	ioStream := io.Open(proc)
+
+	inPayload, _ := primitive.MarshalText(HTTPPayload{
+		Method: http.MethodGet,
+		Path:   "/",
+	})
+	inPck := packet.New(inPayload)
+
+	ioStream.Send(inPck)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	select {
+	case outPck := <-ioStream.Receive():
+		assert.Equal(t, 1, called)
+
+		var outPayload HTTPPayload
+		err := primitive.Unmarshal(outPck.Payload(), &outPayload)
+		assert.NoError(t, err)
+	case <-ctx.Done():
+		assert.Fail(t, "timeout")
+	}
+}
+
+func BenchmarkProxyNode_Send(b *testing.B) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	}))
+	defer server.Close()
+
+	proxy := NewProxyNode(server.URL)
+	defer proxy.Close()
+
+	io := port.New()
+	ioPort, _ := proxy.Port(node.PortIO)
+	ioPort.Link(io)
+
+	proc := process.New()
+	defer proc.Exit(nil)
+
+	ioStream := io.Open(proc)
+
+	inPayload, _ := primitive.MarshalText(HTTPPayload{
+		Method: http.MethodGet,
+		Path:   "/",
+	})
+	inPck := packet.New(inPayload)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		ioStream.Send(inPck)
+		<-ioStream.Receive()
+	}
+}

pkg/plugin/networkx/router_test.go

@@ -18,7 +18,7 @@ func TestNewRouterNode(t *testing.T) {
 	n := NewRouterNode()
 	assert.NotNil(t, n)
 
-	_ = n.Close()
+	assert.NoError(t, n.Close())
 }
 
 func TestRouterNode_Send(t *testing.T) {

pkg/plugin/systemx/reflect_test.go

@@ -31,7 +31,7 @@ func TestNewReflectNode(t *testing.T) {
 	})
 	assert.NotNil(t, n)
 
-	_ = n.Close()
+	assert.NoError(t, n.Close())
 }
 
 func TestReflectNode_Send(t *testing.T) {

pkg/primitive/map.go

@@ -287,10 +287,8 @@ func NewMapDecoder(decoder encoding.Decoder[Value, any]) encoding.Decoder[Value,
 						}
 
 						value, ok := s.Get(NewString(tag.alias))
-						if !ok || reflect.ValueOf(value.Interface()).IsZero() {
-							if tag.omitempty {
-								continue
-							} else {
+						if !ok {
+							if !tag.omitempty {
 								return errors.WithMessage(encoding.ErrUnsupportedValue, fmt.Sprintf("key(%v) is zero value", field.Name))
 							}
 						} else if err := decoder.Decode(value, v.Addr().Interface()); err != nil {