func (h *Executor) Subscribe(_ context.Context, cmd *SubscribeRequest) *SubscribeResponse {
	defer observe(time.Now(), h.protocol, "subscribe")
	resp := &SubscribeResponse{}
	user := cmd.User
	channel := cmd.Channel
	if user == "" {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "user required for subscribe", map[string]interface{}{"channel": channel, "user": user}))
		resp.Error = ErrorBadRequest
		return resp
	}
	if channel == "" {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "channel required for subscribe", map[string]interface{}{"channel": channel, "user": user}))
		resp.Error = ErrorBadRequest
		return resp
	}
	chOpts, found, err := h.ruleContainer.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
	}
	useRecover := chOpts.Recover
	if cmd.Override != nil && cmd.Override.Recover != nil {
		useRecover = cmd.Override.Recover.Value
	}
	position := chOpts.Position
	if cmd.Override != nil && cmd.Override.Position != nil {
		position = cmd.Override.Position.Value
	}
	var recoverSince *centrifuge.StreamPosition
	if cmd.RecoverSince != nil {
		recoverSince = &centrifuge.StreamPosition{
			Offset: cmd.RecoverSince.Offset,
			Epoch:  cmd.RecoverSince.Epoch,
		}
	}
	err = h.node.Subscribe(user, channel,
		centrifuge.WithSubscribeData(cmd.Data),
		centrifuge.WithSubscribeClient(cmd.Client),
		centrifuge.WithChannelInfo(cmd.Info),
		centrifuge.WithExpireAt(cmd.ExpireAt),
		centrifuge.WithJoinLeave(joinLeave),
		centrifuge.WithRecover(useRecover),
		centrifuge.WithPosition(position),
		centrifuge.WithPresence(presence),
		centrifuge.WithRecoverSince(recoverSince),
	)
	if err != nil {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error subscribing user to a channel", map[string]interface{}{"channel": channel, "user": user, "error": err.Error()}))
		resp.Error = ErrorInternal
		return resp
	}
	return resp
}
Subscribe subscribes user to a channel and sends subscribe
control message to other nodes so they could also subscribe user.
package api
import (
	"context"
	"encoding/base64"
	"errors"
	"sync"
	"time"
	. "github.com/centrifugal/centrifugo/v3/internal/apiproto"
	"github.com/centrifugal/centrifugo/v3/internal/rule"
	"github.com/centrifugal/centrifuge"
)
type RPCHandler func(ctx context.Context, params Raw) (Raw, error)
type Executor struct {
	node          *centrifuge.Node
	ruleContainer *rule.Container
	protocol      string
	rpcExtension  map[string]RPCHandler
	surveyCaller  SurveyCaller
}
type SurveyCaller interface {
	Channels(ctx context.Context, cmd *ChannelsRequest) (map[string]*ChannelInfo, error)
}
func NewExecutor(n *centrifuge.Node, ruleContainer *rule.Container, surveyCaller SurveyCaller, protocol string) *Executor {
	e := &Executor{
		node:          n,
		ruleContainer: ruleContainer,
		protocol:      protocol,
		surveyCaller:  surveyCaller,
		rpcExtension:  make(map[string]RPCHandler),
	}
	return e
}
func (h Executor) SetRPCExtension(method string, handler RPCHandler) {
	h.rpcExtension[method] = handler
}
func (h *Executor) Publish(_ context.Context, cmd *PublishRequest) *PublishResponse {
	defer observe(time.Now(), h.protocol, "publish")
	ch := cmd.Channel
	resp := &PublishResponse{}
	if ch == "" {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "channel required for publish", nil))
		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 len(data) == 0 {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "data required for publish", nil))
		resp.Error = ErrorBadRequest
		return resp
	}
	chOpts, found, err := h.ruleContainer.ChannelOptions(ch)
	if err != nil {
		resp.Error = ErrorInternal
		return resp
	}
	if !found {
		resp.Error = ErrorUnknownChannel
		return resp
	}
	historySize := chOpts.HistorySize
	historyTTL := chOpts.HistoryTTL
	if cmd.SkipHistory {
		historySize = 0
		historyTTL = 0
	}
	result, err := h.node.Publish(
		cmd.Channel, cmd.Data,
		centrifuge.WithHistory(historySize, time.Duration(historyTTL)),
	)
	if err != nil {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error publishing message in engine", map[string]interface{}{"error": err.Error(), "channel": cmd.Channel}))
		resp.Error = ErrorInternal
		return resp
	}
	resp.Result = &PublishResult{
		Offset: result.StreamPosition.Offset,
		Epoch:  result.StreamPosition.Epoch,
	}
	return resp
}
func (h *Executor) Broadcast(_ context.Context, cmd *BroadcastRequest) *BroadcastResponse {
	defer observe(time.Now(), h.protocol, "broadcast")
	resp := &BroadcastResponse{}
	channels := cmd.Channels
	if len(channels) == 0 {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "channels required for broadcast", nil))
		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 len(data) == 0 {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "data required for broadcast", nil))
		resp.Error = ErrorBadRequest
		return resp
	}
	responses := make([]*PublishResponse, len(channels))
	var wg sync.WaitGroup
	wg.Add(len(channels))
	for i, ch := range channels {
		go func(i int, ch string) {
			defer wg.Done()
			if ch == "" {
				h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "channel can not be blank in broadcast", nil))
				responses[i] = &PublishResponse{Error: ErrorBadRequest}
				return
			}
			chOpts, found, err := h.ruleContainer.ChannelOptions(ch)
			if err != nil {
				h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error getting options for channel", map[string]interface{}{"channel": ch, "error": err.Error()}))
				responses[i] = &PublishResponse{Error: ErrorInternal}
				return
			}
			if !found {
				h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "can't find namespace for channel", map[string]interface{}{"channel": ch}))
				responses[i] = &PublishResponse{Error: ErrorUnknownChannel}
				return
			}
			historySize := chOpts.HistorySize
			historyTTL := chOpts.HistoryTTL
			if cmd.SkipHistory {
				historySize = 0
				historyTTL = 0
			}
			result, err := h.node.Publish(
				ch, data,
				centrifuge.WithHistory(historySize, time.Duration(historyTTL)),
			)
			resp := &PublishResponse{}
			if err == nil {
				resp.Result = &PublishResult{
					Offset: result.StreamPosition.Offset,
					Epoch:  result.StreamPosition.Epoch,
				}
			} else {
				h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error publishing data to channel", map[string]interface{}{"channel": ch, "error": err.Error()}))
				resp.Error = ErrorInternal
			}
			responses[i] = resp
		}(i, ch)
	}
	wg.Wait()
	resp.Result = &BroadcastResult{Responses: responses}
	return resp
}