v0.0.18.3 Фикс переподключения к rabbit
All checks were successful
continuous-integration/drone/push Build is passing
All checks were successful
continuous-integration/drone/push Build is passing
This commit is contained in:
parent
15ae17189d
commit
7f48fb767f
@ -3,6 +3,7 @@ package graph
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log"
|
||||||
"tailly_back_v2/internal/domain"
|
"tailly_back_v2/internal/domain"
|
||||||
"tailly_back_v2/proto"
|
"tailly_back_v2/proto"
|
||||||
"time"
|
"time"
|
||||||
@ -120,25 +121,36 @@ func (r *queryResolver) GetUserChats(ctx context.Context, userID int) ([]*domain
|
|||||||
|
|
||||||
// MessageStream реализация подписки на новые сообщения
|
// MessageStream реализация подписки на новые сообщения
|
||||||
func (r *subscriptionResolver) MessageStream(ctx context.Context, userID int) (<-chan *domain.Message, error) {
|
func (r *subscriptionResolver) MessageStream(ctx context.Context, userID int) (<-chan *domain.Message, error) {
|
||||||
stream, err := r.MessageClient.StreamMessages(ctx, &proto.StreamMessagesRequest{
|
|
||||||
UserId: int32(userID),
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to stream messages: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
messageChan := make(chan *domain.Message)
|
messageChan := make(chan *domain.Message)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer close(messageChan)
|
defer close(messageChan)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
msg, err := stream.Recv()
|
stream, err := r.MessageClient.StreamMessages(ctx, &proto.StreamMessagesRequest{
|
||||||
|
UserId: int32(userID),
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
log.Printf("Failed to stream messages: %v", err)
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-time.After(5 * time.Second): // Задержка перед повторной попыткой
|
||||||
|
continue
|
||||||
|
}
|
||||||
}
|
}
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
for {
|
||||||
return
|
msg, err := stream.Recv()
|
||||||
case messageChan <- protoMessageToDomain(msg.Message):
|
if err != nil {
|
||||||
|
log.Printf("Stream receive error: %v", err)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case messageChan <- protoMessageToDomain(msg.Message):
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user