1341 lines
37 KiB
Go
1341 lines
37 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"errors"
|
|
"sync"
|
|
"time"
|
|
|
|
. "github.com/centrifugal/centrifugo/v6/internal/apiproto"
|
|
"github.com/centrifugal/centrifugo/v6/internal/config"
|
|
"github.com/centrifugal/centrifugo/v6/internal/configtypes"
|
|
"github.com/centrifugal/centrifugo/v6/internal/metrics"
|
|
"github.com/centrifugal/centrifugo/v6/internal/subsource"
|
|
|
|
"github.com/centrifugal/centrifuge"
|
|
"github.com/rs/zerolog/log"
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/trace"
|
|
)
|
|
|
|
// RPCHandler allows to handle custom RPC.
|
|
type RPCHandler func(ctx context.Context, params Raw) (Raw, error)
|
|
|
|
// Executor can run API methods.
|
|
type Executor struct {
|
|
node *centrifuge.Node
|
|
cfgContainer *config.Container
|
|
config ExecutorConfig
|
|
rpcExtension map[string]RPCHandler
|
|
surveyCaller SurveyCaller
|
|
}
|
|
|
|
// SurveyCaller can do surveys.
|
|
type SurveyCaller interface {
|
|
Channels(ctx context.Context, cmd *ChannelsRequest) (map[string]*ChannelInfo, error)
|
|
}
|
|
|
|
type ExecutorConfig struct {
|
|
Protocol string
|
|
UseOpenTelemetry bool
|
|
}
|
|
|
|
// NewExecutor ...
|
|
func NewExecutor(n *centrifuge.Node, cfgContainer *config.Container, surveyCaller SurveyCaller, config ExecutorConfig) *Executor {
|
|
e := &Executor{
|
|
node: n,
|
|
cfgContainer: cfgContainer,
|
|
config: config,
|
|
surveyCaller: surveyCaller,
|
|
rpcExtension: make(map[string]RPCHandler),
|
|
}
|
|
return e
|
|
}
|
|
|
|
// SetRPCExtension ...
|
|
func (h *Executor) SetRPCExtension(method string, handler RPCHandler) {
|
|
h.rpcExtension[method] = handler
|
|
}
|
|
|
|
func (h *Executor) processCmd(ctx context.Context, cmd *Command, i int, replies []*Reply) {
|
|
if cmd == nil {
|
|
// A batch can carry a nil command (e.g. a JSON `null` array element). In
|
|
// parallel mode processCmd runs in its own goroutine, so a nil-deref here
|
|
// would escape net/http's per-request recover and crash the whole process;
|
|
// reject it as a bad request instead.
|
|
replies[i].Error = ErrorBadRequest
|
|
return
|
|
}
|
|
var method string
|
|
if cmd.Publish != nil {
|
|
method = "publish"
|
|
res := h.Publish(ctx, cmd.Publish)
|
|
replies[i].Publish, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Broadcast != nil {
|
|
method = "broadcast"
|
|
res := h.Broadcast(ctx, cmd.Broadcast)
|
|
replies[i].Broadcast, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Subscribe != nil {
|
|
method = "subscribe"
|
|
res := h.Subscribe(ctx, cmd.Subscribe)
|
|
replies[i].Subscribe, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Unsubscribe != nil {
|
|
method = "unsubscribe"
|
|
res := h.Unsubscribe(ctx, cmd.Unsubscribe)
|
|
replies[i].Unsubscribe, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Disconnect != nil {
|
|
method = "disconnect"
|
|
res := h.Disconnect(ctx, cmd.Disconnect)
|
|
replies[i].Disconnect, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.History != nil {
|
|
method = "history"
|
|
res := h.History(ctx, cmd.History)
|
|
replies[i].History, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.HistoryRemove != nil {
|
|
method = "history_remove"
|
|
res := h.HistoryRemove(ctx, cmd.HistoryRemove)
|
|
replies[i].HistoryRemove, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Presence != nil {
|
|
method = "presence"
|
|
res := h.Presence(ctx, cmd.Presence)
|
|
replies[i].Presence, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.PresenceStats != nil {
|
|
method = "presence_stats"
|
|
res := h.PresenceStats(ctx, cmd.PresenceStats)
|
|
replies[i].PresenceStats, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Info != nil {
|
|
method = "info"
|
|
res := h.Info(ctx, cmd.Info)
|
|
replies[i].Info, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Rpc != nil {
|
|
method = "rpc"
|
|
res := h.RPC(ctx, cmd.Rpc)
|
|
replies[i].Rpc, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Refresh != nil {
|
|
method = "refresh"
|
|
res := h.Refresh(ctx, cmd.Refresh)
|
|
replies[i].Refresh, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.Channels != nil {
|
|
method = "channels"
|
|
res := h.Channels(ctx, cmd.Channels)
|
|
replies[i].Channels, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapPublish != nil {
|
|
method = "map_publish"
|
|
res := h.MapPublish(ctx, cmd.MapPublish)
|
|
replies[i].MapPublish, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapRemove != nil {
|
|
method = "map_remove"
|
|
res := h.MapRemove(ctx, cmd.MapRemove)
|
|
replies[i].MapRemove, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapReadState != nil {
|
|
method = "map_read_state"
|
|
res := h.MapReadState(ctx, cmd.MapReadState)
|
|
replies[i].MapReadState, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapReadStream != nil {
|
|
method = "map_read_stream"
|
|
res := h.MapReadStream(ctx, cmd.MapReadStream)
|
|
replies[i].MapReadStream, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapStats != nil {
|
|
method = "map_stats"
|
|
res := h.MapStats(ctx, cmd.MapStats)
|
|
replies[i].MapStats, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.MapClear != nil {
|
|
method = "map_clear"
|
|
res := h.MapClear(ctx, cmd.MapClear)
|
|
replies[i].MapClear, replies[i].Error = res.Result, res.Error
|
|
} else if cmd.SharedPollPublish != nil {
|
|
method = "shared_poll_publish"
|
|
res := h.SharedPollPublish(ctx, cmd.SharedPollPublish)
|
|
replies[i].SharedPollPublish, replies[i].Error = res.Result, res.Error
|
|
} else {
|
|
method = "unknown"
|
|
replies[i].Error = ErrorNotFound
|
|
}
|
|
if replies[i].Error != nil {
|
|
metrics.IncAPIError(h.config.Protocol, method, replies[i].Error.Code)
|
|
}
|
|
}
|
|
|
|
// batchRequestMaxConcurrency is applied for the parallel batch request.
|
|
const batchRequestMaxConcurrency = 1024
|
|
|
|
func (h *Executor) Batch(ctx context.Context, req *BatchRequest) *BatchResponse {
|
|
replies := make([]*Reply, len(req.Commands))
|
|
|
|
var sem chan struct{}
|
|
if req.Parallel {
|
|
sem = make(chan struct{}, batchRequestMaxConcurrency)
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
for i, cmd := range req.Commands {
|
|
replies[i] = new(Reply)
|
|
if !req.Parallel {
|
|
h.processCmd(ctx, cmd, i, replies)
|
|
} else {
|
|
sem <- struct{}{}
|
|
wg.Add(1)
|
|
go func(i int, cmd *Command) {
|
|
defer func() { <-sem }()
|
|
defer wg.Done()
|
|
h.processCmd(ctx, cmd, i, replies)
|
|
}(i, cmd)
|
|
}
|
|
}
|
|
wg.Wait()
|
|
|
|
return &BatchResponse{Replies: replies}
|
|
}
|
|
|
|
// Publish publishes data into channel.
|
|
func (h *Executor) Publish(ctx context.Context, cmd *PublishRequest) *PublishResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "publish")
|
|
|
|
ch := cmd.Channel
|
|
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &PublishResponse{}
|
|
|
|
if ch == "" {
|
|
log.Error().Err(errors.New("channel required for publish")).Msg("bad publish request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
var data []byte
|
|
if cmd.B64Data != "" {
|
|
byteInfo, err := base64.StdEncoding.DecodeString(cmd.B64Data)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
data = byteInfo
|
|
} else {
|
|
data = cmd.Data
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
// Stream-only API. Map channels must use map_publish; shared poll channels
|
|
// must use shared_poll_publish. Empty subscription_type defaults to stream.
|
|
switch chOpts.SubscriptionType {
|
|
case "", "stream":
|
|
// ok
|
|
default:
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("publish called on non-stream namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
// Data format validation
|
|
if err := config.ValidatePublicationData(data, chOpts.PublicationDataFormat); err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("bad publish request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
historySize := chOpts.HistorySize
|
|
historyTTL := chOpts.HistoryTTL
|
|
historyMetaTTL := chOpts.HistoryMetaTTL
|
|
if cmd.SkipHistory {
|
|
historySize = 0
|
|
historyTTL = configtypes.Duration(0)
|
|
}
|
|
|
|
delta := cmd.Delta
|
|
if chOpts.DeltaPublish {
|
|
delta = true
|
|
}
|
|
|
|
result, err := h.node.Publish(
|
|
cmd.Channel, data,
|
|
centrifuge.WithHistory(historySize, historyTTL.ToDuration(), historyMetaTTL.ToDuration()),
|
|
centrifuge.WithTags(cmd.GetTags()),
|
|
centrifuge.WithIdempotencyKey(cmd.GetIdempotencyKey()),
|
|
centrifuge.WithDelta(delta),
|
|
centrifuge.WithVersion(cmd.Version, cmd.VersionEpoch),
|
|
)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", cmd.Channel).Msg("error publishing data to channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &PublishResult{
|
|
Offset: result.StreamPosition.Offset,
|
|
Epoch: result.StreamPosition.Epoch,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
const broadcastRequestMaxConcurrency = 1024
|
|
|
|
// Broadcast publishes the same data into many channels.
|
|
func (h *Executor) Broadcast(ctx context.Context, cmd *BroadcastRequest) *BroadcastResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "broadcast")
|
|
|
|
resp := &BroadcastResponse{}
|
|
|
|
channels := cmd.Channels
|
|
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.Int("centrifugo.num_channels", len(channels)))
|
|
}
|
|
|
|
if len(channels) == 0 {
|
|
log.Error().Err(errors.New("channels required for broadcast")).Msg("bad broadcast request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
var data []byte
|
|
if cmd.B64Data != "" {
|
|
byteInfo, err := base64.StdEncoding.DecodeString(cmd.B64Data)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
data = byteInfo
|
|
} else {
|
|
data = cmd.Data
|
|
}
|
|
|
|
sem := make(chan struct{}, broadcastRequestMaxConcurrency)
|
|
|
|
responses := make([]*PublishResponse, len(channels))
|
|
var wg sync.WaitGroup
|
|
wg.Add(len(channels))
|
|
for i, ch := range channels {
|
|
sem <- struct{}{}
|
|
go func(i int, ch string) {
|
|
defer func() { <-sem }()
|
|
defer wg.Done()
|
|
if ch == "" {
|
|
respError := ErrorBadRequest
|
|
metrics.IncAPIError(h.config.Protocol, "broadcast_publish", respError.Code)
|
|
log.Error().Err(errors.New("channel can not be blank in broadcast")).Msg("bad broadcast request")
|
|
responses[i] = &PublishResponse{Error: respError}
|
|
return
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
respError := ErrorInternal
|
|
metrics.IncAPIError(h.config.Protocol, "broadcast_publish", respError.Code)
|
|
log.Error().Err(err).Str("channel", ch).Msg("error getting options for channel")
|
|
responses[i] = &PublishResponse{Error: respError}
|
|
return
|
|
}
|
|
if !found {
|
|
respError := ErrorUnknownChannel
|
|
metrics.IncAPIError(h.config.Protocol, "broadcast_publish", respError.Code)
|
|
log.Error().Err(errors.New("channel not found")).Str("channel", ch).Msg("error getting options for channel")
|
|
responses[i] = &PublishResponse{Error: respError}
|
|
return
|
|
}
|
|
|
|
// Data format validation
|
|
if err := config.ValidatePublicationData(data, chOpts.PublicationDataFormat); err != nil {
|
|
respError := ErrorBadRequest
|
|
metrics.IncAPIError(h.config.Protocol, "broadcast_publish", respError.Code)
|
|
log.Error().Err(err).Str("channel", ch).Msg("bad broadcast request")
|
|
responses[i] = &PublishResponse{Error: respError}
|
|
return
|
|
}
|
|
|
|
historySize := chOpts.HistorySize
|
|
historyTTL := chOpts.HistoryTTL
|
|
historyMetaTTL := chOpts.HistoryMetaTTL
|
|
if cmd.SkipHistory {
|
|
historySize = 0
|
|
historyTTL = configtypes.Duration(0)
|
|
}
|
|
|
|
delta := cmd.Delta
|
|
if chOpts.DeltaPublish {
|
|
delta = true
|
|
}
|
|
|
|
result, err := h.node.Publish(
|
|
ch, data,
|
|
centrifuge.WithHistory(historySize, historyTTL.ToDuration(), historyMetaTTL.ToDuration()),
|
|
centrifuge.WithTags(cmd.GetTags()),
|
|
centrifuge.WithIdempotencyKey(cmd.GetIdempotencyKey()),
|
|
centrifuge.WithDelta(delta),
|
|
centrifuge.WithVersion(cmd.Version, cmd.VersionEpoch),
|
|
)
|
|
resp := &PublishResponse{}
|
|
if err == nil {
|
|
resp.Result = &PublishResult{
|
|
Offset: result.StreamPosition.Offset,
|
|
Epoch: result.StreamPosition.Epoch,
|
|
}
|
|
} else {
|
|
respError := ErrorInternal
|
|
metrics.IncAPIError(h.config.Protocol, "broadcast_publish", respError.Code)
|
|
log.Error().Err(err).Str("channel", ch).Msg("error publishing data to channel during broadcast")
|
|
resp.Error = respError
|
|
}
|
|
responses[i] = resp
|
|
}(i, ch)
|
|
}
|
|
wg.Wait()
|
|
resp.Result = &BroadcastResult{Responses: responses}
|
|
return resp
|
|
}
|
|
|
|
// Subscribe subscribes user to a channel and sends subscribe
|
|
// control message to other nodes, so they could also subscribe user.
|
|
func (h *Executor) Subscribe(_ context.Context, cmd *SubscribeRequest) *SubscribeResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "subscribe")
|
|
|
|
resp := &SubscribeResponse{}
|
|
|
|
user := cmd.User
|
|
channel := cmd.Channel
|
|
|
|
if channel == "" {
|
|
log.Error().Err(errors.New("channel required for subscribe")).Msg("bad subscribe request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(channel)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
|
|
presence := chOpts.Presence
|
|
if cmd.Override != nil && cmd.Override.Presence != nil {
|
|
presence = cmd.Override.Presence.Value
|
|
}
|
|
joinLeave := chOpts.JoinLeave
|
|
if cmd.Override != nil && cmd.Override.JoinLeave != nil {
|
|
joinLeave = cmd.Override.JoinLeave.Value
|
|
}
|
|
pushJoinLeave := chOpts.ForcePushJoinLeave
|
|
if cmd.Override != nil && cmd.Override.ForcePushJoinLeave != nil {
|
|
pushJoinLeave = cmd.Override.ForcePushJoinLeave.Value
|
|
}
|
|
useRecover := chOpts.ForceRecovery
|
|
if cmd.Override != nil && cmd.Override.ForceRecovery != nil {
|
|
useRecover = cmd.Override.ForceRecovery.Value
|
|
}
|
|
position := chOpts.ForcePositioning
|
|
if cmd.Override != nil && cmd.Override.ForcePositioning != nil {
|
|
position = cmd.Override.ForcePositioning.Value
|
|
}
|
|
|
|
var recoverSince *centrifuge.StreamPosition
|
|
if cmd.RecoverSince != nil {
|
|
recoverSince = ¢rifuge.StreamPosition{
|
|
Offset: cmd.RecoverSince.Offset,
|
|
Epoch: cmd.RecoverSince.Epoch,
|
|
}
|
|
}
|
|
|
|
// Prefer the base64 fields when set (they exist for binary payloads that
|
|
// can't be carried in a JSON bytes field), matching publish/broadcast. Without
|
|
// this, b64data/b64info were silently dropped and the subscription got empty
|
|
// data/info.
|
|
data := cmd.Data
|
|
if cmd.B64Data != "" {
|
|
decoded, err := base64.StdEncoding.DecodeString(cmd.B64Data)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
data = decoded
|
|
}
|
|
info := cmd.Info
|
|
if cmd.B64Info != "" {
|
|
decoded, err := base64.StdEncoding.DecodeString(cmd.B64Info)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
info = decoded
|
|
}
|
|
|
|
err = h.node.Subscribe(user, channel,
|
|
centrifuge.WithSubscribeData(data),
|
|
centrifuge.WithSubscribeClient(cmd.Client),
|
|
centrifuge.WithSubscribeSession(cmd.Session),
|
|
centrifuge.WithChannelInfo(info),
|
|
centrifuge.WithExpireAt(cmd.ExpireAt),
|
|
centrifuge.WithEmitJoinLeave(joinLeave),
|
|
centrifuge.WithPushJoinLeave(pushJoinLeave),
|
|
centrifuge.WithRecovery(useRecover),
|
|
centrifuge.WithPositioning(position),
|
|
centrifuge.WithEmitPresence(presence),
|
|
centrifuge.WithRecoverSince(recoverSince),
|
|
centrifuge.WithSubscribeSource(subsource.ServerAPI),
|
|
centrifuge.WithSubscribeHistoryMetaTTL(chOpts.HistoryMetaTTL.ToDuration()),
|
|
)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", channel).Str("user", user).Msg("error subscribing user to channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &SubscribeResult{}
|
|
return resp
|
|
}
|
|
|
|
// Unsubscribe unsubscribes user from channel and sends unsubscribe
|
|
// control message to other nodes, so they could also unsubscribe user.
|
|
func (h *Executor) Unsubscribe(_ context.Context, cmd *UnsubscribeRequest) *UnsubscribeResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "unsubscribe")
|
|
|
|
resp := &UnsubscribeResponse{}
|
|
|
|
user := cmd.User
|
|
channel := cmd.Channel
|
|
|
|
if channel != "" {
|
|
_, _, _, found, err := h.cfgContainer.ChannelOptions(channel)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
}
|
|
|
|
err := h.node.Unsubscribe(user, channel, centrifuge.WithUnsubscribeClient(cmd.Client), centrifuge.WithUnsubscribeSession(cmd.Session))
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", channel).Str("user", user).Msg("error unsubscribing user from channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &UnsubscribeResult{}
|
|
return resp
|
|
}
|
|
|
|
// Disconnect disconnects user by its ID and sends disconnect
|
|
// control message to other nodes, so they could also disconnect user.
|
|
func (h *Executor) Disconnect(_ context.Context, cmd *DisconnectRequest) *DisconnectResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "disconnect")
|
|
|
|
resp := &DisconnectResponse{}
|
|
|
|
user := cmd.User
|
|
|
|
disconnect := centrifuge.DisconnectForceNoReconnect
|
|
if cmd.Disconnect != nil {
|
|
disconnect = centrifuge.Disconnect{
|
|
Code: cmd.Disconnect.Code,
|
|
Reason: cmd.Disconnect.Reason,
|
|
}
|
|
}
|
|
|
|
err := h.node.Disconnect(
|
|
user,
|
|
centrifuge.WithCustomDisconnect(disconnect),
|
|
centrifuge.WithDisconnectClient(cmd.Client),
|
|
centrifuge.WithDisconnectSession(cmd.Session),
|
|
centrifuge.WithDisconnectClientWhitelist(cmd.Whitelist))
|
|
if err != nil {
|
|
log.Error().Err(err).Str("user", user).Msg("error disconnecting user")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &DisconnectResult{}
|
|
return resp
|
|
}
|
|
|
|
// Refresh user connection by its ID.
|
|
func (h *Executor) Refresh(_ context.Context, cmd *RefreshRequest) *RefreshResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "refresh")
|
|
|
|
resp := &RefreshResponse{}
|
|
user := cmd.User
|
|
|
|
err := h.node.Refresh(
|
|
user,
|
|
centrifuge.WithRefreshClient(cmd.Client),
|
|
centrifuge.WithRefreshSession(cmd.Session),
|
|
centrifuge.WithRefreshExpired(cmd.Expired),
|
|
centrifuge.WithRefreshExpireAt(cmd.ExpireAt),
|
|
centrifuge.WithRefreshInfo(cmd.Info),
|
|
)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("user", user).Msg("error refreshing user")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &RefreshResult{}
|
|
return resp
|
|
}
|
|
|
|
// Presence returns response with presence information for channel.
|
|
func (h *Executor) Presence(_ context.Context, cmd *PresenceRequest) *PresenceResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "presence")
|
|
|
|
resp := &PresenceResponse{}
|
|
|
|
ch := cmd.Channel
|
|
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
|
|
if !chOpts.Presence {
|
|
log.Warn().Str("channel", ch).Msg("presence is not enabled for channel in Centrifugo configuration")
|
|
resp.Error = ErrorNotAvailable
|
|
return resp
|
|
}
|
|
|
|
presence, err := h.node.Presence(ch)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error getting presence for channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
apiPresence := make(map[string]*ClientInfo, len(presence.Presence))
|
|
for k, v := range presence.Presence {
|
|
apiPresence[k] = &ClientInfo{
|
|
User: v.UserID,
|
|
Client: v.ClientID,
|
|
ConnInfo: v.ConnInfo,
|
|
ChanInfo: v.ChanInfo,
|
|
}
|
|
}
|
|
|
|
resp.Result = &PresenceResult{
|
|
Presence: apiPresence,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// PresenceStats returns response with presence stats information for channel.
|
|
func (h *Executor) PresenceStats(_ context.Context, cmd *PresenceStatsRequest) *PresenceStatsResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "presence_stats")
|
|
|
|
resp := &PresenceStatsResponse{}
|
|
|
|
ch := cmd.Channel
|
|
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
|
|
if !chOpts.Presence {
|
|
log.Warn().Str("channel", ch).Msg("presence is not enabled for channel in Centrifugo configuration")
|
|
resp.Error = ErrorNotAvailable
|
|
return resp
|
|
}
|
|
|
|
stats, err := h.node.PresenceStats(cmd.Channel)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error getting presence stats for channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
resp.Result = &PresenceStatsResult{
|
|
NumClients: uint32(stats.NumClients),
|
|
NumUsers: uint32(stats.NumUsers),
|
|
}
|
|
|
|
return resp
|
|
}
|
|
|
|
// History returns response with history information for channel.
|
|
func (h *Executor) History(_ context.Context, cmd *HistoryRequest) *HistoryResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "history")
|
|
|
|
resp := &HistoryResponse{}
|
|
|
|
ch := cmd.Channel
|
|
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
|
|
if chOpts.HistorySize <= 0 || chOpts.HistoryTTL <= 0 {
|
|
log.Warn().Str("channel", ch).Msg("history is not enabled for channel in Centrifugo configuration (history_size and history_ttl must both be > 0)")
|
|
resp.Error = ErrorNotAvailable
|
|
return resp
|
|
}
|
|
|
|
var sp *centrifuge.StreamPosition
|
|
if cmd.Since != nil {
|
|
sp = ¢rifuge.StreamPosition{
|
|
Epoch: cmd.Since.Epoch,
|
|
Offset: cmd.Since.Offset,
|
|
}
|
|
}
|
|
|
|
historyMetaTTL := chOpts.HistoryMetaTTL
|
|
|
|
history, err := h.node.History(
|
|
ch,
|
|
centrifuge.WithHistoryMetaTTL(historyMetaTTL.ToDuration()),
|
|
centrifuge.WithLimit(int(cmd.Limit)),
|
|
centrifuge.WithSince(sp),
|
|
centrifuge.WithReverse(cmd.Reverse),
|
|
)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error getting history for channel")
|
|
if errors.Is(err, centrifuge.ErrorUnrecoverablePosition) {
|
|
resp.Error = ErrorUnrecoverablePosition
|
|
return resp
|
|
}
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
apiPubs := make([]*Publication, len(history.Publications))
|
|
|
|
for i, pub := range history.Publications {
|
|
apiPub := &Publication{
|
|
Data: Raw(pub.Data),
|
|
Offset: pub.Offset,
|
|
Tags: pub.Tags,
|
|
}
|
|
if pub.Info != nil {
|
|
apiPub.Info = &ClientInfo{
|
|
User: pub.Info.UserID,
|
|
Client: pub.Info.ClientID,
|
|
ConnInfo: pub.Info.ConnInfo,
|
|
ChanInfo: pub.Info.ChanInfo,
|
|
}
|
|
}
|
|
apiPubs[i] = apiPub
|
|
}
|
|
|
|
resp.Result = &HistoryResult{
|
|
Publications: apiPubs,
|
|
Offset: history.Offset,
|
|
Epoch: history.Epoch,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// HistoryRemove removes all history information for channel.
|
|
func (h *Executor) HistoryRemove(_ context.Context, cmd *HistoryRemoveRequest) *HistoryRemoveResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "history_remove")
|
|
|
|
resp := &HistoryRemoveResponse{}
|
|
|
|
ch := cmd.Channel
|
|
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
|
|
if chOpts.HistorySize <= 0 || chOpts.HistoryTTL <= 0 {
|
|
log.Warn().Str("channel", ch).Msg("history is not enabled for channel in Centrifugo configuration (history_size and history_ttl must both be > 0)")
|
|
resp.Error = ErrorNotAvailable
|
|
return resp
|
|
}
|
|
|
|
err = h.node.RemoveHistory(ch)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error removing history for channel")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &HistoryRemoveResult{}
|
|
return resp
|
|
}
|
|
|
|
// Info returns information about running nodes.
|
|
func (h *Executor) Info(_ context.Context, _ *InfoRequest) *InfoResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "info")
|
|
|
|
resp := &InfoResponse{}
|
|
|
|
info, err := h.node.Info()
|
|
if err != nil {
|
|
log.Error().Err(err).Msg("error calling info")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
nodes := make([]*NodeResult, len(info.Nodes))
|
|
for i, node := range info.Nodes {
|
|
res := &NodeResult{
|
|
Uid: node.UID,
|
|
Version: node.Version,
|
|
Name: node.Name,
|
|
NumClients: node.NumClients,
|
|
NumUsers: node.NumUsers,
|
|
NumSubs: node.NumSubs,
|
|
NumChannels: node.NumChannels,
|
|
Uptime: node.Uptime,
|
|
Process: nil,
|
|
}
|
|
if node.Metrics != nil {
|
|
res.Metrics = &Metrics{
|
|
Interval: node.Metrics.Interval,
|
|
Items: node.Metrics.Items,
|
|
}
|
|
}
|
|
nodes[i] = res
|
|
}
|
|
resp.Result = &InfoResult{
|
|
Nodes: nodes,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// RPC can call arbitrary methods.
|
|
func (h *Executor) RPC(ctx context.Context, cmd *RPCRequest) *RPCResponse {
|
|
started := time.Now()
|
|
defer metrics.ObserveAPICommand(started, h.config.Protocol, "rpc")
|
|
|
|
resp := &RPCResponse{}
|
|
|
|
if cmd.Method == "" {
|
|
log.Error().Err(errors.New("rpc method required")).Msg("bad rpc request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
handler, ok := h.rpcExtension[cmd.Method]
|
|
if !ok {
|
|
resp.Error = ErrorNotFound
|
|
return resp
|
|
}
|
|
|
|
defer metrics.ObserveRPC(started, h.config.Protocol, cmd.Method)
|
|
|
|
data, err := handler(ctx, cmd.Params)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("method", cmd.Method).Msg("error calling rpc method")
|
|
resp.Error = toAPIErr(err)
|
|
return resp
|
|
}
|
|
|
|
resp.Result = &RPCResult{
|
|
Data: data,
|
|
}
|
|
|
|
return resp
|
|
}
|
|
|
|
// Channels in the system.
|
|
func (h *Executor) Channels(ctx context.Context, cmd *ChannelsRequest) *ChannelsResponse {
|
|
started := time.Now()
|
|
defer metrics.ObserveAPICommand(started, h.config.Protocol, "channels")
|
|
|
|
resp := &ChannelsResponse{}
|
|
|
|
channels, err := h.surveyCaller.Channels(ctx, cmd)
|
|
if err != nil {
|
|
log.Error().Err(err).Msg("error calling channels")
|
|
resp.Error = toAPIErr(err)
|
|
return resp
|
|
}
|
|
|
|
resp.Result = &ChannelsResult{
|
|
Channels: channels,
|
|
}
|
|
|
|
return resp
|
|
}
|
|
|
|
// MapPublish publishes data to a map channel key.
|
|
func (h *Executor) MapPublish(ctx context.Context, cmd *MapPublishRequest) *MapPublishResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_publish")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapPublishResponse{}
|
|
if ch == "" || cmd.Key == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_publish called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
var data []byte
|
|
if cmd.B64Data != "" {
|
|
byteInfo, err := base64.StdEncoding.DecodeString(cmd.B64Data)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
data = byteInfo
|
|
} else {
|
|
data = cmd.Data
|
|
}
|
|
|
|
if err := config.ValidatePublicationData(data, chOpts.PublicationDataFormat); err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("bad map publish request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
delta := cmd.Delta
|
|
if chOpts.DeltaPublish {
|
|
delta = true
|
|
}
|
|
|
|
opts := centrifuge.MapPublishOptions{
|
|
Data: data,
|
|
Tags: cmd.GetTags(),
|
|
IdempotencyKey: cmd.GetIdempotencyKey(),
|
|
UseDelta: delta,
|
|
}
|
|
opts.Version = cmd.Version
|
|
opts.VersionEpoch = cmd.VersionEpoch
|
|
if cmd.KeyMode != "" {
|
|
switch cmd.KeyMode {
|
|
case "if_new", "if_exists":
|
|
opts.KeyMode = centrifuge.KeyMode(cmd.KeyMode)
|
|
default:
|
|
log.Warn().Str("channel", ch).Str("key_mode", cmd.KeyMode).Msg("map_publish: invalid key_mode")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
}
|
|
|
|
result, err := h.node.MapPublish(ctx, ch, cmd.Key, opts)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map publish")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &MapPublishResult{
|
|
Offset: result.Position.Offset,
|
|
Epoch: result.Position.Epoch,
|
|
Suppressed: result.Suppressed,
|
|
SuppressReason: string(result.SuppressReason),
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// MapRemove removes a key from a map channel.
|
|
func (h *Executor) MapRemove(ctx context.Context, cmd *MapRemoveRequest) *MapRemoveResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_remove")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapRemoveResponse{}
|
|
if ch == "" || cmd.Key == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_remove called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
opts := centrifuge.MapRemoveOptions{
|
|
IdempotencyKey: cmd.GetIdempotencyKey(),
|
|
}
|
|
|
|
result, err := h.node.MapRemove(ctx, ch, cmd.Key, opts)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map remove")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &MapRemoveResult{
|
|
Offset: result.Position.Offset,
|
|
Epoch: result.Position.Epoch,
|
|
Suppressed: result.Suppressed,
|
|
SuppressReason: string(result.SuppressReason),
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// MapReadState reads the current state of a map channel.
|
|
func (h *Executor) MapReadState(ctx context.Context, cmd *MapReadStateRequest) *MapReadStateResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_read_state")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapReadStateResponse{}
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_read_state called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
opts := centrifuge.MapReadStateOptions{
|
|
Cursor: cmd.Cursor,
|
|
Limit: int(cmd.Limit),
|
|
Key: cmd.Key,
|
|
Asc: cmd.Asc,
|
|
}
|
|
if cmd.RevisionOffset > 0 || cmd.RevisionEpoch != "" {
|
|
opts.Revision = ¢rifuge.StreamPosition{Offset: cmd.RevisionOffset, Epoch: cmd.RevisionEpoch}
|
|
}
|
|
|
|
result, err := h.node.MapStateRead(ctx, ch, opts)
|
|
if err != nil {
|
|
if errors.Is(err, centrifuge.ErrorUnrecoverablePosition) {
|
|
resp.Error = ErrorUnrecoverablePosition
|
|
return resp
|
|
}
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map read state")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
entries := make([]*MapEntry, 0, len(result.Publications))
|
|
for _, pub := range result.Publications {
|
|
entries = append(entries, &MapEntry{
|
|
Key: pub.Key,
|
|
Data: pub.Data,
|
|
Tags: pub.Tags,
|
|
Offset: pub.Offset,
|
|
Score: pub.Score,
|
|
Removed: pub.Removed,
|
|
Time: pub.Time,
|
|
})
|
|
}
|
|
|
|
resp.Result = &MapReadStateResult{
|
|
Entries: entries,
|
|
Offset: result.Position.Offset,
|
|
Epoch: result.Position.Epoch,
|
|
Cursor: result.Cursor,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// MapReadStream reads the stream of a map channel.
|
|
func (h *Executor) MapReadStream(ctx context.Context, cmd *MapReadStreamRequest) *MapReadStreamResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_read_stream")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapReadStreamResponse{}
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_read_stream called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
opts := centrifuge.MapReadStreamOptions{
|
|
Filter: centrifuge.StreamFilter{
|
|
Limit: int(cmd.Limit),
|
|
Reverse: cmd.Reverse,
|
|
},
|
|
}
|
|
if cmd.SinceOffset > 0 || cmd.SinceEpoch != "" {
|
|
opts.Filter.Since = ¢rifuge.StreamPosition{Offset: cmd.SinceOffset, Epoch: cmd.SinceEpoch}
|
|
}
|
|
|
|
result, err := h.node.MapStreamRead(ctx, ch, opts)
|
|
if err != nil {
|
|
if errors.Is(err, centrifuge.ErrorUnrecoverablePosition) {
|
|
resp.Error = ErrorUnrecoverablePosition
|
|
return resp
|
|
}
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map read stream")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
|
|
entries := make([]*MapEntry, 0, len(result.Publications))
|
|
for _, pub := range result.Publications {
|
|
entries = append(entries, &MapEntry{
|
|
Key: pub.Key,
|
|
Data: pub.Data,
|
|
Tags: pub.Tags,
|
|
Offset: pub.Offset,
|
|
Score: pub.Score,
|
|
Removed: pub.Removed,
|
|
Time: pub.Time,
|
|
})
|
|
}
|
|
|
|
resp.Result = &MapReadStreamResult{
|
|
Entries: entries,
|
|
Offset: result.Position.Offset,
|
|
Epoch: result.Position.Epoch,
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// MapStats returns stats for a map channel.
|
|
func (h *Executor) MapStats(ctx context.Context, cmd *MapStatsRequest) *MapStatsResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_stats")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapStatsResponse{}
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_stats called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
result, err := h.node.MapStats(ctx, ch)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map stats")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &MapStatsResult{
|
|
NumKeys: int32(result.NumKeys),
|
|
}
|
|
return resp
|
|
}
|
|
|
|
// MapClear removes all data from a map channel.
|
|
func (h *Executor) MapClear(ctx context.Context, cmd *MapClearRequest) *MapClearResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "map_clear")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &MapClearResponse{}
|
|
if ch == "" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "map" && chOpts.SubscriptionType != "map_clients" && chOpts.SubscriptionType != "map_users" {
|
|
log.Warn().Str("channel", ch).Str("subscription_type", chOpts.SubscriptionType).Msg("map_clear called on non-map namespace")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
err = h.node.MapClear(ctx, ch, centrifuge.MapClearOptions{})
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in map clear")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &MapClearResult{}
|
|
return resp
|
|
}
|
|
|
|
func (h *Executor) SharedPollPublish(ctx context.Context, cmd *SharedPollPublishRequest) *SharedPollPublishResponse {
|
|
defer metrics.ObserveAPICommand(time.Now(), h.config.Protocol, "shared_poll_publish")
|
|
|
|
ch := cmd.Channel
|
|
if h.config.UseOpenTelemetry {
|
|
span := trace.SpanFromContext(ctx)
|
|
span.SetAttributes(attribute.String("centrifugo.channel", ch))
|
|
}
|
|
|
|
resp := &SharedPollPublishResponse{}
|
|
if ch == "" || cmd.Key == "" || cmd.Version == 0 {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
_, _, chOpts, found, err := h.cfgContainer.ChannelOptions(ch)
|
|
if err != nil {
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
if !found {
|
|
resp.Error = ErrorUnknownChannel
|
|
return resp
|
|
}
|
|
if chOpts.SubscriptionType != "shared_poll" {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
if !chOpts.SharedPoll.PublishEnabled {
|
|
log.Warn().Str("channel", ch).Msg("shared_poll publish_enabled is not enabled for channel in Centrifugo configuration")
|
|
resp.Error = ErrorNotAvailable
|
|
return resp
|
|
}
|
|
|
|
var data []byte
|
|
if cmd.B64Data != "" {
|
|
byteInfo, err := base64.StdEncoding.DecodeString(cmd.B64Data)
|
|
if err != nil {
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
data = byteInfo
|
|
} else {
|
|
data = cmd.Data
|
|
}
|
|
|
|
if err := config.ValidatePublicationData(data, chOpts.PublicationDataFormat); err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("bad shared poll publish request")
|
|
resp.Error = ErrorBadRequest
|
|
return resp
|
|
}
|
|
|
|
err = h.node.SharedPollPublish(ctx, ch, cmd.Key, cmd.Version, cmd.Epoch, data)
|
|
if err != nil {
|
|
log.Error().Err(err).Str("channel", ch).Msg("error in shared poll publish")
|
|
resp.Error = ErrorInternal
|
|
return resp
|
|
}
|
|
resp.Result = &SharedPollPublishResult{}
|
|
return resp
|
|
}
|
|
|
|
func toAPIErr(err error) *Error {
|
|
var apiErr *Error
|
|
if errors.As(err, &apiErr) {
|
|
return apiErr
|
|
}
|
|
return ErrorInternal
|
|
}
|