feat: added long pull to sdk

api/openapi.yaml

@@ -105,6 +105,16 @@ paths:
             minimum: 1
             maximum: 32
             default: 16
+        - name: PubSub-Long-Pull-Duration
+          in: header
+          description: The maximum amount of milliseconds to keep an connection open, awaiting a new message on the server.
+          required: false
+          schema:
+            type: integer
+            format: int32
+            minimum: 100
+            maximum: 5000
+            default: 5000
       responses:
         '200':
           description: Messages pulled successfully.

pkg/pubsub/pulljob.go

@@ -7,11 +7,12 @@ import (
 )
 
 type pullJob struct {
-	subscription    *Subscriber
-	maxPullMessages int32
-	interval        time.Duration
-	bufferSize      int
-	errHandler      func(err error) bool
+	subscription     *Subscriber
+	maxPullMessages  int32
+	longPullDuration *int32
+	interval         time.Duration
+	bufferSize       int
+	errHandler       func(err error) bool
 }
 
 var ErrMissingCallback = NewConfigurationError("callback function is required", nil)
@@ -24,6 +25,12 @@ func WithPullMaxMessages(maximum int32) PullJobOption {
 	}
 }
 
+func WithPullLongPullDuration(milliseconds int32) PullJobOption {
+	return func(b *pullJob) {
+		b.longPullDuration = &milliseconds
+	}
+}
+
 func WithInterval(interval time.Duration) PullJobOption {
 	return func(b *pullJob) {
 		b.interval = interval
@@ -74,7 +81,11 @@ func (b *pullJob) runLoop(ctx context.Context, handler func(context.Context, Pul
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			messages, err := b.subscription.Pull(ctx, WithMaxMessages(b.maxPullMessages))
+			pullOpts := []PullOption{WithMaxMessages(b.maxPullMessages)}
+			if b.longPullDuration != nil {
+				pullOpts = append(pullOpts, WithLongPullDuration(*b.longPullDuration))
+			}
+			messages, err := b.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

pkg/pubsub/subscriber.go

@@ -121,7 +121,8 @@ func toSdkMessages(m []api.Message, subscription *Subscriber) PullMessages {
 }
 
 type pullOptions struct {
-	maxMessages int32
+	maxMessages      int32
+	longPullDuration *int32
 }
 
 type PullOption func(*pullOptions)
@@ -132,6 +133,12 @@ func WithMaxMessages(maximum int32) PullOption {
 	}
 }
 
+func WithLongPullDuration(milliseconds int32) PullOption {
+	return func(opts *pullOptions) {
+		opts.longPullDuration = &milliseconds
+	}
+}
+
 func (s *Subscriber) Pull(ctx context.Context, opts ...PullOption) (PullMessages, error) {
 	cfg := &pullOptions{
 		maxMessages: 64,
@@ -141,8 +148,21 @@ func (s *Subscriber) Pull(ctx context.Context, opts ...PullOption) (PullMessages
 		opt(cfg)
 	}
 
+	// 0 and nil both mean disabled; any other value must be in [100, 5000].
+	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),
+			}
+		}
+		longPullDuration = &ms
+	}
+
 	reqBody := api.PullMessagesParams{
-		PubSubMaxMessages: &cfg.maxMessages,
+		PubSubMaxMessages:      &cfg.maxMessages,
+		PubSubLongPullDuration: longPullDuration,
 	}
 
 	s.logger.V(4).Info("pulling messages",

pkg/pubsub/subscriber_test.go

@@ -2,6 +2,7 @@ package pubsub_test
 
 import (
 	"context"
+	"errors"
 	"sync/atomic"
 	"time"
 
@@ -64,6 +65,34 @@ var _ = Describe("Acknowledge messages", func() {
 	})
 })
 
+var _ = Describe("WithLongPullDuration validation", func() {
+	DescribeTable("invalid durations return ConfigurationError",
+		func(ctx context.Context, ms int32) {
+			subscriber := pubsub.NewSubscriber(topicId, subscriptionId)
+			_, err := subscriber.Pull(ctx, pubsub.WithLongPullDuration(ms))
+			Expect(err).To(HaveOccurred())
+			var cfgErr *pubsub.ConfigurationError
+			Expect(errors.As(err, &cfgErr)).To(BeTrue())
+		},
+		Entry("below minimum", int32(50)),
+		Entry("above maximum", int32(6000)),
+	)
+
+	DescribeTable("valid durations do not return ConfigurationError",
+		func(ms int32) {
+			subscriber := pubsub.NewSubscriber(topicId, subscriptionId)
+			_, err := subscriber.Pull(context.Background(), pubsub.WithLongPullDuration(ms))
+			if err != nil {
+				var cfgErr *pubsub.ConfigurationError
+				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)),
+	)
+})
+
 var _ = Describe("Pull messages", func() {
 	Context("pulling messages", func() {
 		BeforeEach(func(ctx context.Context) {
@@ -80,6 +109,12 @@ var _ = Describe("Pull messages", func() {
 			Expect(resp).ToNot(BeNil())
 			Expect(err).ToNot(HaveOccurred())
 		})
+		It("should pull messages with long pull duration set", func(ctx context.Context) {
+			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
+			resp, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(128), pubsub.WithLongPullDuration(500))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(resp).ToNot(BeNil())
+		})
 		It(
 			"should pull only one Message, MaxMessages is set to 1 but more messages will be available",
 			func(ctx context.Context) {
@@ -129,6 +164,32 @@ var _ = Describe("PullJob", func() {
 			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)
+			defer cancel()
+
+			jobChan, err := subscriber.PullJobChan(ctx,
+				pubsub.WithInterval(100*time.Millisecond),
+				pubsub.WithPullLongPullDuration(500),
+			)
+			Expect(err).ToNot(HaveOccurred())
+
+			var receivedMessages pubsub.PullMessages
+			Eventually(jobChan, "10s").Should(Receive(&receivedMessages))
+			Expect(receivedMessages).To(HaveLen(1))
+
+			decodedStrings, err := pubsub.Base64ToStrings(string(receivedMessages[0].Data))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(decodedStrings[0]).To(Equal("testMessage"))
+			err = subscriber.Ack(ctx, receivedMessages.GetAckIDs())
+			Expect(err).ToNot(HaveOccurred())
+
+			cancel()
+			subscriber.Wait()
+		})
 	})
 
 	Context("using PullJobCallback", func() {