This commit is contained in:
parent
2a3ae72db9
commit
15ae17189d
@ -3,7 +3,6 @@ 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"
|
||||||
@ -121,10 +120,6 @@ 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) {
|
||||||
// Создаем канал для GraphQL подписки
|
|
||||||
messageChan := make(chan *domain.Message)
|
|
||||||
|
|
||||||
// Вызываем gRPC стриминг
|
|
||||||
stream, err := r.MessageClient.StreamMessages(ctx, &proto.StreamMessagesRequest{
|
stream, err := r.MessageClient.StreamMessages(ctx, &proto.StreamMessagesRequest{
|
||||||
UserId: int32(userID),
|
UserId: int32(userID),
|
||||||
})
|
})
|
||||||
@ -132,21 +127,14 @@ func (r *subscriptionResolver) MessageStream(ctx context.Context, userID int) (<
|
|||||||
return nil, fmt.Errorf("failed to stream messages: %w", err)
|
return nil, fmt.Errorf("failed to stream messages: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
messageChan := make(chan *domain.Message)
|
||||||
go func() {
|
go func() {
|
||||||
defer close(messageChan)
|
defer close(messageChan)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
msg, err := stream.Recv()
|
msg, err := stream.Recv()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("gRPC stream error: %v", err)
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Пропускаем heartbeat сообщения
|
|
||||||
if msg.Message.Content == "__heartbeat__" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
|
|||||||
@ -5,9 +5,10 @@ import (
|
|||||||
"database/sql"
|
"database/sql"
|
||||||
"github.com/99designs/gqlgen/graphql/handler"
|
"github.com/99designs/gqlgen/graphql/handler"
|
||||||
"github.com/99designs/gqlgen/graphql/handler/transport"
|
"github.com/99designs/gqlgen/graphql/handler/transport"
|
||||||
|
"github.com/99designs/gqlgen/graphql/playground"
|
||||||
"github.com/go-chi/chi/v5"
|
"github.com/go-chi/chi/v5"
|
||||||
|
"github.com/gorilla/websocket"
|
||||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||||
"github.com/vektah/gqlparser/v2/ast"
|
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
@ -71,29 +72,26 @@ func (s *Server) configureRouter() {
|
|||||||
Resolvers: resolver,
|
Resolvers: resolver,
|
||||||
}))
|
}))
|
||||||
|
|
||||||
srv.AddTransport(&transport.POST{})
|
wsTransport := transport.Websocket{
|
||||||
srv.AddTransport(&transport.GET{})
|
Upgrader: websocket.Upgrader{
|
||||||
srv.AddTransport(&transport.Options{})
|
CheckOrigin: func(r *http.Request) bool {
|
||||||
|
origin := r.Header.Get("Origin")
|
||||||
|
for _, allowed := range allowedOrigins {
|
||||||
|
if origin == allowed {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
// Настройка SSE транспорта (правильный способ)
|
srv.AddTransport(&wsTransport)
|
||||||
sseTransport := &transport.SSE{}
|
|
||||||
srv.AddTransport(sseTransport)
|
|
||||||
|
|
||||||
// Простейший кеш запросов (альтернатива MemoryCache)
|
|
||||||
srv.SetQueryCache(&NoopCache{})
|
|
||||||
})
|
|
||||||
|
|
||||||
s.router.Handle("/", playground.Handler("GraphQL playground", "/query"))
|
s.router.Handle("/", playground.Handler("GraphQL playground", "/query"))
|
||||||
s.router.Handle("/query", srv)
|
s.router.Handle("/query", srv)
|
||||||
s.router.Handle("/uploads/*", http.StripPrefix("/uploads/", http.FileServer(http.Dir("./uploads"))))
|
s.router.Handle("/uploads/*", http.StripPrefix("/uploads/", http.FileServer(http.Dir("./uploads"))))
|
||||||
}
|
}
|
||||||
type NoopCache struct{}
|
|
||||||
|
|
||||||
func (n *NoopCache) Get(ctx context.Context, key string) (*ast.QueryDocument, bool) {
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
|
|
||||||
func (n *NoopCache) Add(ctx context.Context, key string, value *ast.QueryDocument) {}
|
|
||||||
|
|
||||||
func (s *Server) configureMetrics() {
|
func (s *Server) configureMetrics() {
|
||||||
metricsRouter := chi.NewRouter()
|
metricsRouter := chi.NewRouter()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user