func (qc *QueueClient) checkAndEnforceStreamQueueMessageRequest(req *pb.StreamQueueMessagesRequest) error {
	if req.RequestID == "" {
		req.RequestID = nuid.Next()
	}
	if req.ClientID == "" {
		return entities.ErrInvalidClientID
	}
	switch req.StreamRequestTypeData {
	case pb.StreamRequestType_ReceiveMessage:
		if err := qc.checkChannelNameValidity(req.Channel); err != nil {
			return err
		}
		if req.VisibilitySeconds > qc.policyCfg.MaxVisibilitySeconds || req.VisibilitySeconds < 0 {
			return entities.ErrInvalidVisibility
		}
		if req.VisibilitySeconds == 0 {
			req.VisibilitySeconds = qc.policyCfg.DefaultVisibilitySeconds
		}
		if req.WaitTimeSeconds > qc.policyCfg.MaxWaitTimeoutSeconds || req.WaitTimeSeconds < 0 {
			return entities.ErrInvalidWaitTimeout
		}
		if req.WaitTimeSeconds == 0 {
			req.WaitTimeSeconds = qc.policyCfg.DefaultWaitTimeoutSeconds
		}
	case pb.StreamRequestType_AckMessage, pb.StreamRequestType_RejectMessage:
		if req.RefSequence == 0 {
			return entities.ErrInvalidAckSeq
		}
	case pb.StreamRequestType_ResendMessage:
		if err := qc.checkChannelNameValidity(req.Channel); err != nil {
			return err
		}
	case pb.StreamRequestType_ModifyVisibility:
		if req.VisibilitySeconds > qc.policyCfg.MaxVisibilitySeconds || req.VisibilitySeconds < 0 {
			return entities.ErrInvalidVisibility
		}
		if req.VisibilitySeconds == 0 {
			req.VisibilitySeconds = qc.policyCfg.DefaultVisibilitySeconds
		}
	case pb.StreamRequestType_SendModifiedMessage:
		if req.ModifiedMessage == nil {
			return entities.ErrInvalidQueueMessage
		}
		if req.ModifiedMessage.Attributes == nil {
			req.ModifiedMessage.Attributes = &pb.QueueMessageAttributes{
				Timestamp:         0,
				Sequence:          0,
				MD5OfBody:         "",
				ReceiveCount:      0,
				ReRouted:          false,
				ReRoutedFromQueue: "",
				ExpirationAt:      0,
				DelayedTo:         0,
			}
		}
		if req.ModifiedMessage.Policy == nil {
			req.ModifiedMessage.Policy = &pb.QueueMessagePolicy{
				ExpirationSeconds: 0,
				DelaySeconds:      0,
				MaxReceiveCount:   0,
				MaxReceiveQueue:   "",
			}
		}
		if err := ValidateQueueMessage(req.ModifiedMessage); err != nil {
			return err
		}
		if err := qc.checkAndEnforceMessagePolicy(req.ModifiedMessage.Policy); err != nil {
			return err
		}
		if req.VisibilitySeconds > qc.policyCfg.MaxVisibilitySeconds || req.VisibilitySeconds < 0 {
			return entities.ErrInvalidVisibility
		}
	default:
		return entities.ErrInvalidStreamRequestType
	}
	return nil
}
function to check and enforce receive queue message request
package client
import (
	"context"
	"fmt"
	"github.com/kubemq-io/kubemq-community/pkg/uuid"
	"github.com/nats-io/nuid"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/kubemq-io/kubemq-community/pkg/entities"
	pb "github.com/kubemq-io/protobuf/go"
	"github.com/kubemq-io/kubemq-community/config"
	"go.uber.org/atomic"
	nats "github.com/kubemq-io/broker/client/nats"
	stan "github.com/kubemq-io/broker/client/stan"
	"github.com/kubemq-io/kubemq-community/pkg/logging"
)
const (
	defaultWaitTimout        = 1
	defaultMaxAceMessageTime = 12 * time.Hour
)
type delayedQueueMessage struct {
	raw       *stan.Msg
	msg       *pb.QueueMessage
	delayedTo int64
	id        string
}
type byTime []*delayedQueueMessage
func (d byTime) Less(i, j int) bool {
	if d[i].delayedTo == d[j].delayedTo {
		return d[i].raw.Sequence < d[j].raw.Sequence
	}
	return d[i].delayedTo < d[j].delayedTo
}
func (d byTime) Swap(i, j int) {
	d[i], d[j] = d[j], d[i]
}
func (d byTime) Len() int {
	return len(d)
}
type QueueClient struct {
	mu     sync.Mutex
	logger *logging.Logger
	isUp                      *atomic.Bool
	baseConn                  *nats.Conn
	queueConn                 stan.Conn
	writeDeadline             time.Duration
	policyCfg                 *config.QueueConfig
	delayedMessages           sync.Map
	delayProcessorDone        chan bool
	queueStreamMiddlewareFunc func(qc *QueueClient, msg *pb.StreamQueueMessagesRequest) error
	queuePollMiddlewareFunc   func(qc *QueueClient, req *pb.QueuesDownstreamRequest) error
}
func NewQueueClient(opts *Options, policyCfg *config.QueueConfig) (*QueueClient, error) {
	if opts.ClientID == "" {
		return nil, entities.ErrInvalidClientID
	}
	qc := &QueueClient{
		logger:             logging.GetLogFactory().NewLogger("client-" + opts.ClientID),
		isUp:               atomic.NewBool(false),
		policyCfg:          policyCfg,
		delayProcessorDone: make(chan bool, 1),
	}
	if opts.Deadline == 0 {
		qc.writeDeadline = defaultDeadline
	} else {
		qc.writeDeadline = opts.Deadline
	}
	var err error
	qc.queueConn, err = qc.connect(opts)
	if err != nil {
		return nil, err
	}
	qc.isUp.Store(true)
	return qc, nil
}
func (qc *QueueClient) SetQueueStreamMiddlewareFunc(fn func(qc *QueueClient, msg *pb.StreamQueueMessagesRequest) error) {
	qc.queueStreamMiddlewareFunc = fn
}
func (qc *QueueClient) SetQueuesDownstreamMiddlewareFunc(fn func(qc *QueueClient, req *pb.QueuesDownstreamRequest) error) {
	qc.queuePollMiddlewareFunc = fn
}
func (qc *QueueClient) connect(opts *Options) (stan.Conn, error) {
	natsOpts := nats.Options{
		Url: fmt.Sprintf("nats://0.0.0.0:%d", 4224),
		DisconnectedErrCB: func(conn *nats.Conn, err error) {
			conn.Close()
		},
	}
	if opts.AutoReconnect {
		natsOpts.AllowReconnect = true
		natsOpts.MaxReconnect = -1
		natsOpts.ReconnectWait = 100 * time.Millisecond
	}
	if opts.MemoryPipe != nil {
		natsOpts.CustomDialer = opts.MemoryPipe
	} else {
		return nil, fmt.Errorf("no memory pipe exist")
	}
	conn, err := natsOpts.Connect()
	if err != nil {
		return nil, err
	}
	qc.baseConn = conn
	storeConn, err := stan.Connect("kubemq", opts.ClientID,
		stan.NatsConn(conn),
		stan.NatsURL(fmt.Sprintf("nats://0.0.0.0:%d", 4224)),
		stan.ConnectWait(storeConnectionTime),
		stan.SetConnectionLostHandler(func(conn stan.Conn, err error) {
			qc.logger.Errorf("queue client %s connection lost", opts.ClientID)
		}),
		stan.Pings(storePingInterval, storePingMaxOut),
		stan.MaxPubAcksInflight(4096),
		stan.PubAckWait(120*time.Second),
	)
	return storeConn, err
}
func (qc *QueueClient) Disconnect() error {
	if !qc.isUp.Load() {
		return nil
	}
	qc.isUp.Store(false)
	qc.mu.Lock()
	defer qc.mu.Unlock()
	err := qc.queueConn.Close()
	if err != nil {
		return err
	}
	qc.baseConn.Close()
	return nil
}
func (qc *QueueClient) checkChannelNameValidity(channel string) error {
	if channel == "" {
		return entities.ErrInvalidQueueName
	}
	if strings.HasSuffix(channel, ".") {
		return entities.ErrInvalidQueueName
	}
	if strings.Contains(channel, " ") {
		return entities.ErrInvalidQueueName
	}
	if strings.Contains(channel, "*") || strings.Contains(channel, ">") {
		return entities.ErrInvalidQueueName
	}
	return nil
}
func (qc *QueueClient) checkAndEnforceAckAllRequest(req *pb.AckAllQueueMessagesRequest) error {
	if req.ClientID == "" {
		return entities.ErrInvalidClientID
	}
	if err := qc.checkChannelNameValidity(req.Channel); err != nil {
		return err
	}
	if req.WaitTimeSeconds > qc.policyCfg.MaxWaitTimeoutSeconds || req.WaitTimeSeconds < 0 {
		return entities.ErrInvalidWaitTimeout
	}
	if req.WaitTimeSeconds == 0 {
		req.WaitTimeSeconds = qc.policyCfg.DefaultWaitTimeoutSeconds
	}
	return nil
}
func (qc *QueueClient) checkAndEnforceMessagePolicy(mp *pb.QueueMessagePolicy) error {
	if mp.ExpirationSeconds > qc.policyCfg.MaxExpirationSeconds || mp.ExpirationSeconds < 0 {
		return entities.ErrInvalidExpiration
	}
	if mp.MaxReceiveCount > qc.policyCfg.MaxReceiveCount || mp.MaxReceiveCount < 0 {
		return entities.ErrInvalidMaxReceiveCount
	}
	if mp.MaxReceiveCount == 0 {
		mp.MaxReceiveCount = qc.policyCfg.MaxReceiveCount
	}
	if mp.DelaySeconds > qc.policyCfg.MaxDelaySeconds || mp.DelaySeconds < 0 {
		return entities.ErrInvalidDelay
	}
	if mp.MaxReceiveQueue != "" {
		if err := qc.checkChannelNameValidity(mp.MaxReceiveQueue); err != nil {
			return err
		}
	}
	return nil
}
func (qc *QueueClient) checkAndEnforceReceiveQueueMessageRequest(req *pb.ReceiveQueueMessagesRequest) error {
	err := ValidateReceiveQueueMessageRequest(req)
	if err != nil {
		return err
	}
	if req.MaxNumberOfMessages > qc.policyCfg.MaxNumberOfMessages || req.MaxNumberOfMessages < 1 {
		return entities.ErrInvalidMaxMessages
	}
	if req.WaitTimeSeconds > qc.policyCfg.MaxWaitTimeoutSeconds || req.WaitTimeSeconds < 0 {
		return entities.ErrInvalidWaitTimeout
	}
	if req.WaitTimeSeconds == 0 {
		req.WaitTimeSeconds = qc.policyCfg.DefaultWaitTimeoutSeconds
	}
	return nil
}