refactor: migrate lib/grpcentity to lib/grpc/basic and update path references
This commit is contained in:
@@ -0,0 +1,271 @@
|
||||
package basic
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"grpccanary/lib/grpc/basic/protoapi"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/reflection"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
var min = 0
|
||||
var max = 100
|
||||
var port = ":8080"
|
||||
|
||||
type UploadedFile struct {
|
||||
FileName string
|
||||
Content []byte
|
||||
UploadedAt int64
|
||||
}
|
||||
|
||||
var (
|
||||
fileStore = make(map[string]*UploadedFile)
|
||||
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
|
||||
}
|
||||
|
||||
// Extra function for creating secure random numbers
|
||||
//
|
||||
// func randomSecure(min, max int) int {
|
||||
// v, err := rand.Int(rand.Reader, big.NewInt(int64(max)))
|
||||
// if err != nil {
|
||||
// fmt.Println(err)
|
||||
// return min
|
||||
// }
|
||||
// fmt.Println("**", v, min, max)
|
||||
|
||||
// return min + int(v.Uint64())
|
||||
// }
|
||||
|
||||
func getString(len int64, src rand.Source) string {
|
||||
temp := ""
|
||||
startChar := "!"
|
||||
var i int64 = 1
|
||||
for {
|
||||
// For getting valid ASCII characters
|
||||
myRand := random(0, 94, src)
|
||||
newChar := string(startChar[0] + byte(myRand))
|
||||
temp = temp + newChar
|
||||
if i == len {
|
||||
break
|
||||
}
|
||||
i++
|
||||
}
|
||||
return temp
|
||||
}
|
||||
|
||||
type IoTServer struct {
|
||||
protoapi.UnimplementedIoTServiceServer
|
||||
}
|
||||
|
||||
func (IoTServer) GetDate(ctx context.Context, r *protoapi.RequestDateTime) (*protoapi.DateTime, error) {
|
||||
currentTime := time.Now()
|
||||
response := &protoapi.DateTime{
|
||||
Value: currentTime.String(),
|
||||
}
|
||||
|
||||
return response, nil
|
||||
}
|
||||
|
||||
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()),
|
||||
}
|
||||
|
||||
return response, nil
|
||||
}
|
||||
|
||||
func (IoTServer) GetRandomPass(ctx context.Context, r *protoapi.RequestPass) (*protoapi.RandomPass, error) {
|
||||
src := rand.NewSource(r.GetSeed())
|
||||
temp := getString(r.GetLength(), src)
|
||||
|
||||
response := &protoapi.RandomPass{
|
||||
Password: temp,
|
||||
}
|
||||
|
||||
return response, nil
|
||||
}
|
||||
|
||||
func (IoTServer) UploadFile(stream protoapi.IoTService_UploadFileServer) error {
|
||||
var totalBytes int64
|
||||
var fileName string
|
||||
var buffer []byte
|
||||
|
||||
for {
|
||||
chunk, err := stream.Recv()
|
||||
if err == io.EOF {
|
||||
fmt.Printf("File upload completed. Received %d bytes for file '%s'\n", totalBytes, fileName)
|
||||
|
||||
if fileName != "" {
|
||||
storeMu.Lock()
|
||||
fileStore[fileName] = &UploadedFile{
|
||||
FileName: fileName,
|
||||
Content: buffer,
|
||||
UploadedAt: time.Now().Unix(),
|
||||
}
|
||||
storeMu.Unlock()
|
||||
}
|
||||
|
||||
return stream.SendAndClose(&protoapi.UploadStatus{
|
||||
Success: true,
|
||||
Message: fmt.Sprintf("File '%s' uploaded successfully.", fileName),
|
||||
BytesUploaded: totalBytes,
|
||||
})
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Println("File upload error:", err)
|
||||
return err
|
||||
}
|
||||
|
||||
if fileName == "" {
|
||||
fileName = chunk.GetFileName()
|
||||
}
|
||||
buffer = append(buffer, chunk.GetContent()...)
|
||||
totalBytes += int64(len(chunk.GetContent()))
|
||||
}
|
||||
}
|
||||
|
||||
func (IoTServer) ListFiles(ctx context.Context, r *protoapi.EmptyRequest) (*protoapi.FileList, error) {
|
||||
storeMu.RLock()
|
||||
defer storeMu.RUnlock()
|
||||
|
||||
var files []*protoapi.FileMetadata
|
||||
for _, f := range fileStore {
|
||||
files = append(files, &protoapi.FileMetadata{
|
||||
FileName: f.FileName,
|
||||
FileSize: int64(len(f.Content)),
|
||||
UploadedAt: f.UploadedAt,
|
||||
})
|
||||
}
|
||||
|
||||
return &protoapi.FileList{Files: files}, nil
|
||||
}
|
||||
|
||||
func (IoTServer) DownloadFile(r *protoapi.DownloadRequest, stream protoapi.IoTService_DownloadFileServer) error {
|
||||
storeMu.RLock()
|
||||
f, exists := fileStore[r.GetFileName()]
|
||||
storeMu.RUnlock()
|
||||
|
||||
if !exists {
|
||||
return status.Errorf(codes.NotFound, "file %s not found", r.GetFileName())
|
||||
}
|
||||
|
||||
chunkSize := 1024 // 1KB 청크 단위
|
||||
totalBytes := len(f.Content)
|
||||
|
||||
for i := 0; i < totalBytes; i += chunkSize {
|
||||
end := i + chunkSize
|
||||
if end > totalBytes {
|
||||
end = totalBytes
|
||||
}
|
||||
|
||||
err := stream.Send(&protoapi.FileChunk{
|
||||
FileName: f.FileName,
|
||||
Content: f.Content[i:end],
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
protoapi.RegisterIoTServiceServer(server, iotServer)
|
||||
|
||||
reflection.Register(server)
|
||||
|
||||
listen, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Println("Serving requests...")
|
||||
server.Serve(listen)
|
||||
}
|
||||
Reference in New Issue
Block a user