Stream mode when http2

integration/grpc_test.go

@@ -1,8 +1,10 @@
 package integration
 
 import (
+	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"io/ioutil"
 	"net"
 	"os"
@@ -22,7 +24,9 @@ var LocalhostKey []byte
 // GRPCSuite
 type GRPCSuite struct{ BaseSuite }
 
-type myserver struct{}
+type myserver struct {
+	stopStreamExample chan bool
+}
 
 func (s *GRPCSuite) SetUpSuite(c *check.C) {
 	var err error
@@ -36,7 +40,15 @@ func (s *myserver) SayHello(ctx context.Context, in *helloworld.HelloRequest) (*
 	return &helloworld.HelloReply{Message: "Hello " + in.Name}, nil
 }
 
-func startGRPCServer(lis net.Listener) error {
+func (s *myserver) StreamExample(in *helloworld.StreamExampleRequest, server helloworld.Greeter_StreamExampleServer) error {
+	data := make([]byte, 512)
+	rand.Read(data)
+	server.Send(&helloworld.StreamExampleReply{Data: string(data)})
+	<-s.stopStreamExample
+	return nil
+}
+
+func startGRPCServer(lis net.Listener, server *myserver) error {
 	cert, err := tls.X509KeyPair(LocalhostCert, LocalhostKey)
 	if err != nil {
 		return err
@@ -45,26 +57,30 @@ func startGRPCServer(lis net.Listener) error {
 	creds := credentials.NewServerTLSFromCert(&cert)
 	serverOption := grpc.Creds(creds)
 
-	var s *grpc.Server = grpc.NewServer(serverOption)
+	s := grpc.NewServer(serverOption)
 	defer s.Stop()
 
-	helloworld.RegisterGreeterServer(s, &myserver{})
+	helloworld.RegisterGreeterServer(s, server)
 	return s.Serve(lis)
 }
-
-func callHelloClientGRPC() (string, error) {
+func getHelloClientGRPC() (helloworld.GreeterClient, func() error, error) {
 	roots := x509.NewCertPool()
 	roots.AppendCertsFromPEM(LocalhostCert)
 	credsClient := credentials.NewClientTLSFromCert(roots, "")
 	conn, err := grpc.Dial("127.0.0.1:4443", grpc.WithTransportCredentials(credsClient))
+	if err != nil {
+		return nil, func() error { return nil }, err
+	}
+	return helloworld.NewGreeterClient(conn), conn.Close, nil
+
+}
+
+func callHelloClientGRPC(name string) (string, error) {
+	client, closer, err := getHelloClientGRPC()
+	defer closer()
 	if err != nil {
 		return "", err
 	}
-
-	defer conn.Close()
-	client := helloworld.NewGreeterClient(conn)
-
-	name := "World"
 	r, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: name})
 	if err != nil {
 		return "", err
@@ -72,13 +88,26 @@ func callHelloClientGRPC() (string, error) {
 	return r.Message, nil
 }
 
+func callStreamExampleClientGRPC() (helloworld.Greeter_StreamExampleClient, func() error, error) {
+	client, closer, err := getHelloClientGRPC()
+	if err != nil {
+		return nil, closer, err
+	}
+	t, err := client.StreamExample(context.Background(), &helloworld.StreamExampleRequest{})
+	if err != nil {
+		return nil, closer, err
+	}
+
+	return t, closer, nil
+}
+
 func (s *GRPCSuite) TestGRPC(c *check.C) {
 	lis, err := net.Listen("tcp", ":0")
 	_, port, err := net.SplitHostPort(lis.Addr().String())
 	c.Assert(err, check.IsNil)
 
 	go func() {
-		err := startGRPCServer(lis)
+		err := startGRPCServer(lis, &myserver{})
 		c.Log(err)
 		c.Assert(err, check.IsNil)
 	}()
@@ -106,7 +135,7 @@ func (s *GRPCSuite) TestGRPC(c *check.C) {
 	c.Assert(err, check.IsNil)
 	var response string
 	err = try.Do(1*time.Second, func() error {
-		response, err = callHelloClientGRPC()
+		response, err = callHelloClientGRPC("World")
 		return err
 	})
 
@@ -120,7 +149,7 @@ func (s *GRPCSuite) TestGRPCInsecure(c *check.C) {
 	c.Assert(err, check.IsNil)
 
 	go func() {
-		err := startGRPCServer(lis)
+		err := startGRPCServer(lis, &myserver{})
 		c.Log(err)
 		c.Assert(err, check.IsNil)
 	}()
@@ -148,10 +177,68 @@ func (s *GRPCSuite) TestGRPCInsecure(c *check.C) {
 	c.Assert(err, check.IsNil)
 	var response string
 	err = try.Do(1*time.Second, func() error {
-		response, err = callHelloClientGRPC()
+		response, err = callHelloClientGRPC("World")
 		return err
 	})
 
 	c.Assert(err, check.IsNil)
 	c.Assert(response, check.Equals, "Hello World")
 }
+
+func (s *GRPCSuite) TestGRPCBuffer(c *check.C) {
+	stopStreamExample := make(chan bool)
+	defer func() { stopStreamExample <- true }()
+	lis, err := net.Listen("tcp", ":0")
+	_, port, err := net.SplitHostPort(lis.Addr().String())
+	c.Assert(err, check.IsNil)
+
+	go func() {
+		err := startGRPCServer(lis, &myserver{
+			stopStreamExample: stopStreamExample,
+		})
+		c.Log(err)
+		c.Assert(err, check.IsNil)
+	}()
+
+	file := s.adaptFile(c, "fixtures/grpc/config.toml", struct {
+		CertContent    string
+		KeyContent     string
+		GRPCServerPort string
+	}{
+		CertContent:    string(LocalhostCert),
+		KeyContent:     string(LocalhostKey),
+		GRPCServerPort: port,
+	})
+
+	defer os.Remove(file)
+	cmd, display := s.traefikCmd(withConfigFile(file))
+	defer display(c)
+
+	err = cmd.Start()
+	c.Assert(err, check.IsNil)
+	defer cmd.Process.Kill()
+
+	// wait for Traefik
+	err = try.GetRequest("http://127.0.0.1:8080/api/providers", 1*time.Second, try.BodyContains("Host:127.0.0.1"))
+	c.Assert(err, check.IsNil)
+	var client helloworld.Greeter_StreamExampleClient
+	client, closer, err := callStreamExampleClientGRPC()
+	defer closer()
+
+	received := make(chan bool)
+	go func() {
+		tr, _ := client.Recv()
+		c.Assert(len(tr.Data), check.Equals, 512)
+		received <- true
+	}()
+
+	err = try.Do(time.Second*10, func() error {
+		select {
+		case <-received:
+			return nil
+		default:
+			return errors.New("failed to receive stream data")
+		}
+	})
+	c.Assert(err, check.IsNil)
+}