129 lines
3.5 KiB
Go
129 lines
3.5 KiB
Go
package nats
|
|
|
|
import (
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
const (
|
|
ScraperTopic = "scraper"
|
|
LinkedInSubTopic = "scraper.linkedin"
|
|
IndeedSubTopic = "scraper.indeed"
|
|
BaytSubTopic = "scraper.bayt"
|
|
TokyoDevSubTopic = "scraper.tokyodev"
|
|
JapanDevSubTopic = "scraper.japandev"
|
|
)
|
|
|
|
type NatsClient struct {
|
|
Conn *nats.Conn
|
|
JetStream nats.JetStreamContext
|
|
StreamName string
|
|
}
|
|
|
|
func NewNatsClient() (*NatsClient, error) {
|
|
// Connect to NATS
|
|
nc, err := nats.Connect(nats.DefaultURL, nats.RetryOnFailedConnect(true),
|
|
nats.MaxReconnects(5),
|
|
nats.ReconnectWait(5*time.Second))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error connecting to NATS: %v", err)
|
|
}
|
|
|
|
// Create JetStream Context
|
|
js, err := nc.JetStream()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error getting JetStream context: %v", err)
|
|
}
|
|
|
|
// Create Stream
|
|
streamName := "SCRAPER_STREAM"
|
|
_, err = js.StreamInfo(streamName)
|
|
if err != nil {
|
|
// Stream doesn't exist, let's create it
|
|
_, err := js.AddStream(&nats.StreamConfig{
|
|
Name: streamName,
|
|
Subjects: []string{
|
|
LinkedInSubTopic,
|
|
IndeedSubTopic,
|
|
BaytSubTopic,
|
|
TokyoDevSubTopic,
|
|
JapanDevSubTopic,
|
|
},
|
|
Storage: nats.FileStorage,
|
|
MaxAge: 24 * time.Hour,
|
|
Retention: nats.WorkQueuePolicy,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating stream: %v", err)
|
|
}
|
|
log.Printf("Created new stream: %s", streamName)
|
|
}
|
|
|
|
return &NatsClient{
|
|
Conn: nc,
|
|
JetStream: js,
|
|
StreamName: streamName,
|
|
}, nil
|
|
}
|
|
|
|
// Publish publishes a message to a specific topic
|
|
func (n *NatsClient) Publish(topic string, data []byte) error {
|
|
_, err := n.JetStream.Publish(topic, data)
|
|
if err != nil {
|
|
return fmt.Errorf("error publishing message: %v", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Subscribe creates a subscription to a specific topic with production-ready error handling
|
|
func (n *NatsClient) Subscribe(topic, consumerName string, handler nats.MsgHandler) (*nats.Subscription, error) {
|
|
// Define subscription options
|
|
subscribeOptions := []nats.SubOpt{
|
|
nats.Durable(consumerName), // Durable consumer name
|
|
nats.ManualAck(), // Manual acknowledgment
|
|
nats.AckExplicit(), // Explicit acknowledgment required
|
|
nats.DeliverAll(), // Deliver all messages
|
|
}
|
|
|
|
// First attempt to subscribe
|
|
sub, err := n.JetStream.Subscribe(topic, handler, subscribeOptions...)
|
|
if err != nil {
|
|
// Check if error is due to consumer already being bound
|
|
if strings.Contains(err.Error(), "already bound") ||
|
|
strings.Contains(err.Error(), "consumer is already bound") {
|
|
|
|
log.Printf("Consumer %s is already bound, attempting to delete and recreate", consumerName)
|
|
|
|
// Delete the existing consumer
|
|
deleteErr := n.JetStream.DeleteConsumer(n.StreamName, consumerName)
|
|
if deleteErr != nil {
|
|
log.Printf("Warning: Failed to delete existing consumer: %v", deleteErr)
|
|
}
|
|
|
|
// Retry subscription after deleting consumer
|
|
sub, err = n.JetStream.Subscribe(topic, handler, subscribeOptions...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error subscribing to topic after consumer deletion: %v", err)
|
|
}
|
|
|
|
log.Printf("Successfully recreated consumer %s and subscribed to %s", consumerName, topic)
|
|
} else {
|
|
return nil, fmt.Errorf("error subscribing to topic: %v", err)
|
|
}
|
|
} else {
|
|
log.Printf("Successfully subscribed to %s using existing consumer %s", topic, consumerName)
|
|
}
|
|
|
|
return sub, nil
|
|
}
|
|
|
|
// Close closes the NATS connection
|
|
func (n *NatsClient) Close() {
|
|
if n.Conn != nil {
|
|
n.Conn.Close()
|
|
}
|
|
}
|