golang

var (
	ErrNoAvailableEndpoints = errors.New("rocketmq: no available endpoints")
	EnableSsl               = true
)

var (
	PublishMLatencyMs         = stats.Int64("publish_latency", "Publish latency in milliseconds", "ms")
	ConsumeDeliveryMLatencyMs = stats.Int64("delivery_latency", "Time spent delivering messages from servers to clients", "ms")
	ConsumeAwaitMLatencyMs    = stats.Int64("await_time", "Client side queuing time of messages before getting processed", "ms")
	ConsumeProcessMLatencyMs  = stats.Int64("process_time", "Process message time", "ms")

	PublishLatencyView = view.View{
		Name:        "rocketmq_send_cost_time",
		Description: "Publish latency",
		Measure:     PublishMLatencyMs,
		Aggregation: view.Distribution(1, 5, 10, 20, 50, 200, 500),
		TagKeys:     []tag.Key{topicTag, clientIdTag, invocationStatusTag},
	}

	ConsumeDeliveryLatencyView = view.View{
		Name:        "rocketmq_delivery_latency",
		Description: "Message delivery latency",
		Measure:     ConsumeDeliveryMLatencyMs,
		Aggregation: view.Distribution(1, 5, 10, 20, 50, 200, 500),
		TagKeys:     []tag.Key{topicTag, clientIdTag, consumerGroupTag},
	}

	ConsumeAwaitTimeView = view.View{
		Name:        "rocketmq_await_time",
		Description: "Message await time",
		Measure:     ConsumeAwaitMLatencyMs,
		Aggregation: view.Distribution(1, 5, 20, 100, 1000, 5000, 10000),
		TagKeys:     []tag.Key{topicTag, clientIdTag, consumerGroupTag},
	}

	ConsumeProcessTimeView = view.View{
		Name:        "rocketmq_process_time",
		Description: "Message process time",
		Measure:     ConsumeProcessMLatencyMs,
		Aggregation: view.Distribution(1, 5, 10, 100, 1000, 10000, 60000),
		TagKeys:     []tag.Key{topicTag, clientIdTag, consumerGroupTag, invocationStatusTag},
	}
)

var (
	ErrNoAvailableBrokers = errors.New("rocketmq: no available brokers")
)

var NewClient = func(config *Config, opts ...ClientOption) (Client, error) {
	endpoints, err := utils.ParseTarget(config.Endpoint)
	if err != nil {
		return nil, err
	}
	cli := &defaultClient{
		config:                        config,
		opts:                          defaultNSOptions,
		clientID:                      utils.GenClientID(),
		accessPoint:                   endpoints,
		messageInterceptors:           make([]MessageInterceptor, 0),
		endpointsTelemetryClientTable: make(map[string]*defaultClientSession),
		on:                            *atomic.NewBool(true),
		inited:                        *atomic.NewBool(false),
	}
	cli.log = sugarBaseLogger.With("client_id", cli.clientID)
	for _, opt := range opts {
		opt.apply(&cli.opts)
	}
	cli.done = make(chan struct{}, 1)
	cli.clientMeterProvider = NewDefaultClientMeterProvider(cli)
	return cli, nil
}

var NewClientConcrete = func(config *Config, opts ...ClientOption) (*defaultClient, error) {
	endpoints, err := utils.ParseTarget(config.Endpoint)
	if err != nil {
		return nil, err
	}
	cli := &defaultClient{
		config:                        config,
		opts:                          defaultNSOptions,
		clientID:                      utils.GenClientID(),
		accessPoint:                   endpoints,
		messageInterceptors:           make([]MessageInterceptor, 0),
		endpointsTelemetryClientTable: make(map[string]*defaultClientSession),
		on:                            *atomic.NewBool(true),
		clientManager:                 &MockClientManager{},
	}
	cli.log = sugarBaseLogger.With("client_id", cli.clientID)
	for _, opt := range opts {
		opt.apply(&cli.opts)
	}
	cli.done = make(chan struct{}, 1)
	cli.clientMeterProvider = NewDefaultClientMeterProvider(cli)
	return cli, nil
}

var NewClientConn = func(endpoint string, opts ...ConnOption) (ClientConn, error) {
	client := &clientConn{
		opts:     defaultConnOptions,
		validate: validator.New(),
	}
	if len(endpoint) == 0 {
		return nil, ErrNoAvailableEndpoints
	}
	for _, opt := range opts {
		opt.apply(&client.opts)
	}

	baseCtx := context.TODO()
	if client.opts.Context != nil {
		baseCtx = client.opts.Context
	}

	ctx, cancel := context.WithCancel(baseCtx)

	client.ctx = ctx
	client.cancel = cancel
	client.creds = credentials.NewTLS(client.opts.TLS)

	if client.opts.MaxCallSendMsgSize > 0 || client.opts.MaxCallRecvMsgSize > 0 {
		if client.opts.MaxCallRecvMsgSize > 0 && client.opts.MaxCallSendMsgSize > client.opts.MaxCallRecvMsgSize {
			return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", client.opts.MaxCallRecvMsgSize, client.opts.MaxCallSendMsgSize)
		}
		if client.opts.MaxCallSendMsgSize > 0 {
			client.callOpts = append(client.callOpts, grpc.MaxCallSendMsgSize(client.opts.MaxCallSendMsgSize))
		}
		if client.opts.MaxCallRecvMsgSize > 0 {
			client.callOpts = append(client.callOpts, grpc.MaxCallRecvMsgSize(client.opts.MaxCallRecvMsgSize))
		}
	}

	conn, err := client.dial(endpoint)
	if err != nil {
		client.cancel()
		return nil, err
	}
	client.conn = conn

	return client, nil
}

var NewDefaultClientManager = func() *defaultClientManager {
	return &defaultClientManager{
		rpcClientTable: make(map[string]RpcClient),
		done:           make(chan struct{}),
		opts:           defaultClientManagerOptions,
	}
}

var NewDefaultClientMeter = func(exporter view.Exporter, on bool, endpoints *v2.Endpoints, clientID string) *defaultClientMeter {
	return &defaultClientMeter{
		enabled:     *atomic.NewBool(on),
		endpoints:   endpoints,
		ocaExporter: exporter,
	}
}

var NewDefaultClientMeterProvider = func(client *defaultClient) ClientMeterProvider {
	cmp := &defaultClientMeterProvider{
		client:      client,
		clientMeter: NewDefaultClientMeter(nil, false, nil, "nil"),
	}
	client.registerMessageInterceptor(NewDefaultMessageMeterInterceptor(cmp))
	return cmp
}

var NewDefaultMessageMeterInterceptor = func(clientMeterProvider ClientMeterProvider) *defaultMessageMeterInterceptor {
	return &defaultMessageMeterInterceptor{
		clientMeterProvider: clientMeterProvider,
	}
}

var NewDefultInflightRequestCountInterceptor = func() *defultInflightRequestCountInterceptor {
	return &defultInflightRequestCountInterceptor{
		inflightReceiveRequestCount: *atomic.NewInt64(0),
	}
}

var NewFilterExpression = func(expression string) *FilterExpression {
	return &FilterExpression{
		expression:     expression,
		expressionType: TAG,
	}
}

var NewFilterExpressionWithType = func(expression string, expressionType FilterExpressionType) *FilterExpression {
	return &FilterExpression{
		expression:     expression,
		expressionType: expressionType,
	}
}

var NewProducer = func(config *Config, opts ...ProducerOption) (Producer, error) {
	copyOpt := defaultProducerOptions
	po := &copyOpt
	for _, opt := range opts {
		opt.apply(po)
	}
	cli, err := po.clientFunc(config)
	if err != nil {
		return nil, err
	}
	p := &defaultProducer{
		po:      *po,
		cli:     cli.(*defaultClient),
		checker: po.checker,
	}
	p.cli.initTopics = po.topics
	endpoints, err := utils.ParseTarget(config.Endpoint)
	if err != nil {
		return nil, err
	}
	p.pSetting = &producerSettings{
		clientId:   p.cli.GetClientID(),
		endpoints:  endpoints,
		clientType: v2.ClientType_PRODUCER,
		retryPolicy: &v2.RetryPolicy{
			MaxAttempts: po.maxAttempts,
			Strategy: &v2.RetryPolicy_ExponentialBackoff{
				ExponentialBackoff: &v2.ExponentialBackoff{
					Max:        durationpb.New(time.Duration(0)),
					Initial:    durationpb.New(time.Duration(0)),
					Multiplier: 1,
				},
			},
		},
		requestTimeout:      p.cli.opts.timeout,
		validateMessageType: *atomic.NewBool(true),
		maxBodySizeBytes:    *atomic.NewInt32(4 * 1024 * 1024),
	}
	for _, topic := range po.topics {
		topicResource := &v2.Resource{
			Name:              topic,
			ResourceNamespace: config.NameSpace,
		}
		p.pSetting.topics.Store(topic, topicResource)
	}
	p.cli.settings = p.pSetting
	p.cli.clientImpl = p
	return p, nil
}

var NewPublishingLoadBalancer = func(messageQueues []*v2.MessageQueue) (PublishingLoadBalancer, error) {
	plb := &publishingLoadBalancer{
		messageQueues: messageQueues,
	}
	return plb, nil
}

var NewPublishingMessage = func(msg *Message, namespace string, settings *producerSettings, txEnabled bool) (*PublishingMessage, error) {
	if msg == nil {
		return nil, fmt.Errorf("message is nil")
	}
	pMsg := &PublishingMessage{
		msg: msg,
	}

	maxBodySizeBytes := int(settings.maxBodySizeBytes.Load())

	length := len(msg.Body)
	if length > maxBodySizeBytes {
		return nil, fmt.Errorf("message body size exceeds the threshold, max size=%d bytes", maxBodySizeBytes)
	}

	pMsg.encoding = v2.Encoding_IDENTITY

	pMsg.namespace = namespace

	pMsg.messageId = GetMessageIdCodecInstance().NextMessageId().String()

	if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && !txEnabled {
		pMsg.messageType = v2.MessageType_NORMAL
		return pMsg, nil
	}

	if msg.GetMessageGroup() != nil && !txEnabled {
		pMsg.messageType = v2.MessageType_FIFO
		return pMsg, nil
	}

	if msg.GetDeliveryTimestamp() != nil && !txEnabled {
		pMsg.messageType = v2.MessageType_DELAY
		return pMsg, nil
	}

	if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && txEnabled {
		pMsg.messageType = v2.MessageType_TRANSACTION
		return pMsg, nil
	}

	return nil, fmt.Errorf("transactional message should not set messageGroup or deliveryTimestamp")
}

var NewPushConsumer = func(config *Config, opts ...PushConsumerOption) (PushConsumer, error) {
	copyOpt := defaultPushConsumerOptions
	pcOpts := &copyOpt
	for _, opt := range opts {
		opt.apply(pcOpts)
	}
	if len(config.ConsumerGroup) == 0 {
		return nil, fmt.Errorf("consumerGroup could not be nil")
	}
	if pcOpts.messageListener == nil {
		return nil, fmt.Errorf("messageListener could not be nil")
	}
	if utils.CountSyncMapSize(pcOpts.subscriptionExpressions) == 0 {
		return nil, fmt.Errorf("subscriptionExpressions have not been set yet")
	}
	cli, err := pcOpts.clientFunc(config)
	if err != nil {
		return nil, err
	}
	if pcOpts.subscriptionExpressions == nil {
		pcOpts.subscriptionExpressions = &sync.Map{}
	}
	pc := &defaultPushConsumer{
		pcOpts:    *pcOpts,
		cli:       cli.(*defaultClient),
		groupName: config.ConsumerGroup,

		awaitDuration:           pcOpts.awaitDuration,
		subscriptionExpressions: pcOpts.subscriptionExpressions,

		subTopicRouteDataResultCache:    &sync.Map{},
		cacheAssignments:                &sync.Map{},
		processQueueTable:               &sync.Map{},
		stopping:                        *atomic.NewBool(false),
		inflightRequestCountInterceptor: NewDefultInflightRequestCountInterceptor(),
	}
	pc.cli.initTopics = make([]string, 0)
	pcOpts.subscriptionExpressions.Range(func(key, value interface{}) bool {
		pc.cli.initTopics = append(pc.cli.initTopics, key.(string))
		return true
	})
	endpoints, err := utils.ParseTarget(config.Endpoint)
	if err != nil {
		return nil, err
	}
	pc.pcSettings = &pushConsumerSettings{
		clientId:       pc.cli.GetClientID(),
		endpoints:      endpoints,
		clientType:     v2.ClientType_PUSH_CONSUMER,
		requestTimeout: pc.cli.opts.timeout,

		retryPolicy: &v2.RetryPolicy{
			MaxAttempts: 16,
		},
		isFifo:           false,
		receiveBatchSize: 32,

		groupName: &v2.Resource{
			Name:              pc.groupName,
			ResourceNamespace: config.NameSpace,
		},
		longPollingTimeout:      time.Second * 30,
		subscriptionExpressions: pcOpts.subscriptionExpressions,
	}
	pc.cli.settings = pc.pcSettings
	pc.cli.clientImpl = pc
	pc.cli.registerMessageInterceptor(pc.inflightRequestCountInterceptor)
	return pc, nil
}

var NewRpcClient = func(target string, opts ...RpcClientOption) (RpcClient, error) {
	rc := &rpcClient{
		target: target,
		opts:   defaultRpcClientOptions,
	}
	for _, opt := range opts {
		opt.apply(&rc.opts)
	}
	conn, err := rc.opts.clientConnFunc(target, rc.opts.connOptions...)
	if err != nil {
		return nil, fmt.Errorf("create grpc conn failed, err=%w", err)
	}
	rc.conn = conn
	rc.msc = v2.NewMessagingServiceClient(conn.Conn())
	rc.activityNanoTime = time.Now()
	sugarBaseLogger.Infof("create rpc client success, target=%v", target)
	return rc, nil
}

var NewSimpleConsumer = func(config *Config, opts ...SimpleConsumerOption) (SimpleConsumer, error) {
	copyOpt := defaultSimpleConsumerOptions
	scOpts := &copyOpt
	for _, opt := range opts {
		opt.apply(scOpts)
	}
	if len(config.ConsumerGroup) == 0 {
		return nil, fmt.Errorf("consumerGroup could not be nil")
	}
	cli, err := scOpts.clientFunc(config)
	if err != nil {
		return nil, err
	}
	if scOpts.subscriptionExpressions == nil {
		scOpts.subscriptionExpressions = make(map[string]*FilterExpression)
	}
	sc := &defaultSimpleConsumer{
		scOpts:    *scOpts,
		cli:       cli.(*defaultClient),
		groupName: config.ConsumerGroup,

		awaitDuration:           scOpts.awaitDuration,
		subscriptionExpressions: &scOpts.subscriptionExpressions,
	}

	sc.cli.initTopics = make([]string, 0)
	for topic := range scOpts.subscriptionExpressions {
		sc.cli.initTopics = append(sc.cli.initTopics, topic)
	}
	endpoints, err := utils.ParseTarget(config.Endpoint)
	if err != nil {
		return nil, err
	}
	sc.scSettings = &simpleConsumerSettings{
		clientId:       sc.cli.GetClientID(),
		endpoints:      endpoints,
		clientType:     v2.ClientType_SIMPLE_CONSUMER,
		requestTimeout: sc.cli.opts.timeout,

		groupName: &v2.Resource{
			Name:              sc.groupName,
			ResourceNamespace: config.NameSpace,
		},
		longPollingTimeout:      scOpts.awaitDuration,
		subscriptionExpressions: scOpts.subscriptionExpressions,
	}
	sc.cli.settings = sc.scSettings
	sc.cli.clientImpl = sc
	return sc, nil
}

var NewSubscriptionLoadBalancer = func(messageQueues []*v2.MessageQueue) (SubscriptionLoadBalancer, error) {
	slb := &subscriptionLoadBalancer{
		messageQueues: messageQueues,
	}
	return slb, nil
}

var NewTransactionImpl = func(producerImpl Producer) *transactionImpl {
	return &transactionImpl{
		producerImpl: producerImpl,
		messages:     make(map[string]*PublishingMessage),
	}
}

var SUB_ALL = NewFilterExpression("*")