admin/tailly_messages
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v.0.0.3.3 Добавлено логирование для stream
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
madipo2611 2025-08-17 22:19:49 +03:00
parent cd8b954a70
commit b81f439eaf
1 changed files with 5 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
server.go
View File

@ -10,7 +10,6 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
protobuf "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
"log"
"net"
@ -173,24 +172,6 @@ func (s *server) SendMessage(ctx context.Context, req *proto.SendMessageRequest)
return nil, fmt.Errorf("failed to publish message after 3 attempts: %v", lastErr)
}
func mustMarshal(msg protobuf.Message) []byte {
data, err := protobuf.Marshal(msg)
if err != nil {
log.Fatalf("failed to marshal message: %v", err)
}
return data
}
func protoMessageToMap(msg *proto.Message) map[string]interface{} {
return map[string]interface{}{
"id": msg.Id,
"chatId": msg.ChatId,
"senderId": msg.SenderId,
"receiverId": msg.ReceiverId,
"content": msg.Content,
"status": msg.Status,
"createdAt": msg.CreatedAt.AsTime().Format(time.RFC3339Nano),
}
}
func (s *server) GetChat(ctx context.Context, req *proto.GetChatRequest) (*proto.ChatResponse, error) {
var chat proto.Chat
var createdAt, updatedAt time.Time
@ -366,9 +347,12 @@ func (s *server) StreamMessages(req *proto.StreamMessagesRequest, stream proto.M
func (s *server) runStream(req *proto.StreamMessagesRequest, stream proto.MessageService_StreamMessagesServer) error {
ctx := stream.Context()
log.Printf("ctx: %v", ctx)
queueName := fmt.Sprintf("user_%d_messages", req.UserId)
log.Printf("queue: %v", queueName)
ch, err := s.rabbitConn.Channel()
log.Printf("ch: %v", ch)
if err != nil {
return fmt.Errorf("failed to open channel: %v", err)
}
@ -411,6 +395,7 @@ func (s *server) runStream(req *proto.StreamMessagesRequest, stream proto.Messag
if err != nil {
return fmt.Errorf("failed to consume: %v", err)
}
log.Printf("msgs: %v", msgs)
log.Printf("Starting message stream for user %d", req.UserId)
defer log.Printf("Stopping message stream for user %d", req.UserId)
@ -433,6 +418,7 @@ func (s *server) runStream(req *proto.StreamMessagesRequest, stream proto.Messag
log.Printf("Sending message to user %d: %+v", req.UserId, msg)
if err := stream.Send(&proto.MessageResponse{Message: &msg}); err != nil {
log.Printf("Failed to send message to user %d: %v", req.UserId, err)
d.Nack(false, true) // Возвращаем в очередь при ошибке отправки
return err
}