Selly-Modules
/
natsio
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

refactor

This commit is contained in:
namhq1989 2022-12-15 17:34:47 +07:00
parent cdbe3b80e2
commit 685d324788
8 changed files with 84 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
config.go
View File

@ -24,6 +24,9 @@ type Config struct {
// Stream name // Stream name
StreamName string StreamName string
// Debug
Debug bool
} }
func (c Config) validate() error { func (c Config) validate() error {
@ -31,6 +34,10 @@ func (c Config) validate() error {
return errors.New("connect URL is required") return errors.New("connect URL is required")
} }
if c.StreamName == "" {
return errors.New("stream name is required")
}
return nil return nil
} }

16
jetstream_consumer.go
View File

@ -8,27 +8,27 @@ import (
) )
// GetConsumerInfo ... // GetConsumerInfo ...
func (js JetStream) GetConsumerInfo(name string) (*nats.ConsumerInfo, error) { func (js JetStream) GetConsumerInfo(consumerName string) (*nats.ConsumerInfo, error) {
return js.instance.ConsumerInfo(js.streamName, name) return js.instance.ConsumerInfo(js.streamName, consumerName)
} }
// AddConsumer ... // AddConsumer ...
func (js JetStream) AddConsumer(name, filterSubject string) error { func (js JetStream) AddConsumer(consumerName, filterSubject string) error {
// Get consumer first, return if existed // get consumer first, return if existed
consumer, _ := js.GetConsumerInfo(name) consumer, _ := js.GetConsumerInfo(consumerName)
if consumer != nil { if consumer != nil {
return nil return nil
} }
// Add // add
_, err := js.instance.AddConsumer(js.streamName, &nats.ConsumerConfig{ _, err := js.instance.AddConsumer(js.streamName, &nats.ConsumerConfig{
Durable: name, Durable: consumerName,
AckPolicy: nats.AckExplicitPolicy, AckPolicy: nats.AckExplicitPolicy,
FilterSubject: filterSubject, FilterSubject: filterSubject,
}) })
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - add consumer %s for stream #%s error: %s", name, js.streamName, err.Error()) msg := fmt.Sprintf("[natsio.JetStream] add consumer %s to stream #%s error: %s", consumerName, js.streamName, err.Error())
return errors.New(msg) return errors.New(msg)
} }
return nil return nil

51
jetstream_pubsub.go
View File

@ -3,62 +3,57 @@ package natsio
import ( import (
"errors" "errors"
"fmt" "fmt"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
) )
// Publish ... // Publish ...
func (js JetStream) Publish(subject string, payload []byte) error { func (js JetStream) Publish(subject string, payload []byte) error {
channel := combineStreamAndSubjectName(js.streamName, subject) _, err := js.instance.PublishAsync(subject, payload)
_, err := js.instance.PublishAsync(channel, payload)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - publish message to subject #%s error: %s", channel, err.Error()) msg := fmt.Sprintf("[natsio.JetStream] publish message to subject #%s error: %s", subject, err.Error())
return errors.New(msg) return errors.New(msg)
} }
if js.debug {
fmt.Printf("[natsio.JetStream] published a message to subject #%s \n", subject)
}
return nil return nil
} }
// Subscribe ... // Subscribe ...
func (js JetStream) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) { func (js JetStream) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) {
channel := combineStreamAndSubjectName(js.streamName, subject) sub, err := js.instance.Subscribe(subject, cb)
sub, err := js.instance.Subscribe(channel, cb)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - subscribe subject %s error: %s", channel, err.Error()) msg := fmt.Sprintf("[natsio.JetStream] subscribe subject %s error: %s", subject, err.Error())
return nil, errors.New(msg) return nil, errors.New(msg)
} }
if js.debug {
fmt.Printf("[natsio.JetStream] subscribe new message from subject #%s \n", subject)
}
return sub, nil return sub, nil
} }
// PullSubscribe ...
//
// Example:
//
// js := natsio.GetJetStream()
//
// sub, err := js.PullSubscribe("A_SUBJECT", "A_SUBJECT", "A_CONSUMER")
//
// for {
// messages, err := sub.Fetch(10)
// // process each messages
// }
func (js JetStream) PullSubscribe(subject, consumer string) (*nats.Subscription, error) { func (js JetStream) PullSubscribe(subject, consumer string) (*nats.Subscription, error) {
channel := combineStreamAndSubjectName(js.streamName, subject) // check if consumer existed
// Check if consumer existed
con, err := js.GetConsumerInfo(consumer) con, err := js.GetConsumerInfo(consumer)
if con == nil || err != nil { if con == nil || err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - pull subscribe consumer %s not existed in stream %s", consumer, js.streamName) msg := fmt.Sprintf("[natsio.JetStream] pull subscribe consumer %s not existed in stream %s", consumer, js.streamName)
return nil, errors.New(msg) return nil, errors.New(msg)
} }
// Pull // pull
sub, err := js.instance.PullSubscribe(channel, consumer) sub, err := js.instance.PullSubscribe(subject, consumer)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - pull subscribe subject #%s - consumer #%s error: %s", channel, consumer, err.Error()) msg := fmt.Sprintf("[natsio.JetStream] pull subscribe subject #%s - consumer #%s error: %s", subject, consumer, err.Error())
return nil, errors.New(msg) return nil, errors.New(msg)
} }
if js.debug {
fmt.Printf("[natsio.JetStream] pull subscribe new message from subject #%s \n", subject)
}
return sub, nil return sub, nil
} }

11
jetstream_queue.go
View File

@ -9,12 +9,15 @@ import (
// QueueSubscribe ... // QueueSubscribe ...
func (js JetStream) QueueSubscribe(subject, queueName string, cb nats.MsgHandler) error { func (js JetStream) QueueSubscribe(subject, queueName string, cb nats.MsgHandler) error {
channel := combineStreamAndSubjectName(js.streamName, subject) _, err := js.instance.QueueSubscribe(subject, queueName, cb)
_, err := js.instance.QueueSubscribe(channel, queueName, cb)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - queue subscribe with subject #%s error: %s", channel, err.Error()) msg := fmt.Sprintf("[natsio.JetStream] queue subscribe with subject #%s error: %s", subject, err.Error())
return errors.New(msg) return errors.New(msg)
} }
if js.debug {
fmt.Printf("[natsio.JetStream] queue subscribe new message from subject #%s \n", subject)
}
return nil return nil
} }

38
jetstream_stream.go
View File

@ -25,52 +25,48 @@ func (js JetStream) GetStreamInfo() (*nats.StreamInfo, error) {
return js.instance.StreamInfo(js.streamName) return js.instance.StreamInfo(js.streamName)
} }
// AddStream add new stream, with default config // addStream add new stream, with default config
// Due to subject must have a unique name, subject name will be combined with stream name func (js JetStream) addStream() {
// E.g: stream name is "DEMO", subject name is "Subject-1", so final name in NATS will be: DEMO.Subject-1 // get info about the stream
func (js JetStream) AddStream(subjects []string) error {
// Get info about the stream
stream, _ := js.GetStreamInfo() stream, _ := js.GetStreamInfo()
// If stream not found, create new // if stream not found, create new
if stream == nil { if stream == nil {
subjectNames := generateSubjectNames(js.streamName, subjects) _, err := js.instance.AddStream(generateStreamConfig(js.streamName, []string{}))
_, err := js.instance.AddStream(generateStreamConfig(js.streamName, subjectNames))
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - add stream error: %s", err.Error()) fmt.Printf("[natsio.JetStream] add stream %s error: %s \n", js.streamName, err.Error())
return errors.New(msg)
} }
} }
return nil
} }
// DeleteStream ... // DeleteStream ...
func (js JetStream) DeleteStream() error { func (js JetStream) DeleteStream() error {
if err := js.instance.DeleteStream(js.streamName); err != nil { if err := js.instance.DeleteStream(js.streamName); err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - delete stream error: %s", err.Error()) msg := fmt.Sprintf("[natsio.JetStream] delete stream %s error: %s", js.streamName, err.Error())
return errors.New(msg) return errors.New(msg)
} }
return nil return nil
} }
// AddStreamSubjects ... // AddSubjects ...
func (js JetStream) AddStreamSubjects(subjects []string) error { func (js JetStream) AddSubjects(subjects []string) error {
// Get info about the stream // get stream info
stream, _ := js.GetStreamInfo() stream, _ := js.GetStreamInfo()
if stream == nil { if stream == nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - error when adding stream %s subjects: stream not found", js.streamName) msg := fmt.Sprintf("[natsio.JetStream] error when adding stream #%s subjects: stream not found", js.streamName)
return errors.New(msg) return errors.New(msg)
} }
// Merge current and new subjects // merge current and new subjects
subjectNames := generateSubjectNames(js.streamName, subjects) newSubjects := mergeAndUniqueArrayStrings(subjects, stream.Config.Subjects)
newSubjects := mergeAndUniqueArrayStrings(subjectNames, stream.Config.Subjects)
// update
_, err := js.instance.UpdateStream(generateStreamConfig(js.streamName, newSubjects)) _, err := js.instance.UpdateStream(generateStreamConfig(js.streamName, newSubjects))
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS JETSTREAM] - add stream error: %s", err.Error()) msg := fmt.Sprintf("[natsio.JetStream] add subject to stream #%s error: %s", js.streamName, err.Error())
return errors.New(msg) return errors.New(msg)
} }
return nil return nil
} }

4
natsio.go
View File

@ -10,11 +10,13 @@ import (
// Server ... // Server ...
type Server struct { type Server struct {
instance *nats.Conn instance *nats.Conn
debug bool
} }
// JetStream ... // JetStream ...
type JetStream struct { type JetStream struct {
instance nats.JetStreamContext instance nats.JetStreamContext
debug bool
streamName string streamName string
} }
@ -53,6 +55,7 @@ func Connect(cfg Config) (*Server, *JetStream, error) {
// set client // set client
natsServer.instance = nc natsServer.instance = nc
natsServer.debug = cfg.Debug
// create jet stream context // create jet stream context
js, err := nc.JetStream(nats.PublishAsyncMaxPending(256)) js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
@ -61,6 +64,7 @@ func Connect(cfg Config) (*Server, *JetStream, error) {
return nil, nil, errors.New(msg) return nil, nil, errors.New(msg)
} }
natsJetStream.instance = js natsJetStream.instance = js
natsJetStream.debug = cfg.Debug
natsJetStream.streamName = cfg.StreamName natsJetStream.streamName = cfg.StreamName
// assign client type // assign client type

32
server_reqres.go
View File

@ -3,7 +3,6 @@ package natsio
import ( import (
"errors" "errors"
"fmt" "fmt"
"log"
"time" "time"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
@ -13,21 +12,24 @@ import (
const requestTimeout = 10 * time.Second const requestTimeout = 10 * time.Second
// Request ... // Request ...
func (sv Server) Request(subject string, payload []byte) (*nats.Msg, error) { func (s Server) Request(subject string, payload []byte) (*nats.Msg, error) {
timeout := requestTimeout timeout := requestTimeout
if globalConfig.RequestTimeout > 0 { if globalConfig.RequestTimeout > 0 {
timeout = globalConfig.RequestTimeout timeout = globalConfig.RequestTimeout
} }
msg, err := sv.instance.Request(subject, payload, timeout) msg, err := s.instance.Request(subject, payload, timeout)
if errors.Is(err, nats.ErrNoResponders) { if errors.Is(err, nats.ErrNoResponders) {
log.Printf("[NATS SERVER]: request - no responders for subject: %s", subject) fmt.Printf("[natsio.Server] no responders for subject: %s \n", subject)
} else if s.debug {
fmt.Printf("[natsio.Server] send request to subject #%s successfully \n", subject)
} }
return msg, err return msg, err
} }
// RequestWithBindData ... // RequestWithBindData ...
func (sv Server) RequestWithBindData(subject string, payload []byte, result interface{}) error { func (s Server) RequestWithBindData(subject string, payload []byte, result interface{}) error {
msg, err := sv.Request(subject, payload) msg, err := s.Request(subject, payload)
if msg == nil || err != nil { if msg == nil || err != nil {
return err return err
} }
@ -37,26 +39,28 @@ func (sv Server) RequestWithBindData(subject string, payload []byte, result inte
} }
// Reply ... // Reply ...
func (sv Server) Reply(msg *nats.Msg, payload []byte) error { func (s Server) Reply(msg *nats.Msg, payload []byte) error {
return sv.instance.Publish(msg.Reply, payload) return s.instance.Publish(msg.Reply, payload)
} }
// Subscribe ... // Subscribe ...
func (sv Server) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) { func (s Server) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) {
sub, err := sv.instance.Subscribe(subject, cb) sub, err := s.instance.Subscribe(subject, cb)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS SERVER] - subscribe subject %s error: %s", subject, err.Error()) msg := fmt.Sprintf("[natsio.Server] subscribe subject %s error: %s", subject, err.Error())
return nil, errors.New(msg) return nil, errors.New(msg)
} }
return sub, nil return sub, nil
} }
// QueueSubscribe ... // QueueSubscribe ...
func (sv Server) QueueSubscribe(subject, queue string, cb nats.MsgHandler) (*nats.Subscription, error) { func (s Server) QueueSubscribe(subject, queue string, cb nats.MsgHandler) (*nats.Subscription, error) {
sub, err := sv.instance.QueueSubscribe(subject, queue, cb) sub, err := s.instance.QueueSubscribe(subject, queue, cb)
if err != nil { if err != nil {
msg := fmt.Sprintf("[NATS SERVER] - queue subscribe subject %s, queue %s error: %s", subject, queue, err.Error()) msg := fmt.Sprintf("[natsio.Server] queue subscribe subject %s, queue %s error: %s", subject, queue, err.Error())
return nil, errors.New(msg) return nil, errors.New(msg)
} }
return sub, nil return sub, nil
} }

10
utils.go
View File

@ -16,16 +16,6 @@ func mergeAndUniqueArrayStrings(arr1, arr2 []string) []string {
return result return result
} }
// generateSubjectNames ...
func generateSubjectNames(streamName string, subjects []string) []string {
var result = make([]string, 0)
for _, subject := range subjects {
name := combineStreamAndSubjectName(streamName, subject)
result = append(result, name)
}
return result
}
func combineStreamAndSubjectName(stream, subject string) string { func combineStreamAndSubjectName(stream, subject string) string {
return fmt.Sprintf("%s.%s", stream, subject) return fmt.Sprintf("%s.%s", stream, subject)
} }