feat:ok

queue/nats/nats.go

@@ -3,6 +3,7 @@ package nats
 import (
 	"strings"
 
+	"git.apinb.com/bsm-sdk/core/errcode"
 	"git.apinb.com/bsm-sdk/core/vars"
 	natsgo "github.com/nats-io/nats.go"
 )
@@ -11,8 +12,14 @@ type Nats struct {
 	Client natsgo.JetStreamContext
 }
 
-func NewNats(endpoints []string) (*Nats, error) {
-	jetStream, err := NatsNew(endpoints)
+func NewNats(endpoints []string, space string) (*Nats, error) {
+	if len(endpoints) == 0 {
+		return nil, errcode.ErrMq
+	}
+	if space == "" {
+		space = vars.MQSpaceName
+	}
+	jetStream, err := NatsNew(endpoints, space)
 	if err != nil {
 		return nil, err
 	}
@@ -22,7 +29,7 @@ func NewNats(endpoints []string) (*Nats, error) {
 	}, nil
 }
 
-func NatsNew(endpoints []string) (natsgo.JetStreamContext, error) {
+func NatsNew(endpoints []string, space string) (natsgo.JetStreamContext, error) {
 	var serverUrl string
 	if len(endpoints) > 1 {
 		serverUrl = strings.Join(endpoints, ",")
@@ -42,8 +49,8 @@ func NatsNew(endpoints []string) (natsgo.JetStreamContext, error) {
 	}
 
 	js.AddStream(&natsgo.StreamConfig{
-		Name:      vars.MQSpaceName,
-		Subjects:  []string{vars.MQSpaceName}, //jetstream不支持通配符
+		Name:      space,
+		Subjects:  []string{space}, //jetstream不支持通配符
 		Retention: natsgo.WorkQueuePolicy,
 		MaxBytes:  8,
 	})

queue/pulsar/pulsar.go

@@ -1,88 +0,0 @@
-package pulsar
-
-import (
-	"context"
-	"errors"
-
-	"git.apinb.com/bsm-sdk/core/types"
-	"git.apinb.com/bsm-sdk/core/vars"
-	pulsargo "github.com/apache/pulsar-client-go/pulsar"
-)
-
-type Pulsar struct {
-	Client pulsargo.Client
-}
-
-func NewPulsar(cfg *types.PulsarConf) (*Pulsar, error) {
-	client, err := pulsargo.NewClient(pulsargo.ClientOptions{
-		URL:               cfg.Endpoints, //TODO: 更换为接入点地址（控制台集群管理页完整复制）
-		Authentication:    pulsargo.NewAuthenticationToken(cfg.Token),
-		OperationTimeout:  vars.OperationTimeout,
-		ConnectionTimeout: vars.ConnectionTimeout,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return &Pulsar{
-		Client: client,
-	}, nil
-}
-
-// push to pulsar server, return messageid.
-func (mq *Pulsar) Producer(topic, msg string) (MessageID string, err error) {
-	if msg == "" {
-		return "", errors.New("Message is nil.")
-	}
-
-	producer, err := mq.Client.CreateProducer(pulsargo.ProducerOptions{
-		Topic:           topic,
-		CompressionType: pulsargo.ZSTD,
-	})
-
-	if err != nil {
-		return "", err
-	}
-
-	msgID, err := producer.Send(context.Background(), &pulsargo.ProducerMessage{
-		Payload: []byte(msg),
-	})
-
-	if err != nil {
-		return "", err
-	}
-
-	return msgID.String(), nil
-}
-
-func (mq *Pulsar) Subscribe(topic, subscription string, subType pulsargo.SubscriptionType, do func([]byte)) error {
-	// we can listen this channel
-	channel := make(chan pulsargo.ConsumerMessage, 100)
-
-	options := pulsargo.ConsumerOptions{
-		Topic:            topic,
-		SubscriptionName: subscription,
-		Type:             subType,
-		// fill `MessageChannel` field will create a listener
-		MessageChannel: channel,
-	}
-
-	consumer, err := mq.Client.Subscribe(options)
-	if err != nil {
-		return err
-	}
-
-	defer consumer.Close()
-
-	// Receive messages from channel. The channel returns a struct `ConsumerMessage` which contains message and the consumer from where
-	// the message was received. It's not necessary here since we have 1 single consumer, but the channel could be
-	// shared across multiple consumers as well
-	for cm := range channel {
-		consumer := cm.Consumer
-		msg := cm.Message
-
-		do(msg.Payload())
-		consumer.Ack(msg)
-	}
-	return nil
-}