diff --git a/api/openapi.yaml b/api/openapi.yaml index 160cd3e..3537b1c 100644 --- a/api/openapi.yaml +++ b/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. diff --git a/pkg/pubsub/api/sdk.gen.go b/pkg/pubsub/api/sdk.gen.go index 5f49317..15855fc 100644 --- a/pkg/pubsub/api/sdk.gen.go +++ b/pkg/pubsub/api/sdk.gen.go @@ -77,6 +77,9 @@ type SubscriptionId = openapi_types.UUID type PullMessagesParams struct { // PubSubMaxMessages The maximum number of messages to pull. PubSubMaxMessages *int32 `json:"PubSub-Max-Messages,omitempty"` + + // PubSubLongPullDuration The maximum amount of milliseconds to keep an connection open, awaiting a new message on the server. + PubSubLongPullDuration *int32 `json:"PubSub-Long-Pull-Duration,omitempty"` } // PublishMessagesJSONRequestBody defines body for PublishMessages for application/json ContentType. @@ -484,6 +487,17 @@ func NewPullMessagesRequest(server string, subscriptionId SubscriptionId, params req.Header.Set("PubSub-Max-Messages", headerParam0) } + if params.PubSubLongPullDuration != nil { + var headerParam1 string + + headerParam1, err = runtime.StyleParamWithOptions("simple", false, "PubSub-Long-Pull-Duration", *params.PubSubLongPullDuration, runtime.StyleParamOptions{ParamLocation: runtime.ParamLocationHeader, Type: "integer", Format: "int32"}) + if err != nil { + return nil, err + } + + req.Header.Set("PubSub-Long-Pull-Duration", headerParam1) + } + } return req, nil diff --git a/pkg/pubsub/pulljob.go b/pkg/pubsub/pulljob.go index 75f83ff..48bc52a 100644 --- a/pkg/pubsub/pulljob.go +++ b/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 diff --git a/pkg/pubsub/subscriber.go b/pkg/pubsub/subscriber.go index f4b47af..bed8b47 100644 --- a/pkg/pubsub/subscriber.go +++ b/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", diff --git a/pkg/pubsub/subscriber_test.go b/pkg/pubsub/subscriber_test.go index 7ed8412..6ac7d23 100644 --- a/pkg/pubsub/subscriber_test.go +++ b/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() {