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

откат до версии 0.2.2
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
madipo2611 2025-08-14 15:42:32 +03:00
parent 01c5ae11a2
commit b88154973c
1 changed files with 9 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
server.go
View File

@ -3,7 +3,6 @@ package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/segmentio/kafka-go"
@ -138,26 +137,16 @@ func (s *server) SendMessage(ctx context.Context, req *proto.SendMessageRequest)
return nil, err
}
// Создаем MessageResponse для Kafka
msgResponse := &proto.MessageResponse{Message: &message}
// Сериализуем сообщение в JSON для Kafka
msgBytes, err := json.Marshal(protoMessageToMap(msgResponse.Message))
if err != nil {
log.Printf("Failed to marshal message: %v", err)
return nil, err
}
// Publish to Kafka for real-time delivery
err = s.producer.WriteMessages(ctx, kafka.Message{
Key: []byte(fmt.Sprintf("%d", receiverId)),
Value: msgBytes,
Key: []byte(fmt.Sprintf("%d", receiverId)), // Ключ - ID получателя как строка
Value: mustMarshal(&proto.MessageResponse{Message: &message}), // Сериализуем proto-сообщение
})
if err != nil {
log.Printf("Failed to publish message to Kafka: %v", err)
}
return msgResponse, nil
return &proto.MessageResponse{Message: &message}, nil
}
func mustMarshal(msg protobuf.Message) []byte {
@ -167,17 +156,7 @@ func mustMarshal(msg protobuf.Message) []byte {
}
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
@ -347,29 +326,17 @@ func (s *server) StreamMessages(req *proto.StreamMessagesRequest, stream proto.M
return err
}
log.Printf("Received Kafka message: %s", string(m.Value))
log.Printf("Received Kafka message for user %d: key=%s", req.UserId, string(m.Key))
if string(m.Key) == fmt.Sprintf("%d", req.UserId) {
var msgData map[string]interface{}
if err := json.Unmarshal(m.Value, &msgData); err != nil {
log.Printf("Failed to unmarshal Kafka message: %v", err)
var msg proto.Message
if err := protobuf.Unmarshal(m.Value, &msg); err != nil {
log.Printf("Failed to unmarshal message: %v", err)
continue
}
// Преобразуем обратно в proto.Message
createdAt, _ := time.Parse(time.RFC3339Nano, msgData["createdAt"].(string))
msg := &proto.Message{
Id: msgData["id"].(int32),
ChatId: msgData["chatId"].(int32),
SenderId: msgData["senderId"].(int32),
ReceiverId: msgData["receiverId"].(int32),
Content: msgData["content"].(string),
Status: msgData["status"].(string),
CreatedAt: timestamppb.New(createdAt),
}
log.Printf("Sending message to user %d: %+v", req.UserId, msg)
if err := stream.Send(&proto.MessageResponse{Message: msg}); err != nil {
if err := stream.Send(&proto.MessageResponse{Message: &msg}); err != nil {
return err
}
}