main.go

package main

import (
	"bytes"
	"context"
	"fmt"
	"log"
	"net"
	"time"

	"golang.org/x/net/http2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
)

type simpleService struct {
}

func (s *simpleService) ServerStreamingRpc(req *SimpleRequest, res SimpleService_ServerStreamingRpcServer) error {
	log.Printf("ServerStreamingRpc: %v", req)
	for i := 0; i < 5; i++ {
		res.Send(&SimpleResponse{ResponseMessage: fmt.Sprintf("result %d", i)})
		time.Sleep(1 * time.Second)
	}
	return nil
}

func (s *simpleService) ClientStreamingRpc(SimpleService_ClientStreamingRpcServer) error {
	panic("unimplemented")
}

func (s *simpleService) UnaryRpc(context.Context, *SimpleRequest) (*SimpleResponse, error) {
	panic("unimplemented")
}

func (s *simpleService) BidiStreamingRpc(SimpleService_BidiStreamingRpcServer) error {
	panic("unimplemented")
}

func (s *simpleService) mustEmbedUnimplementedSimpleServiceServer() {
	panic("unimplemented")
}

func main() {
	conn, err := grpc.NewClient(
		"localhost:50051",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		log.Fatalf("fail to dial: %v", err)
	}
	defer conn.Close()

	s, err := conn.NewStream(
		metadata.AppendToOutgoingContext(context.Background(), "zone", "Value: "),
		&grpc.StreamDesc{StreamName: "new", ClientStreams: true, ServerStreams: true},
		"/io.grpc.Tunnel/new",
		grpc.ForceCodec(&RawCodec{}),
	)
	if err != nil {
		log.Fatalf("fail to create stream: %v", err)
	}

	server := grpc.NewServer()
	RegisterSimpleServiceServer(server, &simpleService{})

	if err := server.Serve(ConnListener{conn: ClientChannelConn{stream: s}, done: new(bool)}); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

type RawCodec struct{}

func (c *RawCodec) Marshal(v interface{}) ([]byte, error) {
	b := v.([]byte)
	// hex.Dump(b)
	return b, nil
}

func (c *RawCodec) Unmarshal(data []byte, v interface{}) error {
	result := v.(*bytes.Buffer)

	result.Reset()
	_, err := result.Write(data)
	result.Truncate(len(data))
	// hex.Dump(result.Bytes())
	return err
}

func (c *RawCodec) Name() string {
	return "raw"
}

type ClientChannelConn struct {
	stream grpc.ClientStream
}

var _ net.Conn = ClientChannelConn{}

func (c ClientChannelConn) SetReadDeadline(t time.Time) error {
	return nil
}

func (c ClientChannelConn) SetWriteDeadline(t time.Time) error {
	return nil
}

func (c ClientChannelConn) Read(p []byte) (n int, err error) {
	buf := bytes.NewBuffer(p)
	err = c.stream.RecvMsg(buf)

	f, _ := http2.NewFramer(nil, bytes.NewReader(p)).ReadFrame()
	log.Printf("ClientChannelConn.Read: %v", f)

	return buf.Len(), err
}

func (c ClientChannelConn) Write(p []byte) (n int, err error) {
	f, _ := http2.NewFramer(nil, bytes.NewReader(p)).ReadFrame()
	log.Println("ClientChannelConn.Write:", f)
	err = c.stream.SendMsg(p)
	return len(p), err
}

func (c ClientChannelConn) Close() error {
	return c.stream.CloseSend()
}

func (c ClientChannelConn) LocalAddr() net.Addr {
	return nil
}

func (c ClientChannelConn) RemoteAddr() net.Addr {
	return nil
}

func (c ClientChannelConn) SetDeadline(t time.Time) error {
	return nil
}

type ConnListener struct {
	conn net.Conn
	done *bool
}

func (cl ConnListener) Accept() (net.Conn, error) {
	if *cl.done {
		time.Sleep(24 * time.Hour)
		return nil, nil
	}
	*cl.done = true

	return cl.conn, nil
}

func (cl ConnListener) Close() error {
	return cl.conn.Close()
}

func (cl ConnListener) Addr() net.Addr {
	return &net.TCPAddr{IP: net.IPv4(0, 0, 0, 0), Port: 0}
}

main_server.go

package main

import (
	"bytes"
	"context"
	"io"
	"log"
	"net"
	"time"

	"golang.org/x/net/http2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/encoding"
)

func main() {
	listener, err := net.Listen("tcp", "localhost:50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	encoding.RegisterCodec(&RawCodec{})

	server := grpc.NewServer()

	server.RegisterService(
		&grpc.ServiceDesc{
			ServiceName: "io.grpc.Tunnel",
			Streams: []grpc.StreamDesc{{
				StreamName:    "new",
				ClientStreams: true,
				ServerStreams: true,
				Handler: func(srv interface{}, stream grpc.ServerStream) error {
					client, err := grpc.NewClient(
						"0.0.0.0:0",
						grpc.WithTransportCredentials(insecure.NewCredentials()),
						grpc.WithContextDialer(func(ctx context.Context, s string) (net.Conn, error) {
							return ClientChannelConn{stream: stream}, nil
						}),
					)

					if err != nil {
						panic(err)
					}

					simpleClient := NewSimpleServiceClient(client)

					c, err := simpleClient.ServerStreamingRpc(context.Background(), &SimpleRequest{RequestMessage: "hello"})
					if err != nil {
						panic(err)
					}

					for {
						msg, err := c.Recv()

						if err == io.EOF {
							return nil
						}

						if err != nil {
							panic(err)
						}

						log.Println("ServerStream.Recv:", msg)
					}
				},
			}},
		},
		nil,
	)

	if err := server.Serve(listener); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

type RawCodec struct{}

func (c *RawCodec) Marshal(v interface{}) ([]byte, error) {
	b := v.([]byte)
	// hex.Dump(b)
	return b, nil
}

func (c *RawCodec) Unmarshal(data []byte, v interface{}) error {
	result := v.(*bytes.Buffer)

	result.Reset()
	_, err := result.Write(data)
	result.Truncate(len(data))
	// hex.Dump(result.Bytes())
	return err
}

func (c *RawCodec) Name() string {
	return "raw"
}

type ClientChannelConn struct {
	stream grpc.Stream
}

var _ net.Conn = ClientChannelConn{}

func (c ClientChannelConn) SetReadDeadline(t time.Time) error {
	return nil
}

func (c ClientChannelConn) SetWriteDeadline(t time.Time) error {
	return nil
}

func (c ClientChannelConn) Read(p []byte) (n int, err error) {
	buf := bytes.NewBuffer(p)
	err = c.stream.RecvMsg(buf)

	f, _ := http2.NewFramer(nil, bytes.NewReader(p)).ReadFrame()
	log.Printf("ClientChannelConn.Read: %v", f)

	return buf.Len(), err
}

func (c ClientChannelConn) Write(p []byte) (n int, err error) {
	f, _ := http2.NewFramer(nil, bytes.NewReader(p)).ReadFrame()
	log.Println("ClientChannelConn.Write:", f)
	err = c.stream.SendMsg(p)
	return len(p), err
}

func (c ClientChannelConn) Close() error {
	return nil
}

func (c ClientChannelConn) LocalAddr() net.Addr {
	return nil
}

func (c ClientChannelConn) RemoteAddr() net.Addr {
	return nil
}

func (c ClientChannelConn) SetDeadline(t time.Time) error {
	return nil
}