From 20a818520ce85ad6b2de0fc1a4f88ca37d6b1912 Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 17 Jul 2026 19:24:25 +0900 Subject: [PATCH] feat: implement real-time Pub/Sub alert broadcast system and update docs/GRPC.md --- docs/GRPC.md | 121 +++++++++++++ lib/grpcentity/client.go | 44 +++++ lib/grpcentity/protoapi.proto | 13 ++ lib/grpcentity/protoapi/protoapi.pb.go | 181 +++++++++++++++++--- lib/grpcentity/protoapi/protoapi_grpc.pb.go | 41 +++++ lib/grpcentity/server.go | 68 ++++++++ 6 files changed, 444 insertions(+), 24 deletions(-) diff --git a/docs/GRPC.md b/docs/GRPC.md index 0693c6c..ab18cf3 100644 --- a/docs/GRPC.md +++ b/docs/GRPC.md @@ -429,6 +429,127 @@ func AskDownloadFile(ctx context.Context, m protoapi.IoTServiceClient, fileName * **목록 조회**: 빈 메시지(`EmptyRequest`)를 동봉해 Unary RPC 채널을 트리거하고 메타데이터 배열 결과를 동기 획득합니다. * **파일 다운로드**: 서버 스트리밍 엔드포인트 기동 후, `stream.Recv()` 블로킹 수신 루프에 진입합니다. 채널 해제 지점(`io.EOF`)에 도달할 때까지 메모리 버퍼 슬라이스에 청크 바이트 배열을 병합 누적하여 재조립(Reassembly)을 마친 후 반환합니다. +### 7.4 실시간 알림을 위한 Pub/Sub (발행/구독) 브로드캐스팅 구현 + +스마트 가전이나 센서 등 실시간 경보 통지가 필요한 AIoT 도메인에서는, 서버가 상시 대기하는 여러 디바이스(클라이언트)들에게 비동기로 이벤트를 밀어 넣어주는 **발행/구독(Publish/Subscribe)** 연동 구조가 필수적입니다. gRPC의 **서버 스트리밍(Server Streaming)** 채널을 응용하면, 다수의 클라이언트가 스트림 통로를 상시 유지한 채 대기하고, 서버가 특정 이벤트 발생 시 채널 리스트를 순회하며 실시간 이벤트를 **브로드캐스팅(Broadcasting)**하는 Pub/Sub 인프라를 단순하고 가볍게 완성할 수 있습니다. + +#### 1. 스키마 설계 (`protoapi.proto`) +구독 신청을 위한 파라미터(`AlertSubscription`)와 서버가 밀어 넣어줄 이벤트 규격(`AlertMessage`)을 IDL에 선언합니다: +```proto +message AlertSubscription { + string ClientId = 1; + string Topic = 2; // 구독할 주제 (예: "temperature_warnings") +} + +message AlertMessage { + string AlertId = 1; + string DeviceId = 2; + string Message = 3; // 실시간 발생 경보 문자열 + int64 Timestamp = 4; +} +``` + +#### 2. 서버 사이드 구독자 관리 및 발행 구현 ([server.go](../lib/grpcentity/server.go)) +서버는 구독을 신청한 클라이언트들에게 메시지를 안전하게 분배하기 위해 스레드 세이프 맵과 고루틴 채널(`chan`) 구조를 구성합니다: + +```go +type AlertSubscriber struct { + ClientId string + Channel chan *protoapi.AlertMessage +} + +var ( + subscribers = make(map[string]*AlertSubscriber) + subMu sync.Mutex +) + +// 실시간 모든 구독 채널에 알림 이벤트 분배 (Publish/Broadcast) +func publishAlert(alert *protoapi.AlertMessage) { + subMu.Lock() + defer subMu.Unlock() + for _, sub := range subscribers { + select { + case sub.Channel <- alert: + default: + // 채널 버퍼가 가득 찬 경우 병목 차단을 방지하기 위해 드롭 처리 + fmt.Printf("Alert channel blocked for client %s, dropping event\n", sub.ClientId) + } + } +} + +// 구독 채널 대기 핸들러 +func (IoTServer) SubscribeAlerts(r *protoapi.AlertSubscription, stream protoapi.IoTService_SubscribeAlertsServer) error { + clientId := r.GetClientId() + ch := make(chan *protoapi.AlertMessage, 10) // 버퍼 10의 수신 채널 생성 + sub := &AlertSubscriber{ + ClientId: clientId, + Channel: ch, + } + + subMu.Lock() + subscribers[clientId] = sub + subMu.Unlock() + + fmt.Printf("Client %s subscribed to alerts on topic '%s'\n", clientId, r.GetTopic()) + + // 스트림 연결 유지 및 채널 대기 감시 루프 + for { + select { + case alert := <-ch: + err := stream.Send(alert) + if err != nil { + subMu.Lock() + delete(subscribers, clientId) + subMu.Unlock() + return err + } + case <-stream.Context().Done(): + subMu.Lock() + delete(subscribers, clientId) + subMu.Unlock() + return nil + } + } +} +``` +* **쉬운 설명**: + * **구독 신청**: 클라이언트가 전화를 걸면 서버는 그 선을 닫지 않고 메모장(`subscribers`)에 해당 전화번호와 연결된 통로(Go 채널)를 적어둡니다. 그리고 그 선을 계속 붙잡고 대기(`stream.Send`) 상태를 유지합니다. + * **경보 발행**: 센서값 수신 핸들러(`UpdateSensingData`)에서 임계치(40도)를 초과하는 위험 열기가 감지되면, 메모장에 적힌 모든 연결된 통로에 경보 엽서(`AlertMessage`)를 휙 던져(Broadcast) 줍니다. +* **상세 설명**: + * **구독 신청**: `SubscribeAlerts` 엔드포인트는 호출과 동시에 전용 Go 비동기 버퍼 채널을 생성하고 전역 가입 맵에 등록합니다. `stream.Context().Done()` 채널 수신이나 스트림 유실 이벤트가 포착되기 전까지 루프 대기 상태를 안전하게 고정합니다. + * **경보 발행**: 동시성 경쟁 방지 락(`subMu.Lock()`) 임계 구역 내에서 연결된 모든 채널에 데이터를 `select-default` 논블로킹 패턴으로 분배 기입하여, 특정 클라이언트의 수신 병목이 서버 전체 성능에 미치는 파급 효과를 예방합니다. + +#### 3. 클라이언트 비동기 청취 구현 ([client.go](../lib/grpcentity/client.go)) +클라이언트는 메인 흐름을 방해하지 않고 알림을 백그라운드에서 실시간으로 대기 청취할 수 있도록 별도의 독자적인 비동기 고루틴 구조로 가동합니다. + +```go +func AskSubscribeAlerts(ctx context.Context, m protoapi.IoTServiceClient, clientId string, topic string) { + stream, err := m.SubscribeAlerts(ctx, &protoapi.AlertSubscription{ + ClientId: clientId, + Topic: topic, + }) + if err != nil { + fmt.Println("Failed to subscribe alerts:", err) + return + } + + for { + alert, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + break + } + fmt.Printf("\n🔔 [ALERT RECEIVED] ID: %s | Device: %s | Msg: %s | Time: %s\n\n", + alert.GetAlertId(), alert.GetDeviceId(), alert.GetMessage(), + time.Unix(alert.GetTimestamp(), 0).Format("15:04:05")) + } +} +``` +* **쉬운 설명**: 클라이언트는 메인 로직이 다른 볼일(파일 업로드/다운로드 등)을 보러 간 동안, 옆방에서 전화를 붙잡고 계속 귀를 기울이는 전담 직원(비동기 고루틴)을 기동시킵니다. 서버에서 "벨(알림)"이 울릴 때마다 그 내용을 즉시 가로채 화면에 실시간 경보 창을 출력해 줍니다. +* **상세 설명**: 메인 쓰레드의 블로킹을 방지하기 위해 Go의 경량 쓰레드 고루틴(`go AskSubscribeAlerts`)으로 리스너 루프를 위임 기동합니다. gRPC 스트림 클라이언트의 `stream.Recv()` 메서드는 서버로부터 메시지가 전달될 때까지 스레드 리소스를 낭비하지 않는 대기 상태로 머물며, 데이터 수령 시 콘솔 스트림에 이를 비동기 매핑 출력합니다. + --- ## 8. 한 걸음 더 나아가기 (다음 단계) diff --git a/lib/grpcentity/client.go b/lib/grpcentity/client.go index 13cfe6d..e33b32d 100644 --- a/lib/grpcentity/client.go +++ b/lib/grpcentity/client.go @@ -91,6 +91,31 @@ func AskDownloadFile(ctx context.Context, m protoapi.IoTServiceClient, fileName return buffer, nil } +func AskSubscribeAlerts(ctx context.Context, m protoapi.IoTServiceClient, clientId string, topic string) { + stream, err := m.SubscribeAlerts(ctx, &protoapi.AlertSubscription{ + ClientId: clientId, + Topic: topic, + }) + if err != nil { + fmt.Println("Failed to subscribe alerts:", err) + return + } + + for { + alert, err := stream.Recv() + if err == io.EOF { + fmt.Println("Alert subscription stream closed by server.") + break + } + if err != nil { + break + } + fmt.Printf("\n🔔 [ALERT RECEIVED] ID: %s | Device: %s | Msg: %s | Time: %s\n\n", + alert.GetAlertId(), alert.GetDeviceId(), alert.GetMessage(), + time.Unix(alert.GetTimestamp(), 0).Format("15:04:05")) + } +} + func ClientRun(addr string) { conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { @@ -99,6 +124,14 @@ func ClientRun(addr string) { } client := protoapi.NewIoTServiceClient(conn) + + // 백그라운드에서 실시간 경보 구독 기동 + alertCtx, alertCancel := context.WithCancel(context.Background()) + defer alertCancel() + go AskSubscribeAlerts(alertCtx, client, "client-app-01", "temperature_warnings") + // 구독 리시버 채널 등록을 위해 50ms 슬립 + time.Sleep(50 * time.Millisecond) + r, err := AskingDateTime(context.Background(), client) if err != nil { fmt.Println(err) @@ -171,4 +204,15 @@ func ClientRun(addr string) { } } fmt.Printf("Data Integrity Checked (Upload vs Download matches?): %t\n", isMatch) + + // 7단계: 임계값 초과 온습도 전송을 통한 Pub/Sub 실시간 알림 유발 시뮬레이션 + fmt.Println("Sending abnormal high-temperature sensing data (45.8°C)...") + alertRes, err := AskUpdateSensingData(context.Background(), client, "sensor-room-01", 45.8, 60.1) + if err != nil { + fmt.Println(err) + return + } + fmt.Println("Abnormal Sensing Update Success:", alertRes.Success) + // 알림 이벤트가 비동기로 화면에 출력될 시간을 확보하기 위해 100ms 대기 + time.Sleep(100 * time.Millisecond) } diff --git a/lib/grpcentity/protoapi.proto b/lib/grpcentity/protoapi.proto index 0ff2a48..c530523 100644 --- a/lib/grpcentity/protoapi.proto +++ b/lib/grpcentity/protoapi.proto @@ -9,6 +9,7 @@ service IoTService { rpc UploadFile (stream FileChunk) returns (UploadStatus); rpc ListFiles (EmptyRequest) returns (FileList); rpc DownloadFile (DownloadRequest) returns (stream FileChunk); + rpc SubscribeAlerts (AlertSubscription) returns (stream AlertMessage); } message FileChunk { @@ -64,4 +65,16 @@ message FileList { message DownloadRequest { string FileName = 1; +} + +message AlertSubscription { + string ClientId = 1; + string Topic = 2; +} + +message AlertMessage { + string AlertId = 1; + string DeviceId = 2; + string Message = 3; + int64 Timestamp = 4; } \ No newline at end of file diff --git a/lib/grpcentity/protoapi/protoapi.pb.go b/lib/grpcentity/protoapi/protoapi.pb.go index b131b06..73d6699 100644 --- a/lib/grpcentity/protoapi/protoapi.pb.go +++ b/lib/grpcentity/protoapi/protoapi.pb.go @@ -613,6 +613,126 @@ func (x *DownloadRequest) GetFileName() string { return "" } +type AlertSubscription struct { + state protoimpl.MessageState `protogen:"open.v1"` + ClientId string `protobuf:"bytes,1,opt,name=ClientId,proto3" json:"ClientId,omitempty"` + Topic string `protobuf:"bytes,2,opt,name=Topic,proto3" json:"Topic,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AlertSubscription) Reset() { + *x = AlertSubscription{} + mi := &file_protoapi_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AlertSubscription) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AlertSubscription) ProtoMessage() {} + +func (x *AlertSubscription) ProtoReflect() protoreflect.Message { + mi := &file_protoapi_proto_msgTypes[12] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AlertSubscription.ProtoReflect.Descriptor instead. +func (*AlertSubscription) Descriptor() ([]byte, []int) { + return file_protoapi_proto_rawDescGZIP(), []int{12} +} + +func (x *AlertSubscription) GetClientId() string { + if x != nil { + return x.ClientId + } + return "" +} + +func (x *AlertSubscription) GetTopic() string { + if x != nil { + return x.Topic + } + return "" +} + +type AlertMessage struct { + state protoimpl.MessageState `protogen:"open.v1"` + AlertId string `protobuf:"bytes,1,opt,name=AlertId,proto3" json:"AlertId,omitempty"` + DeviceId string `protobuf:"bytes,2,opt,name=DeviceId,proto3" json:"DeviceId,omitempty"` + Message string `protobuf:"bytes,3,opt,name=Message,proto3" json:"Message,omitempty"` + Timestamp int64 `protobuf:"varint,4,opt,name=Timestamp,proto3" json:"Timestamp,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AlertMessage) Reset() { + *x = AlertMessage{} + mi := &file_protoapi_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AlertMessage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AlertMessage) ProtoMessage() {} + +func (x *AlertMessage) ProtoReflect() protoreflect.Message { + mi := &file_protoapi_proto_msgTypes[13] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AlertMessage.ProtoReflect.Descriptor instead. +func (*AlertMessage) Descriptor() ([]byte, []int) { + return file_protoapi_proto_rawDescGZIP(), []int{13} +} + +func (x *AlertMessage) GetAlertId() string { + if x != nil { + return x.AlertId + } + return "" +} + +func (x *AlertMessage) GetDeviceId() string { + if x != nil { + return x.DeviceId + } + return "" +} + +func (x *AlertMessage) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +func (x *AlertMessage) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + var File_protoapi_proto protoreflect.FileDescriptor const file_protoapi_proto_rawDesc = "" + @@ -652,7 +772,15 @@ const file_protoapi_proto_rawDesc = "" + "\bFileList\x12#\n" + "\x05Files\x18\x01 \x03(\v2\r.FileMetadataR\x05Files\"-\n" + "\x0fDownloadRequest\x12\x1a\n" + - "\bFileName\x18\x01 \x01(\tR\bFileName2\x97\x02\n" + + "\bFileName\x18\x01 \x01(\tR\bFileName\"E\n" + + "\x11AlertSubscription\x12\x1a\n" + + "\bClientId\x18\x01 \x01(\tR\bClientId\x12\x14\n" + + "\x05Topic\x18\x02 \x01(\tR\x05Topic\"|\n" + + "\fAlertMessage\x12\x18\n" + + "\aAlertId\x18\x01 \x01(\tR\aAlertId\x12\x1a\n" + + "\bDeviceId\x18\x02 \x01(\tR\bDeviceId\x12\x18\n" + + "\aMessage\x18\x03 \x01(\tR\aMessage\x12\x1c\n" + + "\tTimestamp\x18\x04 \x01(\x03R\tTimestamp2\xcf\x02\n" + "\n" + "IoTService\x12&\n" + "\aGetDate\x12\x10.RequestDateTime\x1a\t.DateTime\x123\n" + @@ -663,7 +791,8 @@ const file_protoapi_proto_rawDesc = "" + ".FileChunk\x1a\r.UploadStatus(\x01\x12%\n" + "\tListFiles\x12\r.EmptyRequest\x1a\t.FileList\x12.\n" + "\fDownloadFile\x12\x10.DownloadRequest\x1a\n" + - ".FileChunk0\x01B\x16Z\x14./protoapi/;protoapib\x06proto3" + ".FileChunk0\x01\x126\n" + + "\x0fSubscribeAlerts\x12\x12.AlertSubscription\x1a\r.AlertMessage0\x01B\x16Z\x14./protoapi/;protoapib\x06proto3" var ( file_protoapi_proto_rawDescOnce sync.Once @@ -677,20 +806,22 @@ func file_protoapi_proto_rawDescGZIP() []byte { return file_protoapi_proto_rawDescData } -var file_protoapi_proto_msgTypes = make([]protoimpl.MessageInfo, 12) +var file_protoapi_proto_msgTypes = make([]protoimpl.MessageInfo, 14) var file_protoapi_proto_goTypes = []any{ - (*FileChunk)(nil), // 0: FileChunk - (*UploadStatus)(nil), // 1: UploadStatus - (*SensingData)(nil), // 2: SensingData - (*SensingResponse)(nil), // 3: SensingResponse - (*DateTime)(nil), // 4: DateTime - (*RequestDateTime)(nil), // 5: RequestDateTime - (*RequestPass)(nil), // 6: RequestPass - (*RandomPass)(nil), // 7: RandomPass - (*EmptyRequest)(nil), // 8: EmptyRequest - (*FileMetadata)(nil), // 9: FileMetadata - (*FileList)(nil), // 10: FileList - (*DownloadRequest)(nil), // 11: DownloadRequest + (*FileChunk)(nil), // 0: FileChunk + (*UploadStatus)(nil), // 1: UploadStatus + (*SensingData)(nil), // 2: SensingData + (*SensingResponse)(nil), // 3: SensingResponse + (*DateTime)(nil), // 4: DateTime + (*RequestDateTime)(nil), // 5: RequestDateTime + (*RequestPass)(nil), // 6: RequestPass + (*RandomPass)(nil), // 7: RandomPass + (*EmptyRequest)(nil), // 8: EmptyRequest + (*FileMetadata)(nil), // 9: FileMetadata + (*FileList)(nil), // 10: FileList + (*DownloadRequest)(nil), // 11: DownloadRequest + (*AlertSubscription)(nil), // 12: AlertSubscription + (*AlertMessage)(nil), // 13: AlertMessage } var file_protoapi_proto_depIdxs = []int32{ 9, // 0: FileList.Files:type_name -> FileMetadata @@ -700,14 +831,16 @@ var file_protoapi_proto_depIdxs = []int32{ 0, // 4: IoTService.UploadFile:input_type -> FileChunk 8, // 5: IoTService.ListFiles:input_type -> EmptyRequest 11, // 6: IoTService.DownloadFile:input_type -> DownloadRequest - 4, // 7: IoTService.GetDate:output_type -> DateTime - 3, // 8: IoTService.UpdateSensingData:output_type -> SensingResponse - 7, // 9: IoTService.GetRandomPass:output_type -> RandomPass - 1, // 10: IoTService.UploadFile:output_type -> UploadStatus - 10, // 11: IoTService.ListFiles:output_type -> FileList - 0, // 12: IoTService.DownloadFile:output_type -> FileChunk - 7, // [7:13] is the sub-list for method output_type - 1, // [1:7] is the sub-list for method input_type + 12, // 7: IoTService.SubscribeAlerts:input_type -> AlertSubscription + 4, // 8: IoTService.GetDate:output_type -> DateTime + 3, // 9: IoTService.UpdateSensingData:output_type -> SensingResponse + 7, // 10: IoTService.GetRandomPass:output_type -> RandomPass + 1, // 11: IoTService.UploadFile:output_type -> UploadStatus + 10, // 12: IoTService.ListFiles:output_type -> FileList + 0, // 13: IoTService.DownloadFile:output_type -> FileChunk + 13, // 14: IoTService.SubscribeAlerts:output_type -> AlertMessage + 8, // [8:15] is the sub-list for method output_type + 1, // [1:8] is the sub-list for method input_type 1, // [1:1] is the sub-list for extension type_name 1, // [1:1] is the sub-list for extension extendee 0, // [0:1] is the sub-list for field type_name @@ -724,7 +857,7 @@ func file_protoapi_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_protoapi_proto_rawDesc), len(file_protoapi_proto_rawDesc)), NumEnums: 0, - NumMessages: 12, + NumMessages: 14, NumExtensions: 0, NumServices: 1, }, diff --git a/lib/grpcentity/protoapi/protoapi_grpc.pb.go b/lib/grpcentity/protoapi/protoapi_grpc.pb.go index 1321f0e..3143a44 100644 --- a/lib/grpcentity/protoapi/protoapi_grpc.pb.go +++ b/lib/grpcentity/protoapi/protoapi_grpc.pb.go @@ -25,6 +25,7 @@ const ( IoTService_UploadFile_FullMethodName = "/IoTService/UploadFile" IoTService_ListFiles_FullMethodName = "/IoTService/ListFiles" IoTService_DownloadFile_FullMethodName = "/IoTService/DownloadFile" + IoTService_SubscribeAlerts_FullMethodName = "/IoTService/SubscribeAlerts" ) // IoTServiceClient is the client API for IoTService service. @@ -37,6 +38,7 @@ type IoTServiceClient interface { UploadFile(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[FileChunk, UploadStatus], error) ListFiles(ctx context.Context, in *EmptyRequest, opts ...grpc.CallOption) (*FileList, error) DownloadFile(ctx context.Context, in *DownloadRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[FileChunk], error) + SubscribeAlerts(ctx context.Context, in *AlertSubscription, opts ...grpc.CallOption) (grpc.ServerStreamingClient[AlertMessage], error) } type ioTServiceClient struct { @@ -119,6 +121,25 @@ func (c *ioTServiceClient) DownloadFile(ctx context.Context, in *DownloadRequest // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type IoTService_DownloadFileClient = grpc.ServerStreamingClient[FileChunk] +func (c *ioTServiceClient) SubscribeAlerts(ctx context.Context, in *AlertSubscription, opts ...grpc.CallOption) (grpc.ServerStreamingClient[AlertMessage], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &IoTService_ServiceDesc.Streams[2], IoTService_SubscribeAlerts_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[AlertSubscription, AlertMessage]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type IoTService_SubscribeAlertsClient = grpc.ServerStreamingClient[AlertMessage] + // IoTServiceServer is the server API for IoTService service. // All implementations must embed UnimplementedIoTServiceServer // for forward compatibility. @@ -129,6 +150,7 @@ type IoTServiceServer interface { UploadFile(grpc.ClientStreamingServer[FileChunk, UploadStatus]) error ListFiles(context.Context, *EmptyRequest) (*FileList, error) DownloadFile(*DownloadRequest, grpc.ServerStreamingServer[FileChunk]) error + SubscribeAlerts(*AlertSubscription, grpc.ServerStreamingServer[AlertMessage]) error mustEmbedUnimplementedIoTServiceServer() } @@ -157,6 +179,9 @@ func (UnimplementedIoTServiceServer) ListFiles(context.Context, *EmptyRequest) ( func (UnimplementedIoTServiceServer) DownloadFile(*DownloadRequest, grpc.ServerStreamingServer[FileChunk]) error { return status.Error(codes.Unimplemented, "method DownloadFile not implemented") } +func (UnimplementedIoTServiceServer) SubscribeAlerts(*AlertSubscription, grpc.ServerStreamingServer[AlertMessage]) error { + return status.Error(codes.Unimplemented, "method SubscribeAlerts not implemented") +} func (UnimplementedIoTServiceServer) mustEmbedUnimplementedIoTServiceServer() {} func (UnimplementedIoTServiceServer) testEmbeddedByValue() {} @@ -268,6 +293,17 @@ func _IoTService_DownloadFile_Handler(srv interface{}, stream grpc.ServerStream) // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type IoTService_DownloadFileServer = grpc.ServerStreamingServer[FileChunk] +func _IoTService_SubscribeAlerts_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(AlertSubscription) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(IoTServiceServer).SubscribeAlerts(m, &grpc.GenericServerStream[AlertSubscription, AlertMessage]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type IoTService_SubscribeAlertsServer = grpc.ServerStreamingServer[AlertMessage] + // IoTService_ServiceDesc is the grpc.ServiceDesc for IoTService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -303,6 +339,11 @@ var IoTService_ServiceDesc = grpc.ServiceDesc{ Handler: _IoTService_DownloadFile_Handler, ServerStreams: true, }, + { + StreamName: "SubscribeAlerts", + Handler: _IoTService_SubscribeAlerts_Handler, + ServerStreams: true, + }, }, Metadata: "protoapi.proto", } diff --git a/lib/grpcentity/server.go b/lib/grpcentity/server.go index 9274665..4f8c78b 100644 --- a/lib/grpcentity/server.go +++ b/lib/grpcentity/server.go @@ -31,6 +31,28 @@ var ( storeMu sync.RWMutex ) +type AlertSubscriber struct { + ClientId string + Channel chan *protoapi.AlertMessage +} + +var ( + subscribers = make(map[string]*AlertSubscriber) + subMu sync.Mutex +) + +func publishAlert(alert *protoapi.AlertMessage) { + subMu.Lock() + defer subMu.Unlock() + for _, sub := range subscribers { + select { + case sub.Channel <- alert: + default: + fmt.Printf("Alert channel blocked for client %s, dropping event\n", sub.ClientId) + } + } +} + func random(min, max int, src rand.Source) int { return rand.New(src).Intn(max-min) + min } @@ -81,6 +103,17 @@ func (IoTServer) GetDate(ctx context.Context, r *protoapi.RequestDateTime) (*pro func (IoTServer) UpdateSensingData(ctx context.Context, r *protoapi.SensingData) (*protoapi.SensingResponse, error) { fmt.Printf("Received sensing data - Device: %s, Temp: %.2f°C, Humid: %.2f%%\n", r.GetDeviceId(), r.GetTemperature(), r.GetHumidity()) + // 임계값 초과(40도 초과) 시 실시간 Pub/Sub 경보 메시지 발행 + if r.GetTemperature() > 40.0 { + fmt.Printf("⚠️ Critical temperature detected: %.2f°C! Publishing warning...\n", r.GetTemperature()) + publishAlert(&protoapi.AlertMessage{ + AlertId: fmt.Sprintf("alert-%d", time.Now().UnixNano()), + DeviceId: r.GetDeviceId(), + Message: fmt.Sprintf("Critical high temperature: %.2f°C (Humidity: %.2f%%)", r.GetTemperature(), r.GetHumidity()), + Timestamp: time.Now().Unix(), + }) + } + response := &protoapi.SensingResponse{ Success: true, Message: fmt.Sprintf("Sensing data updated successfully for device %s", r.GetDeviceId()), @@ -185,6 +218,41 @@ func (IoTServer) DownloadFile(r *protoapi.DownloadRequest, stream protoapi.IoTSe return nil } +func (IoTServer) SubscribeAlerts(r *protoapi.AlertSubscription, stream protoapi.IoTService_SubscribeAlertsServer) error { + clientId := r.GetClientId() + ch := make(chan *protoapi.AlertMessage, 10) + sub := &AlertSubscriber{ + ClientId: clientId, + Channel: ch, + } + + subMu.Lock() + subscribers[clientId] = sub + subMu.Unlock() + + fmt.Printf("Client %s subscribed to alerts on topic '%s'\n", clientId, r.GetTopic()) + + for { + select { + case alert := <-ch: + err := stream.Send(alert) + if err != nil { + subMu.Lock() + delete(subscribers, clientId) + subMu.Unlock() + fmt.Printf("Client %s alert subscription disconnected: %v\n", clientId, err) + return err + } + case <-stream.Context().Done(): + subMu.Lock() + delete(subscribers, clientId) + subMu.Unlock() + fmt.Printf("Client %s unsubscribed (context done)\n", clientId) + return nil + } + } +} + func ServerRun(addr string) { server := grpc.NewServer() var iotServer IoTServer