Fix/tests and defaults

pkg/pubsub/publisher.go

@@ -21,11 +21,10 @@ type Publisher struct {
 	httpClient *http.Client
 }
 
-// NewPublisher instantiates a new Publisher. It returns an error if the underlying
-// API dataplane client fails to initialize.
+// NewPublisher instantiates a new Publisher.
 func NewPublisher(topicID uuid.UUID, opts ...Option) *Publisher {
 	cfg := &clientConfig{
-		httpClient: http.DefaultClient,
+		httpClient: &http.Client{},
 		host:       "pubsub.eu01.onstackit.cloud",
 		logger:     logr.FromSlogHandler(slog.Default().Handler()),
 	}

pkg/pubsub/pulljob.go

@@ -3,13 +3,14 @@ package pubsub
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 )
 
 type pullJob struct {
 	subscription     *Subscriber
 	maxPullMessages  int32
-	longPullDuration *int32
+	longPullDuration int32
 	interval         time.Duration
 	bufferSize       int
 	errHandler       func(err error) bool
@@ -27,7 +28,7 @@ func WithPullMaxMessages(maximum int32) PullJobOption {
 
 func WithPullLongPullDuration(milliseconds int32) PullJobOption {
 	return func(b *pullJob) {
-		b.longPullDuration = &milliseconds
+		b.longPullDuration = milliseconds
 	}
 }
 
@@ -53,55 +54,59 @@ func WithErrorHandler(handler func(err error) bool) PullJobOption {
 	}
 }
 
-func newPullJob(s *Subscriber, opts []PullJobOption) *pullJob {
-	b := &pullJob{
-		subscription:    s,
-		maxPullMessages: 10,
-		interval:        time.Second * 5,
-		bufferSize:      0,
+func newPullJob(s *Subscriber, opts []PullJobOption) (*pullJob, error) {
+	job := &pullJob{
+		subscription:     s,
+		maxPullMessages:  10,
+		interval:         1,
+		longPullDuration: 5000,
+		bufferSize:       0,
 		errHandler: func(err error) bool {
 			s.logger.Error(err, "fatal background error")
 			return true
 		},
 	}
 
 	for _, opt := range opts {
-		opt(b)
+		opt(job)
 	}
 
-	return b
+	if job.interval < 1 {
+		return nil, &ConfigurationError{
+			Msg: fmt.Sprintf("interval must be at least set to 1, got %d", job.interval),
+		}
+	}
+
+	return job, nil
 }
 
-func (b *pullJob) runLoop(ctx context.Context, handler func(context.Context, PullMessages)) {
-	ticker := time.NewTicker(b.interval)
+func (j *pullJob) runLoop(ctx context.Context, handler func(context.Context, PullMessages)) {
+	ticker := time.NewTicker(j.interval)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			pullOpts := []PullOption{WithMaxMessages(b.maxPullMessages)}
-			if b.longPullDuration != nil {
-				pullOpts = append(pullOpts, WithLongPullDuration(*b.longPullDuration))
-			}
-			messages, err := b.subscription.Pull(ctx, pullOpts...)
+			pullOpts := []PullOption{WithMaxMessages(j.maxPullMessages), WithLongPullDuration(j.longPullDuration)}
+			messages, err := j.subscription.Pull(ctx, pullOpts...)
 			if err != nil { //nolint:nestif
 				var sdkErr SDKError          // Declare the target variable
 				if errors.As(err, &sdkErr) { // Pass a pointer to sdkErr
 					if !sdkErr.IsTransient() {
 						// Only exit the loop if the users error handler returns false
-						if !b.errHandler(err) {
+						if !j.errHandler(err) {
 							return
 						}
 						continue
 					}
 
-					b.subscription.logger.Error(err, "transient error, retrying")
+					j.subscription.logger.Error(err, "transient error, retrying")
 					continue
 				}
 
-				b.subscription.logger.Error(err, "unknown error, retrying")
+				j.subscription.logger.Error(err, "unknown error, retrying")
 				continue
 			}
 
@@ -121,7 +126,11 @@ func (s *Subscriber) PullJobCallback(
 		return ErrMissingCallback
 	}
 
-	job := newPullJob(s, opts)
+	job, err := newPullJob(s, opts)
+	if err != nil {
+		return err
+	}
+
 	s.wg.Go(func() {
 		job.runLoop(ctx, callback)
 	})
@@ -135,7 +144,11 @@ func (s *Subscriber) PullJobCallback(
 }
 
 func (s *Subscriber) PullJobChan(ctx context.Context, opts ...PullJobOption) (<-chan PullMessages, error) {
-	job := newPullJob(s, opts)
+	job, err := newPullJob(s, opts)
+	if err != nil {
+		return nil, err
+	}
+
 	out := make(chan PullMessages, job.bufferSize)
 
 	s.wg.Go(func() {

pkg/pubsub/subscriber.go

@@ -23,11 +23,10 @@ type Subscriber struct {
 	wg             sync.WaitGroup
 }
 
-// NewSubscriber instantiates a new Subscriber. It returns an error if the underlying
-// API dataplane client fails to initialize.
+// NewSubscriber instantiates a new Subscriber.
 func NewSubscriber(topicID uuid.UUID, subscriptionID uuid.UUID, opts ...Option) *Subscriber {
 	cfg := &clientConfig{
-		httpClient: http.DefaultClient,
+		httpClient: &http.Client{},
 		host:       "pubsub.eu01.onstackit.cloud",
 		logger:     logr.FromSlogHandler(slog.Default().Handler()),
 	}
@@ -112,7 +111,7 @@ func toSDKMessages(m []api.Message, subscription *Subscriber) PullMessages {
 
 type pullOptions struct {
 	maxMessages      int32
-	longPullDuration *int32
+	longPullDuration int32
 }
 
 type PullOption func(*pullOptions)
@@ -125,33 +124,29 @@ func WithMaxMessages(maximum int32) PullOption {
 
 func WithLongPullDuration(milliseconds int32) PullOption {
 	return func(opts *pullOptions) {
-		opts.longPullDuration = &milliseconds
+		opts.longPullDuration = milliseconds
 	}
 }
 
 func (s *Subscriber) Pull(ctx context.Context, opts ...PullOption) (PullMessages, error) {
 	cfg := &pullOptions{
-		maxMessages: 32,
+		maxMessages:      32,
+		longPullDuration: 100,
 	}
 
 	for _, opt := range opts {
 		opt(cfg)
 	}
 
-	var longPullDuration *int32
-	if cfg.longPullDuration != nil && *cfg.longPullDuration != 0 {
-		ms := *cfg.longPullDuration
-		if ms < 100 || ms > 5000 {
-			return nil, &ConfigurationError{
-				Msg: fmt.Sprintf("long_pull_duration must be 0 (default) or between 100–5000, got %d", ms),
-			}
+	if cfg.longPullDuration < 100 || cfg.longPullDuration > 5000 {
+		return nil, &ConfigurationError{
+			Msg: fmt.Sprintf("long_pull_duration must be between 100–5000, got %d", cfg.longPullDuration),
 		}
-		longPullDuration = &ms
 	}
 
 	reqBody := api.PullMessagesParams{
 		PubSubMaxMessages:      &cfg.maxMessages,
-		PubSubLongPullDuration: longPullDuration,
+		PubSubLongPullDuration: &cfg.longPullDuration,
 	}
 
 	s.logger.V(4).Info("pulling messages", "max_messages", int(cfg.maxMessages))

pkg/pubsub/subscriber_test.go

@@ -86,7 +86,6 @@ var _ = Describe("WithLongPullDuration validation", func() {
 				Expect(errors.As(err, &cfgErr)).To(BeFalse(), "expected no ConfigurationError for ms=%d", ms)
 			}
 		},
-		Entry("disabled (0)", int32(0)),
 		Entry("minimum (100)", int32(100)),
 		Entry("maximum (5000)", int32(5000)),
 	)
@@ -132,6 +131,8 @@ var _ = Describe("PullJob", func() {
 		err := publisher.Purge(ctx)
 		Expect(err).ToNot(HaveOccurred())
 
+		time.Sleep(1 * time.Second) // wait for topic to be purged
+
 		// publishing test messages
 		_, err = publisher.PublishStrings(ctx, "testMessage")
 		Expect(err).ToNot(HaveOccurred())
@@ -140,36 +141,14 @@ var _ = Describe("PullJob", func() {
 	Context("using PullJobChan", func() {
 		It("should receive a message from channel", func(ctx context.Context) {
 			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
+			defer subscriber.Wait()
 
-			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-			defer cancel()
-
-			jobChan, err := subscriber.PullJobChan(ctx, pubsub.WithInterval(100*time.Millisecond))
-			Expect(err).ToNot(HaveOccurred())
-
-			var receivedMessages pubsub.PullMessages
-			Eventually(jobChan, "5s").Should(Receive(&receivedMessages))
-			Expect(receivedMessages).To(HaveLen(1))
-
-			decodedString, err := receivedMessages[0].DecodeString()
-			Expect(err).ToNot(HaveOccurred())
-			Expect(decodedString).To(Equal("testMessage"))
-			err = subscriber.Ack(ctx, receivedMessages.AckIDs())
-			Expect(err).ToNot(HaveOccurred())
-
-			cancel()
-			subscriber.Wait()
-		})
-
-		It("should receive a message from channel with long pull duration set", func(ctx context.Context) {
-			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-
-			ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
 			defer cancel()
 
 			jobChan, err := subscriber.PullJobChan(ctx,
-				pubsub.WithInterval(100*time.Millisecond),
-				pubsub.WithPullLongPullDuration(500),
+				pubsub.WithPullLongPullDuration(100),
+				pubsub.WithPullMaxMessages(1),
 			)
 			Expect(err).ToNot(HaveOccurred())
 
@@ -184,15 +163,14 @@ var _ = Describe("PullJob", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			cancel()
-			subscriber.Wait()
 		})
 	})
 
 	Context("using PullJobCallback", func() {
 		It("should invoke the callback with messages", func(ctx context.Context) {
 			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
 			defer subscriber.Wait()
-			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
 			defer cancel()
 
 			var callbackInvoked atomic.Bool // check if callback was invoked
@@ -211,8 +189,8 @@ var _ = Describe("PullJob", func() {
 			err := subscriber.PullJobCallback(
 				ctx,
 				callback,
-				pubsub.WithInterval(100*time.Millisecond),
 				pubsub.WithPullMaxMessages(1),
+				pubsub.WithPullLongPullDuration(100),
 			)
 			Expect(err).ToNot(HaveOccurred())