stream sample

add_count.py

import proto.sample_pb2 as pb
import sys
from grpc.beta import implementations

_TIMEOUT_SECONDS = 3

def run(addr, number):
    channel = implementations.insecure_channel(addr, 8080)
    with pb.beta_create_CountService_stub(channel) as stub:

        arg = pb.Count(number=int(number))
        ret = stub.AddCount(arg, _TIMEOUT_SECONDS)

        if ret.code == 0:
            print "Success!"
        else:
            print "Error!"


if __name__ == '__main__':
    addr = sys.argv[1]
    number = sys.argv[2]
    run(addr, number)

make_pb.sh

#!/bin/bash

go get -u github.com/golang/protobuf/protoc-gen-go
mkdir proto
touch ./proto/__init__.py
protoc --go_out=plugins=grpc:./proto sample.proto
protoc --python_out=./proto --grpc_out=./proto --plugin=protoc-gen-grpc=`which grpc_python_plugin` sample.proto

go get google.golang.org/grpc
go get golang.org/x/net/context

monitor_count.py

import proto.sample_pb2 as pb
import sys
from grpc.beta import implementations

_TIMEOUT_SECONDS = 100000

def run(addr):
    channel = implementations.insecure_channel(addr, 8080)
    with pb.beta_create_CountService_stub(channel) as stub:

        arg = pb.Request()
        count = stub.MonitorCount(arg, _TIMEOUT_SECONDS)

        for c in count:
            print(c.number)

if __name__ == '__main__':
    addr = sys.argv[1]
    run(addr)

sample.proto

syntax = "proto3";

package proto;

service CountService {
  rpc MonitorCount(Request) returns (stream Count) {};
  rpc AddCount(Count) returns (Error) {};
}

message Request {
}

message Error {
  int32 code = 1;
}

message Count {
  int32 number = 1;
}

sample_server.go

package main

import (
	"fmt"
	"golang.org/x/net/context"
	"log"
	"net"
	"sync"

	"google.golang.org/grpc"

	pb "./proto"
)

type countService struct {
	count  int32
	mu     sync.Mutex
	chList []chan int32
}

func NewCountService() *countService {
	s := &countService{}
	s.chList = []chan int32{}
	return s
}

func handleResponse(responseCh chan int32, stream pb.CountService_MonitorCountServer) error {

	for number := range responseCh {
		count := &pb.Count{Number: number}
		if err := stream.Send(count); err != nil {
			return err
		}
	}
	return nil
}

func (s *countService) MonitorCount(p *pb.Request, stream pb.CountService_MonitorCountServer) error {

	ch := make(chan int32, 8)
	s.mu.Lock()
	s.chList = append(s.chList, ch)
	s.mu.Unlock()

	return handleResponse(ch, stream)
}

func (s *countService) AddCount(ctx context.Context, c *pb.Count) (*pb.Error, error) {

	s.mu.Lock()
	s.count += c.Number
	fmt.Println("current = ", s.count)
	defer s.mu.Unlock()
	for _, ch := range s.chList {
		ch <- s.count
	}

	return &pb.Error{Code: 0}, nil
}

func main() {
	listen, err := net.Listen("tcp", ":8080")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	server := grpc.NewServer()
	pb.RegisterCountServiceServer(server, NewCountService())
	server.Serve(listen)
}