implement nats-jetstreaming subscription with multiple consumer modes #85
Conversation
This commit implements subscription based on the nats-jetstreaming queue, leveraging JetStream to offer multiple consumer modes and supporting various consumer configurations and consumption methods to meet the demands of different business scenarios.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
10 file(s) reviewed, 20 comment(s)
Edit PR Review Bot Settings | Greptile
| - "--jetstream" | ||
| - "--debug" |
There was a problem hiding this comment.
style: Debug mode is enabled which may impact performance in production. Consider making this configurable or disabling for production deployments.
| ports: | ||
| - "4222:4222" | ||
| - "8222:8222" |
There was a problem hiding this comment.
style: Ports are exposed directly to the host without any network isolation. Consider using internal networks for production deployments.
| log.Fatalf("failed to create consumer manager: %v", err) | ||
| } | ||
|
|
||
| go cm.Start() |
There was a problem hiding this comment.
style: Start() is a blocking function that only returns when Stop() is called, but it's being run in a goroutine without any synchronization mechanism to ensure it's fully initialized before potential shutdown signals arrive.
| cm.Stop() | ||
| time.Sleep(time.Second) |
There was a problem hiding this comment.
style: Using time.Sleep() for shutdown coordination is not ideal. Consider using a WaitGroup or a done channel to properly wait for the consumer manager to fully stop.
| - consumerConfig: | ||
| filterSubjects: ["user.recharge"] | ||
| delivery: | ||
| consumptionMethod: "fetchNoWait" | ||
| streamName: "user" |
There was a problem hiding this comment.
style: This consumer configuration is missing a name and durable property, unlike the other configurations. While this may work, it could make tracking and managing this consumer difficult in production.
| log.Printf("failed to initialize jetstream for stream %q: %v", streamName, err) | ||
| continue | ||
| } | ||
| ctx := context.Background() |
There was a problem hiding this comment.
style: Using a background context without timeout for stream creation could lead to hanging operations if the NATS server is unresponsive.
| // | ||
| // nc - pointer to the NATS connection | ||
| // cfgs - list of JetStreamConfig configurations to register | ||
| func RegisterStreamInstances(nc *nats.Conn, cfgs []*JetStreamConfig) { |
There was a problem hiding this comment.
style: This function lacks validation for the nc parameter. If nil is passed, it will cause panic when used.
| func RegisterStreamInstances(nc *nats.Conn, cfgs []*JetStreamConfig) { | |
| func RegisterStreamInstances(nc *nats.Conn, cfgs []*JetStreamConfig) { | |
| if nc == nil { | |
| log.Printf("error: nil NATS connection provided to RegisterStreamInstances") | |
| return | |
| } |
| for streamName, streamMgr := range streamRegistry { | ||
| streamInstLock.RLock() | ||
| _, exists := streamInstances[streamName] | ||
| streamInstLock.RUnlock() | ||
| if exists { | ||
| log.Printf("streamInstance %q already created", streamName) | ||
| continue | ||
| } | ||
| // Initialize JetStream context | ||
| if err := streamMgr.InitJetStream(nc); err != nil { | ||
| log.Printf("failed to initialize jetstream for stream %q: %v", streamName, err) | ||
| continue | ||
| } | ||
| ctx := context.Background() | ||
| stream, err := streamMgr.CreateStream(ctx) | ||
| if err != nil { | ||
| log.Printf("failed to create stream %q: %v", streamName, err) | ||
| continue | ||
| } | ||
| streamInstLock.Lock() | ||
| streamInstances[streamName] = stream | ||
| streamInstLock.Unlock() | ||
| log.Printf("streamInstance %q created", streamName) | ||
| } |
There was a problem hiding this comment.
style: Potential race condition when iterating over streamRegistry while other goroutines might modify it. Consider making a copy of the registry keys before iteration.
| natsConf *common.NatsConfig | ||
|
|
||
| // List of global JetStream configurations parsed from a config file. | ||
| JetStreamConfigs []*common.JetStreamConfig |
There was a problem hiding this comment.
style: JetStreamConfigs field is exported but doesn't need to be. Consider making it private since it's only used internally.
| func (cm *ConsumerManager) runPullMessages(ctx context.Context, cfg *ConsumerQueueConfig, fetchFn func(num int) (jetstream.MessageBatch, error)) { | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| default: | ||
| } | ||
| msgs, err := fetchFn(cfg.Delivery.FetchCount) | ||
| if err != nil { | ||
| log.Printf("error fetching messages: %v", err) | ||
| continue | ||
| } | ||
| for msg := range msgs.Messages() { | ||
| cm.ackMessage(cfg, msg) | ||
| } | ||
| if fetchErr := msgs.Error(); fetchErr != nil { | ||
| log.Printf("error after fetching messages: %v", fetchErr) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
style: No backoff mechanism when errors occur in the pull loop. Consider adding exponential backoff to avoid hammering the server during error conditions.
There was a problem hiding this comment.
10 file(s) reviewed, 10 comment(s)
Edit PR Review Bot Settings | Greptile
| services: | ||
| nats: |
There was a problem hiding this comment.
style: No volume configuration for persistence. JetStream data will be lost when container restarts.
| var queueConfigs []*consumer.ConsumerQueueConfig | ||
| for i := range c.ConsumerQueues { | ||
| c.ConsumerQueues[i].Handler = &MyConsumeHandler{} | ||
| queueConfigs = append(queueConfigs, &c.ConsumerQueues[i]) | ||
| } |
There was a problem hiding this comment.
style: Creating a new slice and appending to it is redundant. Could directly initialize queueConfigs with the right capacity.
| return &JetStreamManager{ | ||
| streamConf: streamConf, | ||
| } |
There was a problem hiding this comment.
logic: JS field is declared in JetStreamManager struct but not initialized in NewJetStream function. This could lead to nil pointer dereference when using the JS field.
| func RegisterManager(streamID string, mgr *JetStreamManager) { | ||
| registryLock.Lock() | ||
| defer registryLock.Unlock() | ||
| streamRegistry[streamID] = mgr | ||
| } |
There was a problem hiding this comment.
style: This function lacks validation for the mgr parameter. If nil is passed, it will silently accept it and cause issues later.
| func RegisterManager(streamID string, mgr *JetStreamManager) { | |
| registryLock.Lock() | |
| defer registryLock.Unlock() | |
| streamRegistry[streamID] = mgr | |
| } | |
| func RegisterManager(streamID string, mgr *JetStreamManager) { | |
| if mgr == nil { | |
| log.Printf("error: nil JetStreamManager provided for stream %q", streamID) | |
| return | |
| } | |
| registryLock.Lock() | |
| defer registryLock.Unlock() | |
| streamRegistry[streamID] = mgr | |
| } |
| // ConsumerQueueConfig defines the full configuration for building a consumer queue. | ||
| // If StreamName is empty, the default stream (DefaultStream) will be used. | ||
| type ConsumerQueueConfig struct { | ||
| StreamName string `json:"streamName,optional"` // Name of the stream to associate with; if empty, uses default stream |
There was a problem hiding this comment.
logic: The comment mentions 'DefaultStream' but there's no definition of this constant in the file. Make sure it's defined elsewhere or document where it comes from.
| stream = s | ||
| } | ||
|
|
||
| ctx := context.Background() |
There was a problem hiding this comment.
style: Consider using a context with timeout here to prevent blocking indefinitely during consumer creation
| if err := cfg.Handler.Consume(msg); err != nil { | ||
| log.Printf("message processing error: %v", err) | ||
| return | ||
| } |
There was a problem hiding this comment.
style: No handling for message processing errors. Consider implementing negative acknowledgment (NAK) or retry logic when Handler.Consume returns an error
| // This function blocks until Stop() is invoked. | ||
| func (cm *ConsumerManager) Start() { | ||
| // Initialize stream instances (register or update streams) based on the provided JetStream configurations. | ||
| common.RegisterStreamInstances(cm.nc, cm.JetStreamConfigs) |
There was a problem hiding this comment.
logic: No error handling for RegisterStreamInstances. If stream registration fails, the consumer will continue without streams
| func (cm *ConsumerManager) Stop() { | ||
| for _, consumerCtx := range cm.subscribers { | ||
| consumerCtx.Stop() | ||
| } | ||
| for _, cancel := range cm.cancelFuncs { | ||
| cancel() | ||
| } | ||
| if cm.nc != nil { | ||
| cm.nc.Close() | ||
| } | ||
| close(cm.stopCh) |
There was a problem hiding this comment.
style: No graceful shutdown mechanism to ensure in-flight messages are processed before stopping. Consider adding a wait group or timeout
| // JetStreamPublisher implements the Publisher interface by utilizing an internal JetStream context for message publishing. | ||
| // Note: It is recommended to rename the package from "publiser" to "publisher" for better clarity. |
There was a problem hiding this comment.
style: The comment mentions renaming the package from 'publiser' to 'publisher', but the package is already correctly named 'publisher' on line 1. This comment appears to be outdated and should be removed.
| // JetStreamPublisher implements the Publisher interface by utilizing an internal JetStream context for message publishing. | |
| // Note: It is recommended to rename the package from "publiser" to "publisher" for better clarity. | |
| // JetStreamPublisher implements the Publisher interface by utilizing an internal JetStream context for message publishing. |
This commit implements subscription based on the nats-jetstreaming queue, leveraging JetStream to offer multiple consumer modes and supporting various consumer configurations and consumption methods to meet the demands of different business scenarios.
Greptile Summary
This PR implements NATS JetStream subscription functionality in the go-queue library, providing flexible messaging capabilities with multiple consumer modes and configuration options.
natsmq/consumer/consumer.gowithConsumerManagersupporting both push-based and pull-based consumption methodsnatsmq/publisher/publisher.gowithJetStreamPublisherfor message publishing to subjectsnatsmq/commonpackage with stream management utilities and configuration structuresnatsmq/consumer/config.gowith support for different acknowledgment policiesGreptile AI
💡 (1/5) You can manually trigger the bot by mentioning @greptileai in a comment!