v0.0.18.4 Фикс переподключения к rabbit
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
madipo2611 2025-08-17 16:01:29 +03:00
parent 7f48fb767f
commit d48eb31edc
2 changed files with 17 additions and 2 deletions

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
"log" "log"
"math"
"tailly_back_v2/internal/domain" "tailly_back_v2/internal/domain"
"tailly_back_v2/proto" "tailly_back_v2/proto"
"time" "time"
@ -126,16 +127,25 @@ func (r *subscriptionResolver) MessageStream(ctx context.Context, userID int) (<
go func() { go func() {
defer close(messageChan) defer close(messageChan)
retryDelay := time.Second
for { for {
stream, err := r.MessageClient.StreamMessages(ctx, &proto.StreamMessagesRequest{ // Создаем новый контекст с таймаутом
streamCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
stream, err := r.MessageClient.StreamMessages(streamCtx, &proto.StreamMessagesRequest{
UserId: int32(userID), UserId: int32(userID),
}) })
if err != nil { if err != nil {
cancel()
log.Printf("Failed to stream messages: %v", err) log.Printf("Failed to stream messages: %v", err)
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return
case <-time.After(5 * time.Second): // Задержка перед повторной попыткой case <-time.After(retryDelay):
retryDelay = time.Duration(math.Min(float64(retryDelay*2), float64(30*time.Second)))
continue continue
} }
} }
@ -143,11 +153,14 @@ func (r *subscriptionResolver) MessageStream(ctx context.Context, userID int) (<
for { for {
msg, err := stream.Recv() msg, err := stream.Recv()
if err != nil { if err != nil {
cancel()
log.Printf("Stream receive error: %v", err) log.Printf("Stream receive error: %v", err)
break break
} }
select { select {
case <-ctx.Done(): case <-ctx.Done():
cancel()
return return
case messageChan <- protoMessageToDomain(msg.Message): case messageChan <- protoMessageToDomain(msg.Message):
} }

View File

@ -17,6 +17,7 @@ import (
"tailly_back_v2/internal/http/middleware" "tailly_back_v2/internal/http/middleware"
"tailly_back_v2/internal/service" "tailly_back_v2/internal/service"
"tailly_back_v2/pkg/auth" "tailly_back_v2/pkg/auth"
"time"
) )
type Server struct { type Server struct {
@ -84,6 +85,7 @@ func (s *Server) configureRouter() {
return false return false
}, },
}, },
KeepAlivePingInterval: 10 * time.Second,
} }
srv.AddTransport(&wsTransport) srv.AddTransport(&wsTransport)