107 lines
3.4 KiB
Go
107 lines
3.4 KiB
Go
package wt
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/centrifugal/centrifugo/v6/internal/logging"
|
|
|
|
"github.com/centrifugal/centrifuge"
|
|
"github.com/centrifugal/protocol"
|
|
"github.com/quic-go/quic-go"
|
|
"github.com/quic-go/webtransport-go"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// Handler for WebTransport.
|
|
type Handler struct {
|
|
config Config
|
|
server *webtransport.Server
|
|
node *centrifuge.Node
|
|
pingPong centrifuge.PingPongConfig
|
|
}
|
|
|
|
// NewHandler creates new Handler.
|
|
func NewHandler(node *centrifuge.Node, wtServer *webtransport.Server, config Config, pingPong centrifuge.PingPongConfig) *Handler {
|
|
return &Handler{config: config, server: wtServer, node: node, pingPong: pingPong}
|
|
}
|
|
|
|
const bidiStreamAcceptTimeout = 10 * time.Second
|
|
|
|
// ServeHTTP upgrades connection to WebTransport and creates centrifuge.Client from it.
|
|
func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
|
|
conn, err := s.server.Upgrade(rw, r)
|
|
if err != nil {
|
|
log.Info().Str("transport", transportName).Err(err).Msg("error upgrading to webtransport")
|
|
rw.WriteHeader(http.StatusBadRequest)
|
|
return
|
|
}
|
|
rw.WriteHeader(http.StatusOK)
|
|
|
|
acceptCtx, acceptCtxCancel := context.WithTimeout(r.Context(), bidiStreamAcceptTimeout)
|
|
stream, err := conn.AcceptStream(acceptCtx)
|
|
if err != nil {
|
|
acceptCtxCancel()
|
|
log.Error().Err(err).Msg("stream accept error")
|
|
rw.WriteHeader(http.StatusBadRequest)
|
|
return
|
|
}
|
|
acceptCtxCancel()
|
|
|
|
protoType := centrifuge.ProtocolTypeJSON
|
|
if r.URL.RawQuery != "" && r.URL.Query().Get("cf_protocol") == "protobuf" {
|
|
protoType = centrifuge.ProtocolTypeProtobuf
|
|
}
|
|
|
|
transport := newWebtransportTransport(protoType, conn, stream, s.pingPong, r.ProtoMajor)
|
|
c, closeFn, err := centrifuge.NewClient(r.Context(), s.node, transport)
|
|
if err != nil {
|
|
log.Error().Err(err).Msg("error creating client")
|
|
return
|
|
}
|
|
defer func() { _ = closeFn() }()
|
|
|
|
if logging.Enabled(logging.DebugLevel) {
|
|
log.Debug().Str("transport", transportName).Str("client", c.ID()).Msg("client connection established")
|
|
defer func(started time.Time) {
|
|
log.Debug().Str("transport", transportName).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
|
|
}(time.Now())
|
|
}
|
|
|
|
// Coerce a non-positive limit to the default, matching the other transports:
|
|
// the stream decoder requires a positive limit (it panics otherwise), so an
|
|
// explicit message_size_limit of 0 (a natural "use default" choice for
|
|
// operators) or a negative value must not reach it.
|
|
messageSizeLimit := int64(s.config.MessageSizeLimit)
|
|
if messageSizeLimit <= 0 {
|
|
messageSizeLimit = defaultWebTransportMessageSizeLimit
|
|
}
|
|
|
|
var decoder protocol.StreamCommandDecoder
|
|
if protoType == centrifuge.ProtocolTypeJSON {
|
|
decoder = protocol.GetStreamCommandDecoderLimited(protocol.TypeJSON, stream, messageSizeLimit)
|
|
defer protocol.PutStreamCommandDecoder(protocol.TypeJSON, decoder)
|
|
} else {
|
|
decoder = protocol.NewProtobufStreamCommandDecoder(stream, messageSizeLimit)
|
|
defer protocol.PutStreamCommandDecoder(protocol.TypeProtobuf, decoder)
|
|
}
|
|
|
|
for {
|
|
cmd, cmdSize, err := decoder.Decode()
|
|
if err != nil {
|
|
var appErr *quic.ApplicationError
|
|
if !errors.Is(err, io.EOF) && !errors.As(err, &appErr) {
|
|
log.Error().Err(err).Str("transport", transportName).Str("client", c.ID()).Msg("error decoding command")
|
|
}
|
|
return
|
|
}
|
|
ok := c.HandleCommand(cmd, cmdSize)
|
|
if !ok {
|
|
return
|
|
}
|
|
}
|
|
}
|