diff --git a/materials/02-communication/02-communication.pdf b/materials/02-communication/02-communication.pdf new file mode 100644 index 0000000..1d1cef1 Binary files /dev/null and b/materials/02-communication/02-communication.pdf differ diff --git a/materials/02-communication/grpc-practice/docker-compose.yaml b/materials/02-communication/grpc-practice/docker-compose.yaml new file mode 100644 index 0000000..7ddc3f5 --- /dev/null +++ b/materials/02-communication/grpc-practice/docker-compose.yaml @@ -0,0 +1,13 @@ +services: + server: + build: ./go-server + environment: + - SERVER_ADDR=0.0.0.0:51000 + ports: + - "51000:51000" + + client1: + image: hse-grpc-client + environment: + - SERVER_ADDR=server:51000 + - VALUE_TO_PUT=100500 \ No newline at end of file diff --git a/materials/02-communication/grpc-practice/go-server/Dockerfile b/materials/02-communication/grpc-practice/go-server/Dockerfile new file mode 100644 index 0000000..167b326 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/Dockerfile @@ -0,0 +1,34 @@ +# Dockerfile was generated from +# https://github.com/lodthe/dockerfiles/blob/main/go/Dockerfile + +FROM golang:1.21-alpine3.19 AS builder + +# Setup base software for building an app. +RUN apk update && apk add ca-certificates git gcc g++ libc-dev binutils + +WORKDIR /opt + +# Download dependencies. +COPY go.mod go.sum ./ +RUN go mod download && go mod verify + +# Copy application source. +COPY . . + +# Build the application. +RUN go build -o bin/application . + +# Prepare executor image. +FROM alpine:3.19 AS runner + +RUN apk update && apk add ca-certificates libc6-compat openssh bash && rm -rf /var/cache/apk/* + +WORKDIR /opt + +COPY --from=builder /opt/bin/application ./ + +# Add required static files. +#COPY assets assets + +# Run the application. +CMD ["./application"] \ No newline at end of file diff --git a/materials/02-communication/grpc-practice/go-server/README.md b/materials/02-communication/grpc-practice/go-server/README.md new file mode 100644 index 0000000..66bc293 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/README.md @@ -0,0 +1,45 @@ +# gRPC сервер + +Пример gRPC сервера, реализующего 2 rpc: положить и получить значение. + +В env переменной `SERVER_ADDR` передается адрес и порт, которые будет "слушать" сервер. +Если переменная не задана, используется параметр по умолчанию — `0.0.0.0:51000`. + +## protobuf + +protobuf спецификация хранится в [storage.proto](./proto/storage.proto). +Для перегенерации *pb.go-файлов, необходимо в корне проекта вызвать protoc: +```bash +protoc --go_out=. --go_opt=paths=source_relative \ + --go-grpc_out=. --go-grpc_opt=paths=source_relative \ + proto/storage.proto +``` + +Подробнее о gRPC можно узнать в [документации](https://grpc.io/docs/languages/go/basics/) (там же есть блок про stream'ы). С более сложным примером proto-файла можно ознакомиться [здесь](https://github.com/paralin/raft-grpc/blob/master/raft-grpc.proto). + +## Тестирование + +Сервер можно запустить с помощью команды `go run main.go`. + +Подключиться к серверу можно с помощью [grpcurl](https://github.com/fullstorydev/grpcurl): +```bash +# Установить значение 100500 +grpcurl -d '{"value": {"payload": 100500}}' -plaintext localhost:51000 storage.Storage/PutValue + +# Получить актуальное значение +grpcurl -d '{}' -plaintext localhost:51000 storage.Storage/GetValue +``` +## Docker + +```bash +# Собрать Docker образ с сервером +docker build -t hse-grpc-server . + +# Запустить Docker контейнер: +# -d - запуск в фоне +# -p 50000:51000 - проброс портов: по порту 50000 с хоста будет доступен порт 51000 из контейнера +docker run -d -p 50000:51000 hse-grpc-server + +docker ps +docker stop CONTAINER_ID +``` \ No newline at end of file diff --git a/materials/02-communication/grpc-practice/go-server/go.mod b/materials/02-communication/grpc-practice/go-server/go.mod new file mode 100644 index 0000000..39d55c0 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/go.mod @@ -0,0 +1,16 @@ +module hsegrpc + +go 1.20 + +require ( + google.golang.org/grpc v1.58.1 + google.golang.org/protobuf v1.31.0 +) + +require ( + github.com/golang/protobuf v1.5.3 // indirect + golang.org/x/net v0.12.0 // indirect + golang.org/x/sys v0.10.0 // indirect + golang.org/x/text v0.11.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect +) diff --git a/materials/02-communication/grpc-practice/go-server/go.sum b/materials/02-communication/grpc-practice/go-server/go.sum new file mode 100644 index 0000000..411e455 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/go.sum @@ -0,0 +1,20 @@ +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= +golang.org/x/net v0.12.0 h1:cfawfvKITfUsFCeJIHJrbSxpeu/E81khclypR0GVT50= +golang.org/x/net v0.12.0/go.mod h1:zEVYFnQC7m/vmpQFELhcD1EWkZlX69l4oqgmer6hfKA= +golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA= +golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.11.0 h1:LAntKIrcmeSKERyiOh0XMV39LXS8IE9UL2yP7+f5ij4= +golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 h1:bVf09lpb+OJbByTj913DRJioFFAjf/ZGxEz7MajTp2U= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98/go.mod h1:TUfxEVdsvPg18p6AslUXFoLdpED4oBnGwyqk3dV1XzM= +google.golang.org/grpc v1.58.1 h1:OL+Vz23DTtrrldqHK49FUOPHyY75rvFqJfXC84NYW58= +google.golang.org/grpc v1.58.1/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= +google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= diff --git a/materials/02-communication/grpc-practice/go-server/main.go b/materials/02-communication/grpc-practice/go-server/main.go new file mode 100644 index 0000000..80a632f --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/main.go @@ -0,0 +1,44 @@ +//go:generate protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative proto/storage.proto +package main + +import ( + "fmt" + storagepb "hsegrpc/proto" + "hsegrpc/storage" + "log" + "net" + "os" + + "google.golang.org/grpc" + "google.golang.org/grpc/reflection" +) + +func main() { + // Reading server address from env. + addr := os.Getenv("SERVER_ADDR") + if addr == "" { + addr = "0.0.0.0:51000" + + fmt.Println("Missing SERVER_ADDR, using default value: " + addr) + } + + // Creating a TCP socket. + lis, err := net.Listen("tcp", addr) + if err != nil { + log.Fatalf("failed to listen: %v", err) + } + + grpcServer := grpc.NewServer() + // If you want to connect to the server via grpcurl, you have to register the reflection service. + reflection.Register(grpcServer) + + // Creating and registering implementation of the storage service. + storageService := storage.NewServer() + storagepb.RegisterStorageServer(grpcServer, storageService) + + // Starting the server. + err = grpcServer.Serve(lis) + if err != nil { + log.Fatalf("server failed") + } +} diff --git a/materials/02-communication/grpc-practice/go-server/proto/storage.pb.go b/materials/02-communication/grpc-practice/go-server/proto/storage.pb.go new file mode 100644 index 0000000..b39354d --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/proto/storage.pb.go @@ -0,0 +1,416 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.26.0 +// protoc v4.25.2 +// source: proto/storage.proto + +package storagepb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type Value struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Payload uint64 `protobuf:"varint,1,opt,name=payload,proto3" json:"payload,omitempty"` + UpdatedAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=updated_at,json=updatedAt,proto3,oneof" json:"updated_at,omitempty"` +} + +func (x *Value) Reset() { + *x = Value{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_storage_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Value) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Value) ProtoMessage() {} + +func (x *Value) ProtoReflect() protoreflect.Message { + mi := &file_proto_storage_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Value.ProtoReflect.Descriptor instead. +func (*Value) Descriptor() ([]byte, []int) { + return file_proto_storage_proto_rawDescGZIP(), []int{0} +} + +func (x *Value) GetPayload() uint64 { + if x != nil { + return x.Payload + } + return 0 +} + +func (x *Value) GetUpdatedAt() *timestamppb.Timestamp { + if x != nil { + return x.UpdatedAt + } + return nil +} + +type PutRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Value *Value `protobuf:"bytes,1,opt,name=value,proto3" json:"value,omitempty"` +} + +func (x *PutRequest) Reset() { + *x = PutRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_storage_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *PutRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PutRequest) ProtoMessage() {} + +func (x *PutRequest) ProtoReflect() protoreflect.Message { + mi := &file_proto_storage_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PutRequest.ProtoReflect.Descriptor instead. +func (*PutRequest) Descriptor() ([]byte, []int) { + return file_proto_storage_proto_rawDescGZIP(), []int{1} +} + +func (x *PutRequest) GetValue() *Value { + if x != nil { + return x.Value + } + return nil +} + +type PutResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Value uint64 `protobuf:"varint,1,opt,name=value,proto3" json:"value,omitempty"` +} + +func (x *PutResponse) Reset() { + *x = PutResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_storage_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *PutResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PutResponse) ProtoMessage() {} + +func (x *PutResponse) ProtoReflect() protoreflect.Message { + mi := &file_proto_storage_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PutResponse.ProtoReflect.Descriptor instead. +func (*PutResponse) Descriptor() ([]byte, []int) { + return file_proto_storage_proto_rawDescGZIP(), []int{2} +} + +func (x *PutResponse) GetValue() uint64 { + if x != nil { + return x.Value + } + return 0 +} + +type GetRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *GetRequest) Reset() { + *x = GetRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_storage_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *GetRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRequest) ProtoMessage() {} + +func (x *GetRequest) ProtoReflect() protoreflect.Message { + mi := &file_proto_storage_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRequest.ProtoReflect.Descriptor instead. +func (*GetRequest) Descriptor() ([]byte, []int) { + return file_proto_storage_proto_rawDescGZIP(), []int{3} +} + +type GetResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Value *Value `protobuf:"bytes,1,opt,name=value,proto3" json:"value,omitempty"` +} + +func (x *GetResponse) Reset() { + *x = GetResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_storage_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *GetResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetResponse) ProtoMessage() {} + +func (x *GetResponse) ProtoReflect() protoreflect.Message { + mi := &file_proto_storage_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetResponse.ProtoReflect.Descriptor instead. +func (*GetResponse) Descriptor() ([]byte, []int) { + return file_proto_storage_proto_rawDescGZIP(), []int{4} +} + +func (x *GetResponse) GetValue() *Value { + if x != nil { + return x.Value + } + return nil +} + +var File_proto_storage_proto protoreflect.FileDescriptor + +var file_proto_storage_proto_rawDesc = []byte{ + 0x0a, 0x13, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x1a, 0x1f, + 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, + 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, + 0x70, 0x0a, 0x05, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, + 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, + 0x61, 0x64, 0x12, 0x3e, 0x0a, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, + 0x6d, 0x70, 0x48, 0x00, 0x52, 0x09, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74, 0x88, + 0x01, 0x01, 0x42, 0x0d, 0x0a, 0x0b, 0x5f, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, + 0x74, 0x22, 0x32, 0x0a, 0x0a, 0x50, 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, + 0x24, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, + 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x05, + 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x23, 0x0a, 0x0b, 0x50, 0x75, 0x74, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x04, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x0c, 0x0a, 0x0a, 0x47, 0x65, + 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x33, 0x0a, 0x0b, 0x47, 0x65, 0x74, 0x52, + 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x24, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, + 0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x32, 0x77, 0x0a, + 0x07, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x12, 0x35, 0x0a, 0x08, 0x50, 0x75, 0x74, 0x56, + 0x61, 0x6c, 0x75, 0x65, 0x12, 0x13, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x50, + 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x73, 0x74, 0x6f, 0x72, + 0x61, 0x67, 0x65, 0x2e, 0x50, 0x75, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, + 0x35, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x13, 0x2e, 0x73, 0x74, + 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x14, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x47, 0x65, 0x74, 0x52, 0x65, + 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x14, 0x5a, 0x12, 0x68, 0x73, 0x65, 0x67, 0x72, 0x70, + 0x63, 0x2f, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, + 0x6f, 0x74, 0x6f, 0x33, +} + +var ( + file_proto_storage_proto_rawDescOnce sync.Once + file_proto_storage_proto_rawDescData = file_proto_storage_proto_rawDesc +) + +func file_proto_storage_proto_rawDescGZIP() []byte { + file_proto_storage_proto_rawDescOnce.Do(func() { + file_proto_storage_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_storage_proto_rawDescData) + }) + return file_proto_storage_proto_rawDescData +} + +var file_proto_storage_proto_msgTypes = make([]protoimpl.MessageInfo, 5) +var file_proto_storage_proto_goTypes = []interface{}{ + (*Value)(nil), // 0: storage.Value + (*PutRequest)(nil), // 1: storage.PutRequest + (*PutResponse)(nil), // 2: storage.PutResponse + (*GetRequest)(nil), // 3: storage.GetRequest + (*GetResponse)(nil), // 4: storage.GetResponse + (*timestamppb.Timestamp)(nil), // 5: google.protobuf.Timestamp +} +var file_proto_storage_proto_depIdxs = []int32{ + 5, // 0: storage.Value.updated_at:type_name -> google.protobuf.Timestamp + 0, // 1: storage.PutRequest.value:type_name -> storage.Value + 0, // 2: storage.GetResponse.value:type_name -> storage.Value + 1, // 3: storage.Storage.PutValue:input_type -> storage.PutRequest + 3, // 4: storage.Storage.GetValue:input_type -> storage.GetRequest + 2, // 5: storage.Storage.PutValue:output_type -> storage.PutResponse + 4, // 6: storage.Storage.GetValue:output_type -> storage.GetResponse + 5, // [5:7] is the sub-list for method output_type + 3, // [3:5] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name +} + +func init() { file_proto_storage_proto_init() } +func file_proto_storage_proto_init() { + if File_proto_storage_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_proto_storage_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Value); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_proto_storage_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*PutRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_proto_storage_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*PutResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_proto_storage_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*GetRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_proto_storage_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*GetResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + file_proto_storage_proto_msgTypes[0].OneofWrappers = []interface{}{} + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_proto_storage_proto_rawDesc, + NumEnums: 0, + NumMessages: 5, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_proto_storage_proto_goTypes, + DependencyIndexes: file_proto_storage_proto_depIdxs, + MessageInfos: file_proto_storage_proto_msgTypes, + }.Build() + File_proto_storage_proto = out.File + file_proto_storage_proto_rawDesc = nil + file_proto_storage_proto_goTypes = nil + file_proto_storage_proto_depIdxs = nil +} diff --git a/materials/02-communication/grpc-practice/go-server/proto/storage.proto b/materials/02-communication/grpc-practice/go-server/proto/storage.proto new file mode 100644 index 0000000..15513d7 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/proto/storage.proto @@ -0,0 +1,32 @@ +syntax = "proto3"; + +package storage; + +option go_package = "hsegrpc/;storagepb"; + +import "google/protobuf/timestamp.proto"; + +service Storage { + rpc PutValue(PutRequest) returns (PutResponse); + rpc GetValue(GetRequest) returns (GetResponse); +} + +message Value { + uint64 payload = 1; + optional google.protobuf.Timestamp updated_at = 2; +} + +message PutRequest { + Value value = 1; +} + +message PutResponse { + uint64 value = 1; +} + +message GetRequest { +} + +message GetResponse { + Value value = 1; +} diff --git a/materials/02-communication/grpc-practice/go-server/proto/storage_grpc.pb.go b/materials/02-communication/grpc-practice/go-server/proto/storage_grpc.pb.go new file mode 100644 index 0000000..d1db575 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/proto/storage_grpc.pb.go @@ -0,0 +1,141 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.2.0 +// - protoc v4.25.2 +// source: proto/storage.proto + +package storagepb + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.32.0 or later. +const _ = grpc.SupportPackageIsVersion7 + +// StorageClient is the client API for Storage service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type StorageClient interface { + PutValue(ctx context.Context, in *PutRequest, opts ...grpc.CallOption) (*PutResponse, error) + GetValue(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetResponse, error) +} + +type storageClient struct { + cc grpc.ClientConnInterface +} + +func NewStorageClient(cc grpc.ClientConnInterface) StorageClient { + return &storageClient{cc} +} + +func (c *storageClient) PutValue(ctx context.Context, in *PutRequest, opts ...grpc.CallOption) (*PutResponse, error) { + out := new(PutResponse) + err := c.cc.Invoke(ctx, "/storage.Storage/PutValue", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *storageClient) GetValue(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetResponse, error) { + out := new(GetResponse) + err := c.cc.Invoke(ctx, "/storage.Storage/GetValue", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// StorageServer is the server API for Storage service. +// All implementations must embed UnimplementedStorageServer +// for forward compatibility +type StorageServer interface { + PutValue(context.Context, *PutRequest) (*PutResponse, error) + GetValue(context.Context, *GetRequest) (*GetResponse, error) + mustEmbedUnimplementedStorageServer() +} + +// UnimplementedStorageServer must be embedded to have forward compatible implementations. +type UnimplementedStorageServer struct { +} + +func (UnimplementedStorageServer) PutValue(context.Context, *PutRequest) (*PutResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method PutValue not implemented") +} +func (UnimplementedStorageServer) GetValue(context.Context, *GetRequest) (*GetResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetValue not implemented") +} +func (UnimplementedStorageServer) mustEmbedUnimplementedStorageServer() {} + +// UnsafeStorageServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to StorageServer will +// result in compilation errors. +type UnsafeStorageServer interface { + mustEmbedUnimplementedStorageServer() +} + +func RegisterStorageServer(s grpc.ServiceRegistrar, srv StorageServer) { + s.RegisterService(&Storage_ServiceDesc, srv) +} + +func _Storage_PutValue_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(PutRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StorageServer).PutValue(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/storage.Storage/PutValue", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StorageServer).PutValue(ctx, req.(*PutRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Storage_GetValue_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StorageServer).GetValue(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/storage.Storage/GetValue", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StorageServer).GetValue(ctx, req.(*GetRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// Storage_ServiceDesc is the grpc.ServiceDesc for Storage service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var Storage_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "storage.Storage", + HandlerType: (*StorageServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "PutValue", + Handler: _Storage_PutValue_Handler, + }, + { + MethodName: "GetValue", + Handler: _Storage_GetValue_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "proto/storage.proto", +} diff --git a/materials/02-communication/grpc-practice/go-server/storage/server.go b/materials/02-communication/grpc-practice/go-server/storage/server.go new file mode 100644 index 0000000..aaf4f45 --- /dev/null +++ b/materials/02-communication/grpc-practice/go-server/storage/server.go @@ -0,0 +1,65 @@ +package storage + +import ( + "context" + "errors" + storagepb "hsegrpc/proto" + "sync" + "time" + + "google.golang.org/protobuf/types/known/timestamppb" +) + +type Value struct { + payload uint64 + updatedAt time.Time +} + +// toProto transforms struct into protobuf wrapper. +func (v *Value) toProto() *storagepb.Value { + return &storagepb.Value{ + Payload: v.payload, + UpdatedAt: timestamppb.New(v.updatedAt), + } +} + +type Server struct { + // Server must implement the Storage protobuf interface. + storagepb.UnimplementedStorageServer + + // We use mutex to synchronize access to the value. + valueLocker sync.RWMutex + value Value +} + +func NewServer() *Server { + return &Server{} +} + +func (s *Server) PutValue(_ context.Context, request *storagepb.PutRequest) (*storagepb.PutResponse, error) { + s.valueLocker.Lock() + defer s.valueLocker.Unlock() + + // Check if the given value not empty. + if request.GetValue() == nil { + return nil, errors.New("missed value") + } + + s.value = Value{ + payload: request.GetValue().GetPayload(), + updatedAt: time.Now(), + } + + return &storagepb.PutResponse{ + Value: s.value.payload, + }, nil +} + +func (s *Server) GetValue(_ context.Context, _ *storagepb.GetRequest) (*storagepb.GetResponse, error) { + s.valueLocker.RLock() + defer s.valueLocker.RUnlock() + + return &storagepb.GetResponse{ + Value: s.value.toProto(), + }, nil +} diff --git a/materials/02-communication/grpc-practice/python-client/Dockerfile b/materials/02-communication/grpc-practice/python-client/Dockerfile new file mode 100644 index 0000000..befa178 --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.12-slim + +COPY requirements.txt requirements.txt + +RUN pip install -r requirements.txt + +COPY . . + +CMD ["python", "./client.py"] \ No newline at end of file diff --git a/materials/02-communication/grpc-practice/python-client/README.md b/materials/02-communication/grpc-practice/python-client/README.md new file mode 100644 index 0000000..8d619b1 --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/README.md @@ -0,0 +1,33 @@ +# gRPC клиент + +В данной директории реализован простой пример клиента на Python, обращающийся к gRPC серверу. + +Программа запускает бесконечный цикл, в котором получает текущее значение с сервера и пытается +установить собственное значение, после чего засыпает на случайный промежуток времени. +Процедура повторяется в цикле. + +Конфигурация: +- `SERVER_ADDR` — адрес сервера для подключения. +- `VALUE_TO_PUT` — значение, которое будет устанавливать программа. + +## protobuf + +protubuf спецификация хранится в директории [./proto](./proto/). + +protobuf файл сервера должен являться надмножеством с точки зрения API относительно protobuf файла +клиента. Особенно важно сохранять оригинальные tag versions. + +```bash +pip3 install grpcio-tools + +python3 -m grpc_tools.protoc -I./proto --python_out=. --grpc_python_out=. ./proto/storage.proto +``` + +## Docker + +Сборка образа: +```bash +docker build -t hse-grpc-client . + +docker run -d -e SERVER_ADDR=... -e VALUE_TO_PUT=200 hse-grpc-client +``` diff --git a/materials/02-communication/grpc-practice/python-client/client.py b/materials/02-communication/grpc-practice/python-client/client.py new file mode 100644 index 0000000..5bf365f --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/client.py @@ -0,0 +1,35 @@ +import os +import random +import time + +import grpc +from google.protobuf.timestamp_pb2 import Timestamp + +import storage_pb2 +import storage_pb2_grpc + + +SERVER_ADDR = os.getenv('SERVER_ADDR', 'localhost:51000') +VALUE_TO_PUT = int(os.getenv('VALUE_TO_PUT', '100')) + + +channel = grpc.insecure_channel(SERVER_ADDR) +stub = storage_pb2_grpc.StorageStub(channel) + + +while True: + current = stub.GetValue(storage_pb2.GetRequest()) + payload = current.value.payload + updated_at = current.value.updated_at.ToDatetime() + + if payload != VALUE_TO_PUT: + print(f'Current value: {payload} (updated at {updated_at})') + + print(f'Putting {VALUE_TO_PUT}', flush=True) + + stub.PutValue(storage_pb2.PutRequest( + value=storage_pb2.Value(payload=VALUE_TO_PUT) + )) + + delay = random.uniform(1.0, 2.0) + time.sleep(delay) diff --git a/materials/02-communication/grpc-practice/python-client/proto/storage.proto b/materials/02-communication/grpc-practice/python-client/proto/storage.proto new file mode 100644 index 0000000..15513d7 --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/proto/storage.proto @@ -0,0 +1,32 @@ +syntax = "proto3"; + +package storage; + +option go_package = "hsegrpc/;storagepb"; + +import "google/protobuf/timestamp.proto"; + +service Storage { + rpc PutValue(PutRequest) returns (PutResponse); + rpc GetValue(GetRequest) returns (GetResponse); +} + +message Value { + uint64 payload = 1; + optional google.protobuf.Timestamp updated_at = 2; +} + +message PutRequest { + Value value = 1; +} + +message PutResponse { + uint64 value = 1; +} + +message GetRequest { +} + +message GetResponse { + Value value = 1; +} diff --git a/materials/02-communication/grpc-practice/python-client/requirements.txt b/materials/02-communication/grpc-practice/python-client/requirements.txt new file mode 100644 index 0000000..a064c2e --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/requirements.txt @@ -0,0 +1,3 @@ +grpcio==1.66.1 +grpcio-tools==1.66.1 +protobuf==5.28.1 \ No newline at end of file diff --git a/materials/02-communication/grpc-practice/python-client/storage_pb2.py b/materials/02-communication/grpc-practice/python-client/storage_pb2.py new file mode 100644 index 0000000..9abbdc3 --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/storage_pb2.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: storage.proto +# Protobuf Python Version: 5.27.2 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 5, + 27, + 2, + '', + 'storage.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\rstorage.proto\x12\x07storage\x1a\x1fgoogle/protobuf/timestamp.proto\"\\\n\x05Value\x12\x0f\n\x07payload\x18\x01 \x01(\x04\x12\x33\n\nupdated_at\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x00\x88\x01\x01\x42\r\n\x0b_updated_at\"+\n\nPutRequest\x12\x1d\n\x05value\x18\x01 \x01(\x0b\x32\x0e.storage.Value\"\x1c\n\x0bPutResponse\x12\r\n\x05value\x18\x01 \x01(\x04\"\x0c\n\nGetRequest\",\n\x0bGetResponse\x12\x1d\n\x05value\x18\x01 \x01(\x0b\x32\x0e.storage.Value2w\n\x07Storage\x12\x35\n\x08PutValue\x12\x13.storage.PutRequest\x1a\x14.storage.PutResponse\x12\x35\n\x08GetValue\x12\x13.storage.GetRequest\x1a\x14.storage.GetResponseB\x14Z\x12hsegrpc/;storagepbb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'storage_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + _globals['DESCRIPTOR']._loaded_options = None + _globals['DESCRIPTOR']._serialized_options = b'Z\022hsegrpc/;storagepb' + _globals['_VALUE']._serialized_start=59 + _globals['_VALUE']._serialized_end=151 + _globals['_PUTREQUEST']._serialized_start=153 + _globals['_PUTREQUEST']._serialized_end=196 + _globals['_PUTRESPONSE']._serialized_start=198 + _globals['_PUTRESPONSE']._serialized_end=226 + _globals['_GETREQUEST']._serialized_start=228 + _globals['_GETREQUEST']._serialized_end=240 + _globals['_GETRESPONSE']._serialized_start=242 + _globals['_GETRESPONSE']._serialized_end=286 + _globals['_STORAGE']._serialized_start=288 + _globals['_STORAGE']._serialized_end=407 +# @@protoc_insertion_point(module_scope) diff --git a/materials/02-communication/grpc-practice/python-client/storage_pb2_grpc.py b/materials/02-communication/grpc-practice/python-client/storage_pb2_grpc.py new file mode 100644 index 0000000..db506aa --- /dev/null +++ b/materials/02-communication/grpc-practice/python-client/storage_pb2_grpc.py @@ -0,0 +1,140 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc +import warnings + +import storage_pb2 as storage__pb2 + +GRPC_GENERATED_VERSION = '1.66.1' +GRPC_VERSION = grpc.__version__ +_version_not_supported = False + +try: + from grpc._utilities import first_version_is_lower + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f'The grpc package installed is at version {GRPC_VERSION},' + + f' but the generated code in storage_pb2_grpc.py depends on' + + f' grpcio>={GRPC_GENERATED_VERSION}.' + + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}' + + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.' + ) + + +class StorageStub(object): + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.PutValue = channel.unary_unary( + '/storage.Storage/PutValue', + request_serializer=storage__pb2.PutRequest.SerializeToString, + response_deserializer=storage__pb2.PutResponse.FromString, + _registered_method=True) + self.GetValue = channel.unary_unary( + '/storage.Storage/GetValue', + request_serializer=storage__pb2.GetRequest.SerializeToString, + response_deserializer=storage__pb2.GetResponse.FromString, + _registered_method=True) + + +class StorageServicer(object): + """Missing associated documentation comment in .proto file.""" + + def PutValue(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetValue(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_StorageServicer_to_server(servicer, server): + rpc_method_handlers = { + 'PutValue': grpc.unary_unary_rpc_method_handler( + servicer.PutValue, + request_deserializer=storage__pb2.PutRequest.FromString, + response_serializer=storage__pb2.PutResponse.SerializeToString, + ), + 'GetValue': grpc.unary_unary_rpc_method_handler( + servicer.GetValue, + request_deserializer=storage__pb2.GetRequest.FromString, + response_serializer=storage__pb2.GetResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'storage.Storage', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers('storage.Storage', rpc_method_handlers) + + + # This class is part of an EXPERIMENTAL API. +class Storage(object): + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def PutValue(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/storage.Storage/PutValue', + storage__pb2.PutRequest.SerializeToString, + storage__pb2.PutResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def GetValue(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/storage.Storage/GetValue', + storage__pb2.GetRequest.SerializeToString, + storage__pb2.GetResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/materials/02-communication/grpc-streaming/client/Dockerfile b/materials/02-communication/grpc-streaming/client/Dockerfile new file mode 100644 index 0000000..befa178 --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.12-slim + +COPY requirements.txt requirements.txt + +RUN pip install -r requirements.txt + +COPY . . + +CMD ["python", "./client.py"] \ No newline at end of file diff --git a/materials/02-communication/grpc-streaming/client/client.py b/materials/02-communication/grpc-streaming/client/client.py new file mode 100644 index 0000000..01baaf1 --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/client.py @@ -0,0 +1,24 @@ +import grpc +import os +import time + +import queue_pb2 +import queue_pb2_grpc + + +def request_generator(): + for i in range(5): + yield queue_pb2.PushRequest(value=queue_pb2.Value(payload=i)) + + +if __name__ == '__main__': + server_addr = os.getenv('SERVER_ADDR', 'localhost:51000') + with grpc.insecure_channel(server_addr) as channel: + stub = queue_pb2_grpc.QueueStub(channel) + stub.Push(queue_pb2.PushRequest(value=queue_pb2.Value(payload=100))) + response = stub.Pop(queue_pb2.PopRequest()) + print(f'Pop returned payload={response.value.payload}, updated_at={response.value.updated_at.ToDatetime()}') + time.sleep(1) + stub.PushMany(request_generator()) + for response in stub.Drain(queue_pb2.DrainRequest()): + print(f'Drain returned payload={response.value.payload}, updated_at={response.value.updated_at.ToDatetime()}') diff --git a/materials/02-communication/grpc-streaming/client/proto/queue.proto b/materials/02-communication/grpc-streaming/client/proto/queue.proto new file mode 100644 index 0000000..1c3817a --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/proto/queue.proto @@ -0,0 +1,34 @@ +syntax = "proto3"; + +package queue; + +import "google/protobuf/timestamp.proto"; + +service Queue { + rpc Push(PushRequest) returns (PushResponse); + rpc PushMany(stream PushRequest) returns (PushResponse); + rpc Pop(PopRequest) returns (PopResponse); + rpc Drain(DrainRequest) returns (stream PopResponse); +} + +message Value { + uint64 payload = 1; + optional google.protobuf.Timestamp updated_at = 2; +} + +message PushRequest { + Value value = 1; +} + +message PushResponse { +} + +message PopRequest { +} + +message PopResponse { + optional Value value = 1; +} + +message DrainRequest { +} diff --git a/materials/02-communication/grpc-streaming/client/queue_pb2.py b/materials/02-communication/grpc-streaming/client/queue_pb2.py new file mode 100644 index 0000000..0b134ab --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/queue_pb2.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: queue.proto +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0bqueue.proto\x12\x05queue\x1a\x1fgoogle/protobuf/timestamp.proto\"\\\n\x05Value\x12\x0f\n\x07payload\x18\x01 \x01(\x04\x12\x33\n\nupdated_at\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x00\x88\x01\x01\x42\r\n\x0b_updated_at\"*\n\x0bPushRequest\x12\x1b\n\x05value\x18\x01 \x01(\x0b\x32\x0c.queue.Value\"\x0e\n\x0cPushResponse\"\x0c\n\nPopRequest\"9\n\x0bPopResponse\x12 \n\x05value\x18\x01 \x01(\x0b\x32\x0c.queue.ValueH\x00\x88\x01\x01\x42\x08\n\x06_value\"\x0e\n\x0c\x44rainRequest2\xd1\x01\n\x05Queue\x12/\n\x04Push\x12\x12.queue.PushRequest\x1a\x13.queue.PushResponse\x12\x35\n\x08PushMany\x12\x12.queue.PushRequest\x1a\x13.queue.PushResponse(\x01\x12,\n\x03Pop\x12\x11.queue.PopRequest\x1a\x12.queue.PopResponse\x12\x32\n\x05\x44rain\x12\x13.queue.DrainRequest\x1a\x12.queue.PopResponse0\x01\x62\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'queue_pb2', _globals) +if _descriptor._USE_C_DESCRIPTORS == False: + + DESCRIPTOR._options = None + _globals['_VALUE']._serialized_start=55 + _globals['_VALUE']._serialized_end=147 + _globals['_PUSHREQUEST']._serialized_start=149 + _globals['_PUSHREQUEST']._serialized_end=191 + _globals['_PUSHRESPONSE']._serialized_start=193 + _globals['_PUSHRESPONSE']._serialized_end=207 + _globals['_POPREQUEST']._serialized_start=209 + _globals['_POPREQUEST']._serialized_end=221 + _globals['_POPRESPONSE']._serialized_start=223 + _globals['_POPRESPONSE']._serialized_end=280 + _globals['_DRAINREQUEST']._serialized_start=282 + _globals['_DRAINREQUEST']._serialized_end=296 + _globals['_QUEUE']._serialized_start=299 + _globals['_QUEUE']._serialized_end=508 +# @@protoc_insertion_point(module_scope) diff --git a/materials/02-communication/grpc-streaming/client/queue_pb2.pyi b/materials/02-communication/grpc-streaming/client/queue_pb2.pyi new file mode 100644 index 0000000..1cc9a89 --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/queue_pb2.pyi @@ -0,0 +1,38 @@ +from google.protobuf import timestamp_pb2 as _timestamp_pb2 +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from typing import ClassVar as _ClassVar, Mapping as _Mapping, Optional as _Optional, Union as _Union + +DESCRIPTOR: _descriptor.FileDescriptor + +class Value(_message.Message): + __slots__ = ["payload", "updated_at"] + PAYLOAD_FIELD_NUMBER: _ClassVar[int] + UPDATED_AT_FIELD_NUMBER: _ClassVar[int] + payload: int + updated_at: _timestamp_pb2.Timestamp + def __init__(self, payload: _Optional[int] = ..., updated_at: _Optional[_Union[_timestamp_pb2.Timestamp, _Mapping]] = ...) -> None: ... + +class PushRequest(_message.Message): + __slots__ = ["value"] + VALUE_FIELD_NUMBER: _ClassVar[int] + value: Value + def __init__(self, value: _Optional[_Union[Value, _Mapping]] = ...) -> None: ... + +class PushResponse(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... + +class PopRequest(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... + +class PopResponse(_message.Message): + __slots__ = ["value"] + VALUE_FIELD_NUMBER: _ClassVar[int] + value: Value + def __init__(self, value: _Optional[_Union[Value, _Mapping]] = ...) -> None: ... + +class DrainRequest(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... diff --git a/materials/02-communication/grpc-streaming/client/queue_pb2_grpc.py b/materials/02-communication/grpc-streaming/client/queue_pb2_grpc.py new file mode 100644 index 0000000..d5907ad --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/queue_pb2_grpc.py @@ -0,0 +1,165 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc + +import queue_pb2 as queue__pb2 + + +class QueueStub(object): + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.Push = channel.unary_unary( + '/queue.Queue/Push', + request_serializer=queue__pb2.PushRequest.SerializeToString, + response_deserializer=queue__pb2.PushResponse.FromString, + ) + self.PushMany = channel.stream_unary( + '/queue.Queue/PushMany', + request_serializer=queue__pb2.PushRequest.SerializeToString, + response_deserializer=queue__pb2.PushResponse.FromString, + ) + self.Pop = channel.unary_unary( + '/queue.Queue/Pop', + request_serializer=queue__pb2.PopRequest.SerializeToString, + response_deserializer=queue__pb2.PopResponse.FromString, + ) + self.Drain = channel.unary_stream( + '/queue.Queue/Drain', + request_serializer=queue__pb2.DrainRequest.SerializeToString, + response_deserializer=queue__pb2.PopResponse.FromString, + ) + + +class QueueServicer(object): + """Missing associated documentation comment in .proto file.""" + + def Push(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def PushMany(self, request_iterator, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def Pop(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def Drain(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_QueueServicer_to_server(servicer, server): + rpc_method_handlers = { + 'Push': grpc.unary_unary_rpc_method_handler( + servicer.Push, + request_deserializer=queue__pb2.PushRequest.FromString, + response_serializer=queue__pb2.PushResponse.SerializeToString, + ), + 'PushMany': grpc.stream_unary_rpc_method_handler( + servicer.PushMany, + request_deserializer=queue__pb2.PushRequest.FromString, + response_serializer=queue__pb2.PushResponse.SerializeToString, + ), + 'Pop': grpc.unary_unary_rpc_method_handler( + servicer.Pop, + request_deserializer=queue__pb2.PopRequest.FromString, + response_serializer=queue__pb2.PopResponse.SerializeToString, + ), + 'Drain': grpc.unary_stream_rpc_method_handler( + servicer.Drain, + request_deserializer=queue__pb2.DrainRequest.FromString, + response_serializer=queue__pb2.PopResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'queue.Queue', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + + + # This class is part of an EXPERIMENTAL API. +class Queue(object): + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def Push(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/queue.Queue/Push', + queue__pb2.PushRequest.SerializeToString, + queue__pb2.PushResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def PushMany(request_iterator, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.stream_unary(request_iterator, target, '/queue.Queue/PushMany', + queue__pb2.PushRequest.SerializeToString, + queue__pb2.PushResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def Pop(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/queue.Queue/Pop', + queue__pb2.PopRequest.SerializeToString, + queue__pb2.PopResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def Drain(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_stream(request, target, '/queue.Queue/Drain', + queue__pb2.DrainRequest.SerializeToString, + queue__pb2.PopResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) diff --git a/materials/02-communication/grpc-streaming/client/requirements.txt b/materials/02-communication/grpc-streaming/client/requirements.txt new file mode 100644 index 0000000..a064c2e --- /dev/null +++ b/materials/02-communication/grpc-streaming/client/requirements.txt @@ -0,0 +1,3 @@ +grpcio==1.66.1 +grpcio-tools==1.66.1 +protobuf==5.28.1 \ No newline at end of file diff --git a/materials/02-communication/grpc-streaming/docker-compose.yaml b/materials/02-communication/grpc-streaming/docker-compose.yaml new file mode 100644 index 0000000..5953fec --- /dev/null +++ b/materials/02-communication/grpc-streaming/docker-compose.yaml @@ -0,0 +1,16 @@ +services: + server: + build: + context: server + environment: + - SERVER_ADDR=0.0.0.0:51000 + ports: + - "51000:51000" + + client1: + build: + context: client + environment: + - SERVER_ADDR=server:51000 + depends_on: + - server diff --git a/materials/02-communication/grpc-streaming/server/Dockerfile b/materials/02-communication/grpc-streaming/server/Dockerfile new file mode 100644 index 0000000..e461776 --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.12-slim + +COPY requirements.txt requirements.txt + +RUN pip install -r requirements.txt + +COPY . . + +CMD ["python", "./server.py"] diff --git a/materials/02-communication/grpc-streaming/server/proto/queue.proto b/materials/02-communication/grpc-streaming/server/proto/queue.proto new file mode 100644 index 0000000..1c3817a --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/proto/queue.proto @@ -0,0 +1,34 @@ +syntax = "proto3"; + +package queue; + +import "google/protobuf/timestamp.proto"; + +service Queue { + rpc Push(PushRequest) returns (PushResponse); + rpc PushMany(stream PushRequest) returns (PushResponse); + rpc Pop(PopRequest) returns (PopResponse); + rpc Drain(DrainRequest) returns (stream PopResponse); +} + +message Value { + uint64 payload = 1; + optional google.protobuf.Timestamp updated_at = 2; +} + +message PushRequest { + Value value = 1; +} + +message PushResponse { +} + +message PopRequest { +} + +message PopResponse { + optional Value value = 1; +} + +message DrainRequest { +} diff --git a/materials/02-communication/grpc-streaming/server/queue_pb2.py b/materials/02-communication/grpc-streaming/server/queue_pb2.py new file mode 100644 index 0000000..0b134ab --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/queue_pb2.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: queue.proto +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0bqueue.proto\x12\x05queue\x1a\x1fgoogle/protobuf/timestamp.proto\"\\\n\x05Value\x12\x0f\n\x07payload\x18\x01 \x01(\x04\x12\x33\n\nupdated_at\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x00\x88\x01\x01\x42\r\n\x0b_updated_at\"*\n\x0bPushRequest\x12\x1b\n\x05value\x18\x01 \x01(\x0b\x32\x0c.queue.Value\"\x0e\n\x0cPushResponse\"\x0c\n\nPopRequest\"9\n\x0bPopResponse\x12 \n\x05value\x18\x01 \x01(\x0b\x32\x0c.queue.ValueH\x00\x88\x01\x01\x42\x08\n\x06_value\"\x0e\n\x0c\x44rainRequest2\xd1\x01\n\x05Queue\x12/\n\x04Push\x12\x12.queue.PushRequest\x1a\x13.queue.PushResponse\x12\x35\n\x08PushMany\x12\x12.queue.PushRequest\x1a\x13.queue.PushResponse(\x01\x12,\n\x03Pop\x12\x11.queue.PopRequest\x1a\x12.queue.PopResponse\x12\x32\n\x05\x44rain\x12\x13.queue.DrainRequest\x1a\x12.queue.PopResponse0\x01\x62\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'queue_pb2', _globals) +if _descriptor._USE_C_DESCRIPTORS == False: + + DESCRIPTOR._options = None + _globals['_VALUE']._serialized_start=55 + _globals['_VALUE']._serialized_end=147 + _globals['_PUSHREQUEST']._serialized_start=149 + _globals['_PUSHREQUEST']._serialized_end=191 + _globals['_PUSHRESPONSE']._serialized_start=193 + _globals['_PUSHRESPONSE']._serialized_end=207 + _globals['_POPREQUEST']._serialized_start=209 + _globals['_POPREQUEST']._serialized_end=221 + _globals['_POPRESPONSE']._serialized_start=223 + _globals['_POPRESPONSE']._serialized_end=280 + _globals['_DRAINREQUEST']._serialized_start=282 + _globals['_DRAINREQUEST']._serialized_end=296 + _globals['_QUEUE']._serialized_start=299 + _globals['_QUEUE']._serialized_end=508 +# @@protoc_insertion_point(module_scope) diff --git a/materials/02-communication/grpc-streaming/server/queue_pb2.pyi b/materials/02-communication/grpc-streaming/server/queue_pb2.pyi new file mode 100644 index 0000000..1cc9a89 --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/queue_pb2.pyi @@ -0,0 +1,38 @@ +from google.protobuf import timestamp_pb2 as _timestamp_pb2 +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from typing import ClassVar as _ClassVar, Mapping as _Mapping, Optional as _Optional, Union as _Union + +DESCRIPTOR: _descriptor.FileDescriptor + +class Value(_message.Message): + __slots__ = ["payload", "updated_at"] + PAYLOAD_FIELD_NUMBER: _ClassVar[int] + UPDATED_AT_FIELD_NUMBER: _ClassVar[int] + payload: int + updated_at: _timestamp_pb2.Timestamp + def __init__(self, payload: _Optional[int] = ..., updated_at: _Optional[_Union[_timestamp_pb2.Timestamp, _Mapping]] = ...) -> None: ... + +class PushRequest(_message.Message): + __slots__ = ["value"] + VALUE_FIELD_NUMBER: _ClassVar[int] + value: Value + def __init__(self, value: _Optional[_Union[Value, _Mapping]] = ...) -> None: ... + +class PushResponse(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... + +class PopRequest(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... + +class PopResponse(_message.Message): + __slots__ = ["value"] + VALUE_FIELD_NUMBER: _ClassVar[int] + value: Value + def __init__(self, value: _Optional[_Union[Value, _Mapping]] = ...) -> None: ... + +class DrainRequest(_message.Message): + __slots__ = [] + def __init__(self) -> None: ... diff --git a/materials/02-communication/grpc-streaming/server/queue_pb2_grpc.py b/materials/02-communication/grpc-streaming/server/queue_pb2_grpc.py new file mode 100644 index 0000000..d5907ad --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/queue_pb2_grpc.py @@ -0,0 +1,165 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc + +import queue_pb2 as queue__pb2 + + +class QueueStub(object): + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.Push = channel.unary_unary( + '/queue.Queue/Push', + request_serializer=queue__pb2.PushRequest.SerializeToString, + response_deserializer=queue__pb2.PushResponse.FromString, + ) + self.PushMany = channel.stream_unary( + '/queue.Queue/PushMany', + request_serializer=queue__pb2.PushRequest.SerializeToString, + response_deserializer=queue__pb2.PushResponse.FromString, + ) + self.Pop = channel.unary_unary( + '/queue.Queue/Pop', + request_serializer=queue__pb2.PopRequest.SerializeToString, + response_deserializer=queue__pb2.PopResponse.FromString, + ) + self.Drain = channel.unary_stream( + '/queue.Queue/Drain', + request_serializer=queue__pb2.DrainRequest.SerializeToString, + response_deserializer=queue__pb2.PopResponse.FromString, + ) + + +class QueueServicer(object): + """Missing associated documentation comment in .proto file.""" + + def Push(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def PushMany(self, request_iterator, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def Pop(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def Drain(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_QueueServicer_to_server(servicer, server): + rpc_method_handlers = { + 'Push': grpc.unary_unary_rpc_method_handler( + servicer.Push, + request_deserializer=queue__pb2.PushRequest.FromString, + response_serializer=queue__pb2.PushResponse.SerializeToString, + ), + 'PushMany': grpc.stream_unary_rpc_method_handler( + servicer.PushMany, + request_deserializer=queue__pb2.PushRequest.FromString, + response_serializer=queue__pb2.PushResponse.SerializeToString, + ), + 'Pop': grpc.unary_unary_rpc_method_handler( + servicer.Pop, + request_deserializer=queue__pb2.PopRequest.FromString, + response_serializer=queue__pb2.PopResponse.SerializeToString, + ), + 'Drain': grpc.unary_stream_rpc_method_handler( + servicer.Drain, + request_deserializer=queue__pb2.DrainRequest.FromString, + response_serializer=queue__pb2.PopResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'queue.Queue', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + + + # This class is part of an EXPERIMENTAL API. +class Queue(object): + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def Push(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/queue.Queue/Push', + queue__pb2.PushRequest.SerializeToString, + queue__pb2.PushResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def PushMany(request_iterator, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.stream_unary(request_iterator, target, '/queue.Queue/PushMany', + queue__pb2.PushRequest.SerializeToString, + queue__pb2.PushResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def Pop(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/queue.Queue/Pop', + queue__pb2.PopRequest.SerializeToString, + queue__pb2.PopResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def Drain(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_stream(request, target, '/queue.Queue/Drain', + queue__pb2.DrainRequest.SerializeToString, + queue__pb2.PopResponse.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) diff --git a/materials/02-communication/grpc-streaming/server/requirements.txt b/materials/02-communication/grpc-streaming/server/requirements.txt new file mode 100644 index 0000000..a064c2e --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/requirements.txt @@ -0,0 +1,3 @@ +grpcio==1.66.1 +grpcio-tools==1.66.1 +protobuf==5.28.1 \ No newline at end of file diff --git a/materials/02-communication/grpc-streaming/server/server.py b/materials/02-communication/grpc-streaming/server/server.py new file mode 100644 index 0000000..89d26b2 --- /dev/null +++ b/materials/02-communication/grpc-streaming/server/server.py @@ -0,0 +1,64 @@ +import grpc +import os +import queue_pb2 +import queue_pb2_grpc + +from collections import deque +from concurrent import futures +from datetime import datetime +from google.protobuf.timestamp_pb2 import Timestamp +from threading import Lock + + +class Queue: + data = deque() + lock = Lock() + + @classmethod + def push(cls, value): + with cls.lock: + cls.data.append(value) + + @classmethod + def pop(cls): + with cls.lock: + if len(cls.data) > 0: + return cls.data.popleft() + else: + return None + + @classmethod + def drain(cls): + with cls.lock: + data = list(cls.data) + cls.data = deque() + return data + + +class QueueService(queue_pb2_grpc.QueueServicer): + def Push(self, request, context): + request.value.updated_at.GetCurrentTime() + Queue.push(request.value) + return queue_pb2.PushResponse() + + def PushMany(self, request_iterator, context): + for request in request_iterator: + request.value.updated_at.GetCurrentTime() + Queue.push(request.value) + return queue_pb2.PushResponse() + + def Pop(self, request, context): + return queue_pb2.PopResponse(value=Queue.pop()) + + def Drain(self, request, context): + for item in Queue.drain(): + yield queue_pb2.PopResponse(value=item) + + +if __name__ == '__main__': + server_addr = os.getenv('SERVER_ADDR', 'localhost:51000') + server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + queue_pb2_grpc.add_QueueServicer_to_server(QueueService(), server) + server.add_insecure_port(server_addr) + server.start() + server.wait_for_termination(timeout=None) diff --git a/materials/02-communication/readme.md b/materials/02-communication/readme.md new file mode 100644 index 0000000..2eae8fc --- /dev/null +++ b/materials/02-communication/readme.md @@ -0,0 +1,18 @@ +## 2. Взаимодействие между процессами, надежная передача и RPC + +### Лекция + +- [Презентация](02-communication.pdf) +- См. также материалы в конце + +### Семинар + +- [Практика с gRPC](grpc-practice) +- [Пример gRPC Streaming](grpc-streaming) + +### Другие материалы + +- [Туториал по Docker](https://docker-curriculum.com/) +- [Основы работы с docker compose](https://docs.docker.com/compose/gettingstarted/) +- [Основы работы протокола TCP: хендшейки, congestion control и многое другое](https://hpbn.co/building-blocks-of-tcp/) +- [Официальная документация по gRPC с примерами использования](https://grpc.io/docs/languages/python/basics/)