feat: add tcp proxy and stream example (#2)

Author: joway

client.go echo/client.go

@@ -26,7 +26,7 @@ import (
 	"github.com/cloudwego/netpoll"
 	"github.com/cloudwego/netpoll/mux"
 
-	"github.com/cloudwego/netpoll-examples/codec"
+	"github.com/cloudwego/netpoll-examples/echo/codec"
 )
 
 func main() {

codec/msg.go echo/codec/msg.go

@@ -25,7 +25,7 @@ import (
 	"github.com/cloudwego/netpoll"
 	"github.com/cloudwego/netpoll/mux"
 
-	"github.com/cloudwego/netpoll-examples/codec"
+	"github.com/cloudwego/netpoll-examples/echo/codec"
 )
 
 func main() {

codec/rw.go echo/codec/rw.go

@@ -45,9 +45,12 @@ func prepare(connection netpoll.Connection) context.Context {
 }
 
 func handle(ctx context.Context, connection netpoll.Connection) error {
-	reader := connection.Reader()
+	reader, writer := connection.Reader(), connection.Writer()
 	defer reader.Release()
+
 	msg, _ := reader.ReadString(reader.Len())
-	println(msg)
+	_, _ = writer.WriteString(msg)
+	writer.Flush()
+
 	return nil
 }

mux/mux_client.go echo/mux/mux_client.go

@@ -3,3 +3,5 @@ module github.com/cloudwego/netpoll-examples
 go 1.15
 
 require github.com/cloudwego/netpoll v0.1.2
+
+replace github.com/cloudwego/netpoll => github.com/joway/netpoll v0.0.4-0.20220112083038-2eefc181239d

mux/mux_server.go echo/mux/mux_server.go

@@ -1,9 +1,9 @@
 github.com/bytedance/gopkg v0.0.0-20210705062217-74c74ebadcae h1:ERLYTdHnm2E8jwpprhHPvBhbPBaxnwl62tb3lR8Nd+k=
 github.com/bytedance/gopkg v0.0.0-20210705062217-74c74ebadcae/go.mod h1:birsdqRCbwnckJbdAvcSao+AzOyibVEoWB55MjpYpB8=
-github.com/cloudwego/netpoll v0.1.2 h1:NSvqHfCmmR3g0ASshwAB0F2RJvXK8v7ToFmhWzh/ukY=
-github.com/cloudwego/netpoll v0.1.2/go.mod h1:rZOiNI0FYjuvNybXKKhAPUja03loJi/cdv2F55AE6E8=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/joway/netpoll v0.0.4-0.20220112083038-2eefc181239d h1:1mx//ApM0tAisXfemrMgFGIlNQ2D9BkZtAIK058L4Iw=
+github.com/joway/netpoll v0.0.4-0.20220112083038-2eefc181239d/go.mod h1:rZOiNI0FYjuvNybXKKhAPUja03loJi/cdv2F55AE6E8=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

server.go echo/server.go

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2021 CloudWeGo
+ *
+ * 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 main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cloudwego/netpoll"
+)
+
+func main() {
+	network, address := "tcp", ":8082"
+	listener, _ := netpoll.CreateListener(network, address)
+	eventLoop, _ := netpoll.NewEventLoop(
+		nil,
+		netpoll.WithOnConnect(onConnect),
+	)
+
+	// start listen loop ...
+	eventLoop.Serve(listener)
+}
+
+var _ netpoll.OnConnect = onConnect
+
+func onConnect(ctx context.Context, connection netpoll.Connection) context.Context {
+	go func() {
+		for range time.Tick(time.Second) {
+			connection.Writer().WriteString(fmt.Sprintf("%s\n", time.Now().Format(time.RFC3339)))
+			connection.Writer().Flush()
+		}
+	}()
+	return context.Background()
+}

go.mod

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2021 CloudWeGo
+ *
+ * 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 main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/cloudwego/netpoll"
+)
+
+var (
+	downstreamAddr = "127.0.0.1:8080"
+	downstreamKey  = "downstream"
+)
+
+func main() {
+	network, address := "tcp", ":8081"
+	listener, _ := netpoll.CreateListener(network, address)
+	eventLoop, _ := netpoll.NewEventLoop(
+		onRequest,
+		netpoll.WithOnConnect(onConnect),
+		netpoll.WithReadTimeout(time.Second),
+	)
+
+	// start listen loop ...
+	eventLoop.Serve(listener)
+}
+
+var _ netpoll.OnConnect = onConnect
+var _ netpoll.OnRequest = onRequest
+
+func onConnect(ctx context.Context, upstream netpoll.Connection) context.Context {
+	downstream, err := netpoll.DialConnection("tcp", downstreamAddr, time.Second)
+	if err != nil {
+		log.Printf("connect downstream failed: %v", err)
+	}
+	return context.WithValue(ctx, downstreamKey, downstream)
+}
+
+func onRequest(ctx context.Context, upstream netpoll.Connection) error {
+	// read request
+	req, _ := upstream.Reader().ReadString(upstream.Reader().Len())
+
+	// send request to downstream
+	downstream := ctx.Value(downstreamKey).(netpoll.Connection)
+	_, _ = downstream.Writer().WriteString(req)
+	downstream.Writer().Flush()
+
+	// receive response from downstream
+	resp, _ := downstream.Reader().ReadString(len(req))
+
+	// send response to upstream
+	upstream.Writer().WriteString(resp)
+	upstream.Writer().Flush()
+	return nil
+}