@@ -27,6 +27,7 @@ import (
2727 "crypto/x509"
2828 "flag"
2929 "fmt"
30+ "io"
3031 "io/ioutil"
3132 "log"
3233 "net"
@@ -115,6 +116,59 @@ func (s *server) Plus(ctx context.Context, in *pb.PlusRequest) (*pb.PlusReply, e
115116 return & pb.PlusReply {Result : in .A + in .B }, nil
116117}
117118
119+ // SayHelloServerStream streams HelloReply back to the client.
120+ func (s * server ) SayHelloServerStream (req * pb.HelloRequest , stream pb.Greeter_SayHelloServerStreamServer ) error {
121+ log .Printf ("Received server side stream req: %v\n " , req )
122+
123+ // Say Hello 5 times.
124+ for i := 0 ; i < 5 ; i ++ {
125+ if err := stream .Send (& pb.HelloReply {
126+ Message : fmt .Sprintf ("Hello %s" , req .Name ),
127+ }); err != nil {
128+ return status .Errorf (codes .Unavailable , "Unable to stream request back to client: %v" , err )
129+ }
130+ }
131+ return nil
132+ }
133+
134+ // SayHelloClientStream receives a stream of HelloRequest from a client.
135+ func (s * server ) SayHelloClientStream (stream pb.Greeter_SayHelloClientStreamServer ) error {
136+ log .Println ("SayHello client side streaming has been initiated." )
137+ cache := ""
138+ for {
139+ req , err := stream .Recv ()
140+ if err == io .EOF {
141+ return stream .SendAndClose (& pb.HelloReply {Message : cache })
142+ }
143+ if err != nil {
144+ return status .Errorf (codes .Unavailable , "Failed to read client stream: %v" , err )
145+ }
146+ cache = fmt .Sprintf ("%sHello %s!" , cache , req .Name )
147+ }
148+ }
149+
150+ // SayHelloBidirectionalStream establishes a bidirectional stream with the client.
151+ func (s * server ) SayHelloBidirectionalStream (stream pb.Greeter_SayHelloBidirectionalStreamServer ) error {
152+ log .Println ("SayHello bidirectional streaming has been initiated." )
153+
154+ for {
155+ req , err := stream .Recv ()
156+ if err == io .EOF {
157+ return stream .Send (& pb.HelloReply {Message : "stream ended" })
158+ }
159+ if err != nil {
160+ return status .Errorf (codes .Unavailable , "Failed to read client stream: %v" , err )
161+ }
162+
163+ // A small 0.5 sec sleep
164+ time .Sleep (500 * time .Millisecond )
165+
166+ if err := stream .Send (& pb.HelloReply {Message : fmt .Sprintf ("Hello %s" , req .Name )}); err != nil {
167+ return status .Errorf (codes .Unknown , "Failed to stream response back to client: %v" , err )
168+ }
169+ }
170+ }
171+
118172func main () {
119173 flag .Parse ()
120174
@@ -182,7 +236,7 @@ func main() {
182236 }()
183237 }
184238
185- signals := make (chan os.Signal , 1 )
239+ signals := make (chan os.Signal )
186240 signal .Notify (signals , os .Interrupt , syscall .SIGTERM )
187241 sig := <- signals
188242 log .Printf ("get signal %s, exit\n " , sig .String ())
0 commit comments