Godopu/grpccanary
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement real-time Pub/Sub alert broadcast system and update docs/GRPC.md

Browse Source
This commit is contained in:
Godopu
2026-07-17 19:24:25 +09:00
parent 8bb11fe205
commit 20a818520c
6 changed files with 444 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/grpcentity/client.go
+44
View File
@@ -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)
}
lib/grpcentity/protoapi.proto
+13
View File
@@ -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;
}
lib/grpcentity/protoapi/protoapi.pb.go
+157 -24
View File
@@ -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,
},
lib/grpcentity/protoapi/protoapi_grpc.pb.go
+41
View File
@@ -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",
}
lib/grpcentity/server.go
+68
View File
@@ -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