working, need to fix tests still

client.go

@@ -95,6 +95,19 @@ func (c *Client) Publish(ctx context.Context, subject string, message []byte) er
 
 // Subscribe subscribes to a topic and returns a single message.
 func (c *Client) Subscribe(ctx context.Context, topic string) (*pubsub.Message, error) {
+
+	if c.subManager == nil {
+		return nil, errors.New("subscription manager is not initialized")
+	}
+
+	if c.connManager == nil {
+		return nil, errors.New("connection manager is not initialized")
+	}
+
+	if c.connManager.JetStream() == nil {
+		return nil, errors.New("jetstream is not initialized")
+	}
+
 	return c.subManager.Subscribe(ctx, topic, c.connManager.JetStream(), c.Config, c.logger, c.metrics)
 }
 

config.go

@@ -2,6 +2,8 @@ package nats
 
 import (
 	"time"
+
+	"gofr.dev/pkg/gofr/datasource/pubsub"
 )
 
 // Config defines the Client configuration.
@@ -25,7 +27,7 @@ type StreamConfig struct {
 }
 
 // New creates a new Client.
-func New(cfg *Config) *PubSubWrapper {
+func New(cfg *Config, logger pubsub.Logger) *PubSubWrapper {
 	if cfg == nil {
 		cfg = &Config{}
 	}
@@ -37,6 +39,7 @@ func New(cfg *Config) *PubSubWrapper {
 	client := &Client{
 		Config:     cfg,
 		subManager: NewSubscriptionManager(cfg.BatchSize),
+		logger:     logger,
 	}
 
 	return &PubSubWrapper{Client: client}

connection_manager.go

@@ -56,6 +56,20 @@ func NewConnectionManager(
 	logger pubsub.Logger,
 	natsConnector NATSConnector,
 	jetStreamCreator JetStreamCreator) *ConnectionManager {
+
+	// if logger is nil panic
+	if logger == nil {
+		panic("logger is required")
+	}
+
+	if natsConnector == nil {
+		natsConnector = &DefaultNATSConnector{}
+	}
+
+	if jetStreamCreator == nil {
+		jetStreamCreator = &DefaultJetStreamCreator{}
+	}
+
 	return &ConnectionManager{
 		config:           cfg,
 		logger:           logger,
@@ -66,7 +80,15 @@ func NewConnectionManager(
 
 // Connect establishes a connection to NATS and sets up JetStream.
 func (cm *ConnectionManager) Connect() error {
-	connInterface, err := cm.natsConnector.Connect(cm.config.Server, nats.Name("GoFr NATS JetStreamClient"))
+	cm.logger.Logf("Connecting to NATS server at %v", cm.config.Server)
+
+	opts := []nats.Option{nats.Name("GoFr NATS JetStreamClient")}
+
+	if cm.config.CredsFile != "" {
+		opts = append(opts, nats.UserCredentials(cm.config.CredsFile))
+	}
+
+	connInterface, err := cm.natsConnector.Connect(cm.config.Server, opts...)
 	if err != nil {
 		cm.logger.Errorf("failed to connect to NATS server at %v: %v", cm.config.Server, err)
 

connectors.go

@@ -1,3 +1,4 @@
+// Package nats connector.go
 package nats
 
 import (

go.mod

@@ -1,4 +1,4 @@
-module gofr.dev/pkg/gofr/datasource/pubsub/nats
+module github.com/carverauto/gofr-nats
 
 go 1.23.1
 

pubsub_wrapper.go

@@ -2,6 +2,7 @@ package nats
 
 import (
 	"context"
+	"fmt"
 
 	"gofr.dev/pkg/gofr/datasource"
 	"gofr.dev/pkg/gofr/datasource/pubsub"
@@ -45,9 +46,15 @@ func (w *PubSubWrapper) Health() datasource.Health {
 
 // Connect establishes a connection to NATS.
 func (w *PubSubWrapper) Connect() {
+	fmt.Println("Connecting to NATS using PubSubWrapper")
+	if w.Client.connManager != nil && w.Client.connManager.Health().Status == datasource.StatusUp {
+		w.Client.logger.Log("NATS connection already established")
+		return
+	}
+
 	err := w.Client.Connect()
 	if err != nil {
-		w.Client.logger.Errorf("Error connecting to NATS: %v", err)
+		w.Client.logger.Errorf("PubSubWrapper: Error connecting to NATS: %v", err)
 	}
 }
 

subscription_manager.go

@@ -127,6 +127,7 @@ func (*SubscriptionManager) consumeMessages(
 	buffer chan *pubsub.Message,
 	cfg *Config,
 	logger pubsub.Logger) {
+	// TODO: propagate errors to caller
 	for {
 		select {
 		case <-ctx.Done():