gRpc の client-side streaming RPC を試す

syntax = "proto3";

package upload;

option go_package = "github.com/k1350/proto/upload";

service UploadFile {  
  // ファイルを分割アップロードし、結果を返す
  rpc Upload (stream File) returns (Result) {};
}

message File { bytes Data = 1; }

// 脚注：本当は Code というように大文字から始めるべきだったのだが間違って小文字にしてしまった。
// しかしコンパイル結果は勝手に大文字になっていた。
message Result { int32 code = 1; }

// UploadFileServer is the server API for UploadFile service.
// All implementations must embed UnimplementedUploadFileServer
// for forward compatibility
type UploadFileServer interface {
	// ファイルを分割アップロードし、結果を返す
	Upload(UploadFile_UploadServer) error
	mustEmbedUnimplementedUploadFileServer()
}

package upload

import (
    //（中略）
	pb "github.com/k1350/server/api/upload"
)

type Server struct {
    // ここで UnimplementedUploadFileServer を Embed する
	pb.UnimplementedUploadFileServer
}

func (s *Server) Upload(stream pb.UploadFile_UploadServer) error {
	// （中略）
}

func (s *Server) Upload(stream pb.UploadFile_UploadServer) error {
    // tmp ディレクトリ内に tmp.jpg というファイルを作成
	file, err := os.Create(filepath.Join("tmp", "tmp.jpg"))
	defer file.Close()
	if err != nil {
		return err
	}

	for {
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
        // tmp.jpg に書き込み
		file.Write(res.Data)
	}
	err = stream.SendAndClose(&pb.Result{Code: 200})
	if err != nil {
		return err
	}
	return nil
}

type UploadFile_UploadServer interface {
	SendAndClose(*Result) error
	Recv() (*File, error)
	grpc.ServerStream
}

type uploadFileUploadServer struct {
	grpc.ServerStream
}

func (x *uploadFileUploadServer) SendAndClose(m *Result) error {
	return x.ServerStream.SendMsg(m)
}

func (x *uploadFileUploadServer) Recv() (*File, error) {
	m := new(File)
	if err := x.ServerStream.RecvMsg(m); err != nil {
		return nil, err
	}
	return m, nil
}

package main

import (
	"net"
	"log"

	pb "github.com/k1350/server/api/upload"
	up "github.com/k1350/server/internal/upload"
	"google.golang.org/grpc"
)

const (
	port = ":50051"
)

func main() {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterUploadFileServer(s, &up.Server{})
	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

package upload

import (
	"context"
	"log"
	"os"
	"time"
	"io"

	pb "github.com/k1350/client/api/upload"
)

const (
	// 読み取りバッファーサイズ
	buffersize = 256
)

func RunUpload(client pb.UploadFileClient, filepath string) {
	file, err := os.Open(filepath)
	defer file.Close()
	if err != nil {
		log.Fatalf("%v", err);
	}
	buf := make([]byte, buffersize)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	stream, err := client.Upload(ctx)
	if err != nil {
		log.Fatalf("%v.Upload(_) = _, %v", client, err)
	}

	for {
		_, err := file.Read(buf)
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("%v.Send(%v) = %v", stream, buf, err)
		}
		stream.Send(&pb.File{Data: buf})
	}

	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
	}
	log.Printf("Result: %v", res)
}

func (c *uploadFileClient) Upload(ctx context.Context, opts ...grpc.CallOption) (UploadFile_UploadClient, error) {
	stream, err := c.cc.NewStream(ctx, &UploadFile_ServiceDesc.Streams[0], "/upload.UploadFile/Upload", opts...)
	if err != nil {
		return nil, err
	}
	x := &uploadFileUploadClient{stream}
	return x, nil
}

type UploadFile_UploadClient interface {
	Send(*File) error
	CloseAndRecv() (*Result, error)
	grpc.ClientStream
}

type uploadFileUploadClient struct {
	grpc.ClientStream
}

func (x *uploadFileUploadClient) Send(m *File) error {
	return x.ClientStream.SendMsg(m)
}

func (x *uploadFileUploadClient) CloseAndRecv() (*Result, error) {
	if err := x.ClientStream.CloseSend(); err != nil {
		return nil, err
	}
	m := new(Result)
	if err := x.ClientStream.RecvMsg(m); err != nil {
		return nil, err
	}
	return m, nil
}

package main

import (
	"log"

	pb "github.com/k1350/client/api/upload"
	up "github.com/k1350/client/internal/upload"
	"google.golang.org/grpc"
)

const (
	address = "server:50051"
)

func main() {
	var opts []grpc.DialOption
	opts = append(opts, grpc.WithInsecure())
	opts = append(opts, grpc.WithBlock())
	conn, err := grpc.Dial(address, opts...)
	if err != nil {
		log.Fatalf("fail to dial: %v", err)
	}
	defer conn.Close()
	client := pb.NewUploadFileClient(conn)

    // sample.jpg をアップロードする
	up.RunUpload(client, "./sample.jpg")
}