From 021b5796323cba8076e1ad4d4773f644bba6ff49 Mon Sep 17 00:00:00 2001
From: Selina Fehn <selina.fehn@stackit.cloud>
Date: Thu, 6 Aug 2026 14:30:38 +0200
Subject: [PATCH 01/10] fix: fix api access call

---
 scripts/create-pubsub-resources.sh | 21 +++++++++++++--------
 1 file changed, 13 insertions(+), 8 deletions(-)

diff --git a/scripts/create-pubsub-resources.sh b/scripts/create-pubsub-resources.sh
index 282533c..c072064 100755
--- a/scripts/create-pubsub-resources.sh
+++ b/scripts/create-pubsub-resources.sh
@@ -88,29 +88,34 @@ done
 
 #ACCESS
 echo "Granting Publisher Access via curl (Targeting: $REGION)..."
-GPARESPONSE=$(curl -sk -w "\n%{http_code}" -X PUT "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/publishers/$PUBLISHER_MAIL" \
-  -H "Authorization: Bearer $TOKEN" \
+PUBLISHER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/publishers" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
 
+  HTTP_STATUS=$(echo "$PUBLISHER_RESPONSE" | tail -n 1)
+  PUBLISHER_BODY=$(echo "$PUBLISHER_RESPONSE" | sed '$d')
+  echo "Response Body: $PUBLISHER_BODY"
+
 if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
-  echo "API Error (HTTP $HTTP_STATUS)"
-  echo "Response Granting Publisher Access Body: $GPARESPONSE"
+  echo "::error file=scripts/create-pubsub-resources.sh::Failed to grant publisher access (HTTP $HTTP_STATUS) - Response: $PUBLISHER_BODY"
   exit 1
 fi
 
 echo "Response SUBSCRIPTION Body: $GPARESPONSE"
 
 echo "Granting Subscriber Access via curl (Targeting: $REGION)..."
-GSARESPONSE=$(curl -sk -w "\n%{http_code}" -X PUT "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/$SUBSCRIPTION_ID/subscribers/$PUBLISHER_MAIL" \
-  -H "Authorization: Bearer $TOKEN" \
+SUBSCRIBER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}/subscribers" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
 
+HTTP_STATUS=$(echo "$SUBSCRIBER_RESPONSE" | tail -n 1)
+SUBSCRIBER_BODY=$(echo "$SUBSCRIBER_RESPONSE" | sed '$d')
+echo "Response Body: $SUBSCRIBER_BODY"
 
 if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
-  echo "API Error (HTTP $HTTP_STATUS)"
-  echo "Response Granting Subscriber Access Body: $GSARESPONSE"
+  echo "::error file=scripts/create-pubsub-resources.sh::Failed to grant subscriber access (HTTP $HTTP_STATUS) - Response: $SUBSCRIBER_BODY"
   exit 1
 fi
 

From 1c85977060cdbaaa97164fcd652222d635812baa Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Paul=20Gro=C3=9Fmann?= <paul.grossmann@digits.schwarz>
Date: Mon, 1 Jun 2026 14:43:41 +0200
Subject: [PATCH 02/10] =?UTF-8?q?feat:=20added=20test=20for=20pullJobChan?=
 =?UTF-8?q?=20and=20simplified=20before=20each=20in=20PullJo=E2=80=A6=20(#?=
 =?UTF-8?q?15)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* feat: added test for pullJobChan and simplified before each in PullJob tests

* refactor: made reading from PullJobChan safer by using gomega assertion
---
 pkg/pubsub/subscriber_test.go | 48 +++++++++++++----------------------
 1 file changed, 18 insertions(+), 30 deletions(-)

diff --git a/pkg/pubsub/subscriber_test.go b/pkg/pubsub/subscriber_test.go
index 3f303f1..7ed8412 100644
--- a/pkg/pubsub/subscriber_test.go
+++ b/pkg/pubsub/subscriber_test.go
@@ -94,28 +94,15 @@ var _ = Describe("Pull messages", func() {
 
 var _ = Describe("PullJob", func() {
 	BeforeEach(func(ctx context.Context) {
-		// making sure everything is empty, and acking everything, stopping when len = 0
-		subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-
-		Eventually(func(g Gomega) bool {
-			msgs, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(13))
-			g.Expect(err).ToNot(HaveOccurred())
-
-			if len(msgs) == 0 {
-				return true
-			}
-
-			err = subscriber.Ack(ctx, msgs.GetAckIDs())
-			g.Expect(err).ToNot(HaveOccurred())
-
-			return false
-		}).WithTimeout(10 * time.Second).WithPolling(500 * time.Millisecond).Should(BeTrue())
+		publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
+		// making sure everything is empty
+		err := publisher.Purge(ctx)
+		Expect(err).ToNot(HaveOccurred())
 
 		// publishing test messages
-		publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-		messagesToPublish := pubsub.StringsToBase64("testMessage", "testMessage2")
+		messagesToPublish := pubsub.StringsToBase64("testMessage")
 
-		_, err := publisher.Publish(ctx, messagesToPublish)
+		_, err = publisher.Publish(ctx, messagesToPublish)
 		Expect(err).ToNot(HaveOccurred())
 	})
 
@@ -123,23 +110,24 @@ var _ = Describe("PullJob", func() {
 		It("should receive a message from channel", func(ctx context.Context) {
 			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
 
-			var receivedMessages pubsub.PullMessages
-			Eventually(func(g Gomega) {
-				msgs, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(1))
-				g.Expect(err).ToNot(HaveOccurred())
-				if len(msgs) > 0 {
-					receivedMessages = msgs
-				}
-				g.Expect(receivedMessages).ToNot(BeEmpty())
-			}).WithContext(ctx).Should(Succeed())
+			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))
 
 			decodedStrings, err := pubsub.Base64ToStrings(string(receivedMessages[0].Data))
 			Expect(err).ToNot(HaveOccurred())
-			Expect(decodedStrings[0]).To(Or(Equal("testMessage"), Equal("testMessage2")))
+			Expect(decodedStrings[0]).To(Equal("testMessage"))
 			err = subscriber.Ack(ctx, receivedMessages.GetAckIDs())
 			Expect(err).ToNot(HaveOccurred())
+
+			cancel()
+			subscriber.Wait()
 		})
 	})
 
@@ -156,7 +144,7 @@ var _ = Describe("PullJob", func() {
 				Expect(messages).To(HaveLen(1))
 				decoded, err := pubsub.Base64ToStrings(string(messages[0].Data))
 				Expect(err).ToNot(HaveOccurred())
-				Expect(decoded[0]).To(Or(Equal("testMessage"), Equal("testMessage2")))
+				Expect(decoded[0]).To(Equal("testMessage"))
 				err = subscriber.Ack(ctx, messages.GetAckIDs())
 				Expect(err).ToNot(HaveOccurred())
 				callbackInvoked.Store(true)

From de338a4f056c779657aff40f39be84a9db36c7ca Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Paul=20Gro=C3=9Fmann?= <paul.grossmann@digits.schwarz>
Date: Mon, 1 Jun 2026 14:54:43 +0200
Subject: [PATCH 03/10] fix: corrected examples to properly exit on error (#16)

---
 README.md                  |  2 +-
 example/example_publish.go |  5 ++---
 example/example_pull.go    | 14 +++++++++++---
 example/example_purge.go   |  7 ++++---
 4 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/README.md b/README.md
index c53a4ff..9d8fafc 100644
--- a/README.md
+++ b/README.md
@@ -35,7 +35,7 @@ roundTripper, err := auth.DefaultAuth(&config.Configuration{
     ServiceAccountKeyPath: "./service-account-key.json",
 })
 if err != nil {
-    log.Printf("Error creating authentication token: %v", err)
+    log.Fatalf("Error creating authentication token: %v", err)
 }
 
 publisher := pubsub.NewPublisher(topicID,
diff --git a/example/example_publish.go b/example/example_publish.go
index edd52cc..11fe16e 100644
--- a/example/example_publish.go
+++ b/example/example_publish.go
@@ -12,13 +12,12 @@ import (
 
 //nolint:all
 func publish() {
-
 	// Authentication with STACKIT SDK - Returns a Round-Tripper
 	rt, err := auth.DefaultAuth(&config.Configuration{
 		ServiceAccountKeyPath: "./service-account-key.json",
 	})
 	if err != nil {
-		log.Printf("Error creating authentication token: %v", err)
+		log.Fatalf("Error creating authentication token: %v", err)
 	}
 
 	// Setup your Topic ID
@@ -39,7 +38,7 @@ func publish() {
 		message,
 	)
 	if err != nil {
-		log.Printf("Error publishing message: %v", err)
+		log.Fatalf("Error publishing message: %v", err)
 	}
 
 	log.Print("Successfully published messages")
diff --git a/example/example_pull.go b/example/example_pull.go
index 0a331e5..fbe2d88 100644
--- a/example/example_pull.go
+++ b/example/example_pull.go
@@ -12,13 +12,12 @@ import (
 
 //nolint:all
 func pull() {
-
 	// Authentication with STACKIT SDK - Returns a Round-Tripper
 	rt, err := auth.DefaultAuth(&config.Configuration{
 		ServiceAccountKeyPath: "./service-account-key.json",
 	})
 	if err != nil {
-		log.Printf("Error creating authentication token: %v", err)
+		log.Fatalf("Error creating authentication token: %v", err)
 	}
 
 	// Setup your TopicID and Subscription ID
@@ -33,15 +32,24 @@ func pull() {
 	)
 
 	// Pull messages via subscription
-	pulledMessages, _ := subscriber.Pull(context.Background(), pubsub.WithMaxMessages(10))
+	pulledMessages, err := subscriber.Pull(context.Background(), pubsub.WithMaxMessages(10))
+	if err != nil {
+		log.Fatalf("Error pulling messages: %v", err)
+	}
 
 	log.Printf("Successfully pulled message: %v", pulledMessages)
 
 	// Get your AckIDs and acknowledge them
 	ackIDs := pulledMessages.GetAckIDs()
 	err = subscriber.Ack(context.Background(), ackIDs)
+	if err != nil {
+		log.Fatalf("Error ack ids: %v", err)
+	}
 
 	// Get your NackIDs and not acknowledge them
 	nackIDs := pulledMessages.GetAckIDs()
 	err = subscriber.Nack(context.Background(), nackIDs)
+	if err != nil {
+		log.Fatalf("Error nack ids: %v", err)
+	}
 }
diff --git a/example/example_purge.go b/example/example_purge.go
index 221e96e..c38cfe4 100644
--- a/example/example_purge.go
+++ b/example/example_purge.go
@@ -12,13 +12,12 @@ import (
 
 //nolint:all
 func purge() {
-
 	// Authentication with STACKIT SDK - Returns a Round-Tripper
 	rt, err := auth.DefaultAuth(&config.Configuration{
 		ServiceAccountKeyPath: "./service-account-key.json",
 	})
 	if err != nil {
-		log.Printf("Error creating authentication token: %v", err)
+		log.Fatalf("Error creating authentication token: %v", err)
 	}
 
 	// Setup your TopicID and Subscription ID
@@ -31,5 +30,7 @@ func purge() {
 	)
 
 	err = publisher.Purge(context.Background())
-
+	if err != nil {
+		log.Fatalf("Error purging topic: %v", err)
+	}
 }

From 33b04e5c7b966c76bd8840a3023253928b63abed Mon Sep 17 00:00:00 2001
From: selinafehn <144441353+selinafehn@users.noreply.github.com>
Date: Wed, 3 Jun 2026 09:45:17 +0200
Subject: [PATCH 04/10] Rename token in release pipeline (#14)

* fix: change token

---------

Co-authored-by: Selina Fehn <selina.fehn@stackit.cloud>
---
 .github/workflows/release.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 2266802..96bce08 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -22,7 +22,7 @@ jobs:
 
       - name: Create GitHub Release
         env:
-          GITHUB_TOKEN: ${{ secrets.TOKEN }}
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: |
           gh release create ${{ inputs.version }} \
             --title "Release ${{ inputs.version }}" \

From 566259c4e69f3eef58264faa06d57c36767e6413 Mon Sep 17 00:00:00 2001
From: Jennifer Linnenberg <ferdinand_paul_martin.linnenberg@digits.schwarz>
Date: Mon, 29 Jun 2026 11:20:24 +0200
Subject: [PATCH 05/10] feat: added long pull to sdk (#17)

* feat: added long pull to sdk :)

* fix: add 1-99 check (server doesn't accept these)

* fix: clarified error

* between -> be
---
 api/openapi.yaml              | 10 ++++++
 pkg/pubsub/api/sdk.gen.go     | 14 ++++++++
 pkg/pubsub/pulljob.go         | 23 +++++++++----
 pkg/pubsub/subscriber.go      | 24 ++++++++++++--
 pkg/pubsub/subscriber_test.go | 61 +++++++++++++++++++++++++++++++++++
 5 files changed, 124 insertions(+), 8 deletions(-)

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() {

From c1f3eb35e9541a0376a8f534498333af75f2baaa Mon Sep 17 00:00:00 2001
From: Jennifer Linnenberg <ferdinand_paul_martin.linnenberg@digits.schwarz>
Date: Wed, 1 Jul 2026 17:44:04 +0200
Subject: [PATCH 06/10] Update samples for better UX (#18)

* fix. updated samples

* feat: update readme for new methods

* fix 0 -> i

* save

* feat: renamed funcs
---
 README.md                     | 17 +++++---
 example/example_publish.go    |  7 +---
 example/example_pull.go       | 11 ++++-
 go.mod                        |  2 +-
 mise.toml                     |  2 +-
 pkg/pubsub/helper.go          | 32 +++++++--------
 pkg/pubsub/message.go         | 36 +++++++++++++++--
 pkg/pubsub/publisher.go       | 57 +++++++++++++++-----------
 pkg/pubsub/publisher_test.go  |  6 +--
 pkg/pubsub/subscriber.go      | 76 ++++++++++++++---------------------
 pkg/pubsub/subscriber_test.go | 30 ++++++--------
 11 files changed, 150 insertions(+), 126 deletions(-)

diff --git a/README.md b/README.md
index 9d8fafc..127c485 100644
--- a/README.md
+++ b/README.md
@@ -83,10 +83,7 @@ To consume messages, you Pull them, process them, and `Ack` (acknowledge) them.
 topicID := uuid.MustParse("00000000-0000-0000-0000-000000000000")
 publisher := pubsub.NewPublisher(topicID, pubsub.WithHTTPRoundTripper(roundTripper))
 
-messages := [][]byte{
-    []byte("Hello, PubSub!"),
-}
-messageIDs, err := publisher.Publish(ctx, messages)
+messageIDs, err := publisher.PublishStrings(ctx, "Hello, PubSub!")
 
 // Pull
 subscriptionID := uuid.MustParse("00000000-0000-0000-0000-000000000000")
@@ -94,8 +91,16 @@ subscriber := pubsub.NewSubscriber(topicID, subscriptionID, pubsub.WithHTTPRound
 
 pulledMessages, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(10))
 
+for i := 0; i < len(pulledMessages); i++ {
+    msg, err := pulledMessages[i].DecodeString()
+    if err != nil {
+        log.Fatalf("Error converting message to string: %v", err)
+    }
+    log.Printf("Message [%d]: %s:", i, msg)
+}
+
 // Acknowledge
-ackIDs := pulledMessages.GetAckIDs()
+ackIDs := pulledMessages.AckIDs()
 err = subscriber.Ack(ctx, ackIDs)
 ```
 
@@ -107,7 +112,7 @@ The SDK returns specific error types to help you handle different failure scenar
 
 ```go
 // Example of detailed error handling
-_, err := publisher.Publish(ctx, messages)
+_, err := publisher.PublishStrings(ctx, "Hello, PubSub!")
 if err != nil {
     var apiErr *pubsub.APIError
     if errors.As(err, &apiErr) {
diff --git a/example/example_publish.go b/example/example_publish.go
index 11fe16e..eed7895 100644
--- a/example/example_publish.go
+++ b/example/example_publish.go
@@ -29,13 +29,10 @@ func publish() {
 		pubsub.WithHTTPRoundTripper(rt),
 	)
 
-	// Create a message to publish to the topic and encode it to base64 format
-	message := pubsub.StringsToBase64("Hello PubSub from example", "This is another message")
-
 	// Publish the messages to the topic using the publisher client
-	_, err = publisher.Publish(
+	_, err = publisher.PublishStrings(
 		context.Background(),
-		message,
+		"Hello PubSub from example", "This is another message",
 	)
 	if err != nil {
 		log.Fatalf("Error publishing message: %v", err)
diff --git a/example/example_pull.go b/example/example_pull.go
index fbe2d88..79e80e5 100644
--- a/example/example_pull.go
+++ b/example/example_pull.go
@@ -38,16 +38,23 @@ func pull() {
 	}
 
 	log.Printf("Successfully pulled message: %v", pulledMessages)
+	for i := 0; i < len(pulledMessages); i++ {
+		msg, err := pulledMessages[i].DecodeString()
+		if err != nil {
+			log.Fatalf("Error converting message to string: %v", err)
+		}
+		log.Printf("Message [%d]: %s:", i, msg)
+	}
 
 	// Get your AckIDs and acknowledge them
-	ackIDs := pulledMessages.GetAckIDs()
+	ackIDs := pulledMessages.AckIDs()
 	err = subscriber.Ack(context.Background(), ackIDs)
 	if err != nil {
 		log.Fatalf("Error ack ids: %v", err)
 	}
 
 	// Get your NackIDs and not acknowledge them
-	nackIDs := pulledMessages.GetAckIDs()
+	nackIDs := pulledMessages.AckIDs()
 	err = subscriber.Nack(context.Background(), nackIDs)
 	if err != nil {
 		log.Fatalf("Error nack ids: %v", err)
diff --git a/go.mod b/go.mod
index fc9b8e2..6ec98f7 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/stackitcloud/pubsub-sdk-go
 
-go 1.25.0 // Make sure this matches the version specified in the mise.toml
+go 1.25.11 // Make sure this matches the version specified in the mise.toml
 
 require (
 	github.com/go-logr/logr v1.4.3
diff --git a/mise.toml b/mise.toml
index 280e661..ebb4e2d 100644
--- a/mise.toml
+++ b/mise.toml
@@ -1,5 +1,5 @@
 [tools]
-go = "1.25.0" # Make sure this matches the go.mod
+go = "1.25.11" # Make sure this matches the go.mod
 "go:gitlab.com/backbone/changelog/cmd/changelog" = "v1.1.0"
 ginkgo = "2.28.1"
 "go:github.com/axw/gocov/gocov" = "v1.1.0"
diff --git a/pkg/pubsub/helper.go b/pkg/pubsub/helper.go
index c33a00e..d6c848f 100644
--- a/pkg/pubsub/helper.go
+++ b/pkg/pubsub/helper.go
@@ -2,29 +2,25 @@ package pubsub
 
 import (
 	"encoding/base64"
-	"fmt"
 )
 
-func StringsToBase64(strings ...string) [][]byte {
-	result := make([][]byte, len(strings))
-
-	for i, string := range strings {
-		encodedStrings := base64.StdEncoding.EncodeToString([]byte(string))
-		result[i] = []byte(encodedStrings)
+// bytesToBase64 encodes multiple raw byte slices into base64 byte slice.
+func bytesToBase64(messages ...[]byte) [][]byte {
+	result := make([][]byte, len(messages))
+	for i, msg := range messages {
+		dst := make([]byte, base64.StdEncoding.EncodedLen(len(msg)))
+		base64.StdEncoding.Encode(dst, msg)
+		result[i] = dst
 	}
 	return result
 }
 
-func Base64ToStrings(encodedStrings ...string) ([]string, error) {
-	result := make([]string, len(encodedStrings))
-
-	for i, s := range encodedStrings {
-		decodedBytes, err := base64.StdEncoding.DecodeString(s)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode string at index %d: %w", i, err)
-		}
-		result[i] = string(decodedBytes)
+// base64Decode safely decodes a single base64-encoded byte slice.
+func base64Decode(src []byte) ([]byte, error) {
+	dst := make([]byte, base64.StdEncoding.DecodedLen(len(src)))
+	n, err := base64.StdEncoding.Decode(dst, src)
+	if err != nil {
+		return nil, err
 	}
-
-	return result, nil
+	return dst[:n], nil
 }
diff --git a/pkg/pubsub/message.go b/pkg/pubsub/message.go
index 1f860c7..ef4edf1 100644
--- a/pkg/pubsub/message.go
+++ b/pkg/pubsub/message.go
@@ -2,6 +2,7 @@ package pubsub
 
 import (
 	"context"
+	"fmt"
 	"time"
 )
 
@@ -14,8 +15,6 @@ type PullMessage struct {
 	DeliveryAttempts uint64
 }
 
-type PullMessages []PullMessage
-
 func (m *PullMessage) Ack(ctx context.Context) error {
 	return m.subscription.Ack(ctx, []string{m.AckID})
 }
@@ -24,10 +23,41 @@ func (m *PullMessage) Nack(ctx context.Context) error {
 	return m.subscription.Nack(ctx, []string{m.AckID})
 }
 
-func (m PullMessages) GetAckIDs() []string {
+type PullMessages []PullMessage
+
+// AckIDs extracts all AckIDs cleanly from a slice of PullMessages.
+func (m PullMessages) AckIDs() []string {
 	ids := make([]string, len(m))
 	for i, msg := range m {
 		ids[i] = msg.AckID
 	}
 	return ids
 }
+
+// DecodeString reverses the transparent base64 encoding, returning the cleartext string.
+func (m *PullMessage) DecodeString() (string, error) {
+	decoded, err := base64Decode(m.Data)
+	if err != nil {
+		return "", fmt.Errorf("failed to decode message data: %w", err)
+	}
+	return string(decoded), nil
+}
+
+// DecodeStrings decodes an entire slice of messages.
+// If any single message is corrupt, it returns the error immediately instead of swallowing it.
+func (m PullMessages) DecodeStrings() ([]string, error) {
+	strings := make([]string, len(m))
+	for i, msg := range m {
+		str, err := msg.DecodeString()
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode message at index %d: %w", i, err)
+		}
+		strings[i] = str
+	}
+	return strings, nil
+}
+
+// Bytes exposes the underlying raw base64 data.
+func (m *PullMessage) Bytes() []byte {
+	return m.Data
+}
diff --git a/pkg/pubsub/publisher.go b/pkg/pubsub/publisher.go
index adaf3de..09a259c 100644
--- a/pkg/pubsub/publisher.go
+++ b/pkg/pubsub/publisher.go
@@ -15,13 +15,15 @@ import (
 
 type Publisher struct {
 	dataplane  *api.ClientWithResponses
-	TopicId    uuid.UUID
+	TopicID    uuid.UUID
 	logger     logr.Logger
-	topicUrl   url.URL
+	topicURL   url.URL
 	httpClient *http.Client
 }
 
-func NewPublisher(topicId uuid.UUID, opts ...Option) *Publisher {
+// NewPublisher instantiates a new Publisher. It returns an error if the underlying
+// API dataplane client fails to initialize.
+func NewPublisher(topicID uuid.UUID, opts ...Option) *Publisher {
 	cfg := &clientConfig{
 		httpClient: http.DefaultClient,
 		host:       "pubsub.eu01.onstackit.cloud",
@@ -32,26 +34,42 @@ func NewPublisher(topicId uuid.UUID, opts ...Option) *Publisher {
 		opt(cfg)
 	}
 
-	topicUrl := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicId.String(), cfg.host)}
+	topicURL := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicID.String(), cfg.host)}
+
+	// SAFETY: The error here can never be non nil, as WithHTTPClient always returns a nil error.
+	dataplane, _ := api.NewClientWithResponses(
+		topicURL.String(),
+		api.WithHTTPClient(cfg.httpClient),
+	)
 
 	publisher := &Publisher{
-		TopicId:    topicId,
-		topicUrl:   topicUrl,
+		TopicID:    topicID,
+		topicURL:   topicURL,
 		httpClient: cfg.httpClient,
-		logger:     cfg.logger.WithValues("topic_id", topicId),
+		logger:     cfg.logger.WithValues("topic_id", topicID),
+		dataplane:  dataplane,
 	}
 
-	publisher.dataplane, _ = api.NewClientWithResponses(
-		publisher.topicUrl.String(),
-		api.WithHTTPClient(publisher.httpClient),
-	)
-
 	return publisher
 }
 
-func (p *Publisher) Publish(ctx context.Context, messages [][]byte) ([]uint64, error) {
-	messagesToPublish := make([]api.PublishMessage, len(messages))
+// PublishStrings acts as a lightweight adapter converting string slice to byte slices,
+// deferring all encoding safely to the core Publish method.
+func (p *Publisher) PublishStrings(ctx context.Context, messages ...string) ([]uint64, error) {
+	byteMessages := make([][]byte, len(messages))
 	for i, msg := range messages {
+		byteMessages[i] = []byte(msg)
+	}
+	return p.Publish(ctx, byteMessages)
+}
+
+// Publish processes raw bytes, encodes them transparently using bytesToBase64,
+// and transmits them out via the API client.
+func (p *Publisher) Publish(ctx context.Context, messages [][]byte) ([]uint64, error) {
+	encodedMessages := bytesToBase64(messages...)
+
+	messagesToPublish := make([]api.PublishMessage, len(encodedMessages))
+	for i, msg := range encodedMessages {
 		messagesToPublish[i] = api.PublishMessage{
 			Data: msg,
 		}
@@ -64,10 +82,7 @@ func (p *Publisher) Publish(ctx context.Context, messages [][]byte) ([]uint64, e
 	p.logger.V(4).Info("publishing messages", "count", len(messages))
 	resp, err := p.dataplane.PublishMessagesWithResponse(ctx, reqBody)
 	if err != nil {
-		return nil, &NetworkError{
-			Msg: "failed to execute publish messages request",
-			Err: err,
-		}
+		return nil, NewNetworkError("failed to execute publish messages request", err)
 	}
 
 	if resp.StatusCode() != http.StatusOK {
@@ -84,16 +99,12 @@ func (p *Publisher) Publish(ctx context.Context, messages [][]byte) ([]uint64, e
 }
 
 // Purge removes all messages currently stored in the topic.
-// This is a destructive operation and cannot be undone.
 func (p *Publisher) Purge(ctx context.Context) error {
 	p.logger.V(4).Info("purging topic")
 
 	resp, err := p.dataplane.PurgeTopicWithResponse(ctx)
 	if err != nil {
-		return &NetworkError{
-			Msg: "failed to execute purge topic request",
-			Err: err,
-		}
+		return NewNetworkError("failed to execute purge topic request", err)
 	}
 
 	if resp.StatusCode() != http.StatusNoContent {
diff --git a/pkg/pubsub/publisher_test.go b/pkg/pubsub/publisher_test.go
index 884512a..52ba966 100644
--- a/pkg/pubsub/publisher_test.go
+++ b/pkg/pubsub/publisher_test.go
@@ -14,9 +14,8 @@ var _ = Describe("publish a message", func() {
 	Context("publish a message", func() {
 		It("should publish the message and return a message ID", func(ctx context.Context) {
 			publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-			messages := pubsub.StringsToBase64("Hello, Stackit!")
 
-			messageIDs, err := publisher.Publish(ctx, messages)
+			messageIDs, err := publisher.PublishStrings(ctx, "Hello, Stackit!")
 
 			Expect(err).ToNot(HaveOccurred())
 			Expect(messageIDs).ToNot(BeNil())
@@ -28,9 +27,8 @@ var _ = Describe("publish a message", func() {
 		It("should successfully remove all messages from the topic", func(ctx context.Context) {
 			publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
 			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-			messagesToPublish := pubsub.StringsToBase64("message-to-be-purged-1", "message-to-be-purged-2")
 
-			_, err := publisher.Publish(ctx, messagesToPublish)
+			_, err := publisher.PublishStrings(ctx, "message-to-be-purged-1", "message-to-be-purged-2")
 			Expect(err).ToNot(HaveOccurred())
 
 			Eventually(func(g Gomega) {
diff --git a/pkg/pubsub/subscriber.go b/pkg/pubsub/subscriber.go
index bed8b47..c8433e7 100644
--- a/pkg/pubsub/subscriber.go
+++ b/pkg/pubsub/subscriber.go
@@ -15,15 +15,17 @@ import (
 )
 
 type Subscriber struct {
-	SubscriptionId uuid.UUID
+	SubscriptionID uuid.UUID
 	logger         logr.Logger
 	dataplane      *api.ClientWithResponses
-	topicUrl       url.URL
+	topicURL       url.URL
 	httpClient     *http.Client
 	wg             sync.WaitGroup
 }
 
-func NewSubscriber(topicId uuid.UUID, subscriptionId uuid.UUID, opts ...Option) *Subscriber {
+// NewSubscriber instantiates a new Subscriber. It returns an error if the underlying
+// API dataplane client fails to initialize.
+func NewSubscriber(topicID uuid.UUID, subscriptionID uuid.UUID, opts ...Option) *Subscriber {
 	cfg := &clientConfig{
 		httpClient: http.DefaultClient,
 		host:       "pubsub.eu01.onstackit.cloud",
@@ -34,20 +36,22 @@ func NewSubscriber(topicId uuid.UUID, subscriptionId uuid.UUID, opts ...Option)
 		opt(cfg)
 	}
 
-	topicUrl := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicId.String(), cfg.host)}
+	topicURL := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicID.String(), cfg.host)}
+
+	// SAFETY: The error here can never be non nil, as WithHTTPClient always returns a nil error.
+	dataplane, _ := api.NewClientWithResponses(
+		topicURL.String(),
+		api.WithHTTPClient(cfg.httpClient),
+	)
 
 	subscriber := &Subscriber{
-		SubscriptionId: subscriptionId,
-		topicUrl:       topicUrl,
+		SubscriptionID: subscriptionID,
+		topicURL:       topicURL,
 		httpClient:     cfg.httpClient,
-		logger:         cfg.logger.WithValues("subscription_id", topicId),
+		logger:         cfg.logger.WithValues("subscription_id", subscriptionID),
+		dataplane:      dataplane,
 	}
 
-	subscriber.dataplane, _ = api.NewClientWithResponses(
-		subscriber.topicUrl.String(),
-		api.WithHTTPClient(subscriber.httpClient),
-	)
-
 	return subscriber
 }
 
@@ -56,25 +60,18 @@ func (s *Subscriber) Ack(ctx context.Context, ids []string) error {
 		AckIds: ids,
 	}
 
-	s.logger.V(4).Info("acknowledging messages",
-		"count", len(ids),
-	)
+	s.logger.V(4).Info("acknowledging messages", "count", len(ids))
 
-	resp, err := s.dataplane.AckMessagesWithResponse(ctx, s.SubscriptionId, reqBody)
+	resp, err := s.dataplane.AckMessagesWithResponse(ctx, s.SubscriptionID, reqBody)
 	if err != nil {
-		return NewNetworkError(
-			"failed to execute ack messages request",
-			err,
-		)
+		return NewNetworkError("failed to execute ack messages request", err)
 	}
 
 	if resp.StatusCode() != http.StatusNoContent {
 		return NewAPIError(resp.StatusCode(), resp.Body)
 	}
 
-	s.logger.V(4).Info("acknowledged messages",
-		"count", len(ids),
-	)
+	s.logger.V(4).Info("acknowledged messages", "count", len(ids))
 	return nil
 }
 
@@ -83,29 +80,22 @@ func (s *Subscriber) Nack(ctx context.Context, ids []string) error {
 		NackIds: ids,
 	}
 
-	s.logger.V(4).Info("nacking messages",
-		"count", len(ids),
-	)
+	s.logger.V(4).Info("nacking messages", "count", len(ids))
 
-	resp, err := s.dataplane.NackMessagesWithResponse(ctx, s.SubscriptionId, reqBody)
+	resp, err := s.dataplane.NackMessagesWithResponse(ctx, s.SubscriptionID, reqBody)
 	if err != nil {
-		return NewNetworkError(
-			"failed to execute nack messages request",
-			err,
-		)
+		return NewNetworkError("failed to execute nack messages request", err)
 	}
 
 	if resp.StatusCode() != http.StatusNoContent {
 		return NewAPIError(resp.StatusCode(), resp.Body)
 	}
 
-	s.logger.V(4).Info("nacked messages",
-		"count", len(ids),
-	)
+	s.logger.V(4).Info("nacked messages", "count", len(ids))
 	return nil
 }
 
-func toSdkMessages(m []api.Message, subscription *Subscriber) PullMessages {
+func toSDKMessages(m []api.Message, subscription *Subscriber) PullMessages {
 	sdkMessages := make(PullMessages, len(m))
 	for i, msg := range m {
 		sdkMessages[i] = PullMessage{
@@ -148,7 +138,6 @@ 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
@@ -165,28 +154,23 @@ func (s *Subscriber) Pull(ctx context.Context, opts ...PullOption) (PullMessages
 		PubSubLongPullDuration: longPullDuration,
 	}
 
-	s.logger.V(4).Info("pulling messages",
-		"max_messages", int(cfg.maxMessages),
-	)
+	s.logger.V(4).Info("pulling messages", "max_messages", int(cfg.maxMessages))
 
-	resp, err := s.dataplane.PullMessagesWithResponse(ctx, s.SubscriptionId, &reqBody)
+	resp, err := s.dataplane.PullMessagesWithResponse(ctx, s.SubscriptionID, &reqBody)
 	if err != nil {
-		return nil, &NetworkError{
-			Msg: "failed to execute pull messages request",
-			Err: err,
-		}
+		return nil, NewNetworkError("failed to execute pull messages request", err)
 	}
 
 	if resp.StatusCode() != http.StatusOK {
 		return nil, NewAPIError(resp.StatusCode(), resp.Body)
 	}
 
-	messages := toSdkMessages(resp.JSON200.Messages, s)
+	messages := toSDKMessages(resp.JSON200.Messages, s)
 
 	s.logger.V(4).Info(
 		"pulled messages",
 		"count", len(resp.JSON200.Messages),
-		"ack_ids", messages.GetAckIDs(),
+		"ack_ids", messages.AckIDs(),
 	)
 	return messages, nil
 }
diff --git a/pkg/pubsub/subscriber_test.go b/pkg/pubsub/subscriber_test.go
index 6ac7d23..0d17ff1 100644
--- a/pkg/pubsub/subscriber_test.go
+++ b/pkg/pubsub/subscriber_test.go
@@ -17,9 +17,8 @@ var _ = Describe("Acknowledge messages", func() {
 
 	BeforeEach(func(ctx context.Context) {
 		publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-		messagesToPublish := pubsub.StringsToBase64("test1")
 
-		_, err := publisher.Publish(ctx, messagesToPublish)
+		_, err := publisher.PublishStrings(ctx, "test1")
 		Expect(err).ToNot(HaveOccurred())
 
 		subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
@@ -27,7 +26,7 @@ var _ = Describe("Acknowledge messages", func() {
 		Expect(err).ToNot(HaveOccurred())
 		Expect(pulledMessages).ToNot(BeEmpty())
 
-		AckIDs = pulledMessages.GetAckIDs()
+		AckIDs = pulledMessages.AckIDs()
 	})
 
 	Context("acknowledging messages", func() {
@@ -97,9 +96,8 @@ var _ = Describe("Pull messages", func() {
 	Context("pulling messages", func() {
 		BeforeEach(func(ctx context.Context) {
 			publisher := pubsub.NewPublisher(topicId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-			messagesToPublish := pubsub.StringsToBase64("test1", "test2", "test3")
 
-			_, err := publisher.Publish(ctx, messagesToPublish)
+			_, err := publisher.PublishStrings(ctx, "test1", "test2", "test3")
 			Expect(err).ToNot(HaveOccurred())
 		})
 
@@ -135,9 +133,7 @@ var _ = Describe("PullJob", func() {
 		Expect(err).ToNot(HaveOccurred())
 
 		// publishing test messages
-		messagesToPublish := pubsub.StringsToBase64("testMessage")
-
-		_, err = publisher.Publish(ctx, messagesToPublish)
+		_, err = publisher.PublishStrings(ctx, "testMessage")
 		Expect(err).ToNot(HaveOccurred())
 	})
 
@@ -155,10 +151,10 @@ var _ = Describe("PullJob", func() {
 			Eventually(jobChan, "5s").Should(Receive(&receivedMessages))
 			Expect(receivedMessages).To(HaveLen(1))
 
-			decodedStrings, err := pubsub.Base64ToStrings(string(receivedMessages[0].Data))
+			decodedString, err := receivedMessages[0].DecodeString()
 			Expect(err).ToNot(HaveOccurred())
-			Expect(decodedStrings[0]).To(Equal("testMessage"))
-			err = subscriber.Ack(ctx, receivedMessages.GetAckIDs())
+			Expect(decodedString).To(Equal("testMessage"))
+			err = subscriber.Ack(ctx, receivedMessages.AckIDs())
 			Expect(err).ToNot(HaveOccurred())
 
 			cancel()
@@ -181,10 +177,10 @@ var _ = Describe("PullJob", func() {
 			Eventually(jobChan, "10s").Should(Receive(&receivedMessages))
 			Expect(receivedMessages).To(HaveLen(1))
 
-			decodedStrings, err := pubsub.Base64ToStrings(string(receivedMessages[0].Data))
+			decodedString, err := receivedMessages[0].DecodeString()
 			Expect(err).ToNot(HaveOccurred())
-			Expect(decodedStrings[0]).To(Equal("testMessage"))
-			err = subscriber.Ack(ctx, receivedMessages.GetAckIDs())
+			Expect(decodedString).To(Equal("testMessage"))
+			err = subscriber.Ack(ctx, receivedMessages.AckIDs())
 			Expect(err).ToNot(HaveOccurred())
 
 			cancel()
@@ -203,10 +199,10 @@ var _ = Describe("PullJob", func() {
 			callback := func(ctx context.Context, messages pubsub.PullMessages) {
 				defer GinkgoRecover()
 				Expect(messages).To(HaveLen(1))
-				decoded, err := pubsub.Base64ToStrings(string(messages[0].Data))
+				decoded, err := messages[0].DecodeString()
 				Expect(err).ToNot(HaveOccurred())
-				Expect(decoded[0]).To(Equal("testMessage"))
-				err = subscriber.Ack(ctx, messages.GetAckIDs())
+				Expect(decoded).To(Equal("testMessage"))
+				err = subscriber.Ack(ctx, messages.AckIDs())
 				Expect(err).ToNot(HaveOccurred())
 				callbackInvoked.Store(true)
 				cancel()

From de1a4e5ff33dc96291b66c1808bffb47978d7fc4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Paul=20Gro=C3=9Fmann?= <paul.grossmann@digits.schwarz>
Date: Mon, 6 Jul 2026 19:32:23 +0200
Subject: [PATCH 07/10] fix: permission part in script to create PubSub
 resources (#20)

* fix: permission part in script to create PubSub resources

* fix: corrected api response codes in create-pubsub-resources.sh

* feat: better error messages for github pipelines in create-pubsub-resources.sh
---
 scripts/create-pubsub-resources.sh | 40 ++++++++++++------------------
 1 file changed, 16 insertions(+), 24 deletions(-)

diff --git a/scripts/create-pubsub-resources.sh b/scripts/create-pubsub-resources.sh
index c072064..3be9f5a 100755
--- a/scripts/create-pubsub-resources.sh
+++ b/scripts/create-pubsub-resources.sh
@@ -11,7 +11,7 @@ TOKEN=$(stackit auth activate-service-account --only-print-access-token --servic
 echo "SERVICE_ACCOUNT_TOKEN=$TOKEN" >> $GITHUB_ENV
 
 if [ -z "$TOKEN" ] || [ ${#TOKEN} -lt 20 ]; then
-  echo "Error: Retrieved token is empty or too short."
+  echo "::error file=scripts/create-pubsub-resources.sh::Retrieved token is empty or too short."
   exit 1
 fi
 
@@ -26,9 +26,8 @@ HTTP_STATUS=$(echo "$TOPICRESPONSE" | tail -n 1)
 TOPICBODY=$(echo "$TOPICRESPONSE" | sed '$d')
 echo "Response Body: $TOPICBODY"
 
-if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
-  echo "API Error (HTTP $HTTP_STATUS)"
-  echo "Response Topic Body: $TOPICBODY"
+if [ "$HTTP_STATUS" -ne 202 ]; then
+  echo "::error file=scripts/create-pubsub-resources.sh::Failed to create topic (HTTP $HTTP_STATUS) - Response: $TOPICBODY"
   exit 1
 fi
 
@@ -44,7 +43,7 @@ for i in {1..50}; do
     break
   fi
   if [ "$i" -eq 50 ]; then
-    echo "Topic did not become active in time."
+    echo "::error file=scripts/create-pubsub-resources.sh::Topic $TOPIC_ID did not become active in time."
     exit 1
   fi
   sleep 5
@@ -61,9 +60,8 @@ HTTP_STATUS=$(echo "$SUBRESPONSE" | tail -n 1)
 SUBBODY=$(echo "$SUBRESPONSE" | sed '$d')
 echo "Response Body: $SUBBODY"
 
-if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
-  echo "API Error (HTTP $HTTP_STATUS)"
-  echo "Response SUBSCRIPTION Body: $SUBBODY"
+if [ "$HTTP_STATUS" -ne 202 ]; then
+  echo "::error file=scripts/create-pubsub-resources.sh::Failed to create subscription (HTTP $HTTP_STATUS) - Response: $SUBBODY"
   exit 1
 fi
 
@@ -79,7 +77,7 @@ for i in {1..50}; do
     break
   fi
   if [ "$i" -eq 50 ]; then
-    echo "Subscription did not become active in time."
+    echo "::error file=scripts/create-pubsub-resources.sh::Subscription $SUBSCRIPTION_ID did not become active in time."
     exit 1
   fi
   sleep 5
@@ -89,37 +87,31 @@ done
 #ACCESS
 echo "Granting Publisher Access via curl (Targeting: $REGION)..."
 PUBLISHER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/publishers" \
-  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Authorization: Bearer $TOKEN" \
   -H "Content-Type: application/json" \
-  -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
+  -d "{\"emailAddress\": \"$PUBLISHER_MAIL\"}")
 
-  HTTP_STATUS=$(echo "$PUBLISHER_RESPONSE" | tail -n 1)
-  PUBLISHER_BODY=$(echo "$PUBLISHER_RESPONSE" | sed '$d')
-  echo "Response Body: $PUBLISHER_BODY"
+HTTP_STATUS=$(echo "$PUBLISHER_RESPONSE" | tail -n 1)
+PUBLISHER_BODY=$(echo "$PUBLISHER_RESPONSE" | sed '$d')
 
-if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
+if [ "$HTTP_STATUS" -ne 202 ]; then
   echo "::error file=scripts/create-pubsub-resources.sh::Failed to grant publisher access (HTTP $HTTP_STATUS) - Response: $PUBLISHER_BODY"
   exit 1
 fi
 
-echo "Response SUBSCRIPTION Body: $GPARESPONSE"
-
 echo "Granting Subscriber Access via curl (Targeting: $REGION)..."
 SUBSCRIBER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}/subscribers" \
-  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Authorization: Bearer $TOKEN" \
   -H "Content-Type: application/json" \
-  -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
+  -d "{\"emailAddress\": \"$PUBLISHER_MAIL\"}")
 
 HTTP_STATUS=$(echo "$SUBSCRIBER_RESPONSE" | tail -n 1)
 SUBSCRIBER_BODY=$(echo "$SUBSCRIBER_RESPONSE" | sed '$d')
-echo "Response Body: $SUBSCRIBER_BODY"
 
-if [ "$HTTP_STATUS" -ne 202 ] && [ "$HTTP_STATUS" -ne 200 ]; then
+if [ "$HTTP_STATUS" -ne 202 ]; then
   echo "::error file=scripts/create-pubsub-resources.sh::Failed to grant subscriber access (HTTP $HTTP_STATUS) - Response: $SUBSCRIBER_BODY"
   exit 1
 fi
 
-echo "Response SUBSCRIPTION Body: $GSARESPONSE"
-
 echo "Waiting for access permissions to propagate..."
-sleep 15
\ No newline at end of file
+sleep 5
\ No newline at end of file

From 5749987659f319419b4c46c09998fa993bccf981 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Paul=20Gro=C3=9Fmann?= <paul.grossmann@digits.schwarz>
Date: Mon, 6 Jul 2026 21:43:23 +0200
Subject: [PATCH 08/10] Chore/update oapi specs for correct server url (#19)

* fix: silenced echo cmd in Makefile

* chore: updated oapi specs for correctly set url
---
 Makefile         |  2 +-
 api/openapi.yaml | 14 +-------------
 2 files changed, 2 insertions(+), 14 deletions(-)

diff --git a/Makefile b/Makefile
index 952fdba..f77ff95 100644
--- a/Makefile
+++ b/Makefile
@@ -7,7 +7,7 @@ export PATH := $(CURDIR)/bin:$(PATH)
 
 .PHONY: generate
 generate:
-	echo "Generating code..."
+	@echo "Generating code..."
 	oapi-codegen --config=scripts/sdk.cfg.yaml api/openapi.yaml
 
 .PHONY: prepare
diff --git a/api/openapi.yaml b/api/openapi.yaml
index 3537b1c..8b77b48 100644
--- a/api/openapi.yaml
+++ b/api/openapi.yaml
@@ -7,19 +7,7 @@ info:
     This API provides endpoints for managing PubSubs.
 
 servers:
-  - url: https://{topicId}.api.pubsub-dp.eu01.dev.stackit.cloud
-    description: Development endpoint in eu01.
-    variables:
-      topicId:
-        description: The Topic UUID.
-        default: "00000000-0000-0000-0000-000000000000"
-  - url: https://{topicId}.api.pubsub-dp.eu01.qa.stackit.cloud
-    description: QA endpoint in eu01.
-    variables:
-      topicId:
-        description: The Topic UUID.
-        default: "00000000-0000-0000-0000-000000000000"
-  - url: https://{topicId}.api.pubsub-dp.eu01.stackit.cloud
+  - url: https://{topicId}.pubsub.eu01.onstackit.cloud
     description: Production endpoint in eu01.
     variables:
       topicId:

From be416bfee07789cacd55c3946059826c42fd5ee4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Paul=20Gro=C3=9Fmann?= <paul.grossmann@digits.schwarz>
Date: Tue, 7 Jul 2026 13:20:20 +0200
Subject: [PATCH 09/10] fix: ci cleanup of PubSub topic (#21)

* fix: ci cleanup of PubSub topic

* fix: corrected base url in delete-pubsub-resources.sh

* refactor: getting of token and use of stackit curl so that the topic is being cleaned up correctly

* refactor: back to curl with but now with stackit auth get-access-token
---
 .github/workflows/ci.yml           | 14 +++-----------
 scripts/create-pubsub-resources.sh | 22 ++++++++--------------
 scripts/delete-pubsub-resources.sh | 27 +++++++++++++++++----------
 3 files changed, 28 insertions(+), 35 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3c9509d..c0c3de6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -59,17 +59,14 @@ jobs:
         with:
           go-version-file: 'go.mod'
 
-      - name: Load Key from GitHub Secrets
-        run: |
-          echo '${{ secrets.STACKIT_SERVICE_ACCOUNT_KEY }}' > $RUNNER_TEMP/service_account_key.json
-
       - name: Set up Stackit CLI
         run: |
           echo "Stackit CLI Installation"
           sudo snap install stackit --classic
 
       - name: Import QA Profile and Login
-        if: always()
+        env:
+          STACKIT_SERVICE_ACCOUNT_KEY: ${{ secrets.STACKIT_SERVICE_ACCOUNT_KEY }}
         run: |
           echo '${{ secrets.STACKIT_QA_ENV_JSON }}' > $RUNNER_TEMP/qa-env.json
           
@@ -85,13 +82,10 @@ jobs:
             echo "pubsub-qa should been removed. Creating..."
             stackit config profile import -c "@$RUNNER_TEMP/qa-env.json" --name pubsub-qa
             stackit config profile set pubsub-qa
-            stackit auth activate-service-account --service-account-key-path "$RUNNER_TEMP/service_account_key.json"
+            stackit auth activate-service-account # key is being read out from the env STACKIT_SERVICE_ACCOUNT_KEY
 
       - name: Create PubSub Topic and Subscription
         run: ./scripts/create-pubsub-resources.sh
-        env:
-          REPO_ROOT: ${{ github.workspace }}
-          SA_KEY_PATH: ${{ runner.temp }}/service_account_key.json
 
       - name: Run Unit Tests
         run: make test ENABLE_TEST_COVERAGE=true
@@ -101,12 +95,10 @@ jobs:
           ENVIRONMENT: ${{ env.ENVIRONMENT }}
           TOKEN_CUSTOM_URI: ${{ env.TOKEN_CUSTOM_URI }}
           SERVICE_ACCOUNT_TOKEN: ${{ secrets.STACKIT_SERVICE_ACCOUNT_KEY }}
-          TOKEN: ${{ secrets.TOKEN }}
 
       - name: Cleanup - Delete STACKIT Resources
         if: always()
         run: |
-          cp $RUNNER_TEMP/service_account_key.json key.json
           ./scripts/delete-pubsub-resources.sh
 
       - name: Delete QA Profile
diff --git a/scripts/create-pubsub-resources.sh b/scripts/create-pubsub-resources.sh
index 3be9f5a..727e3fe 100755
--- a/scripts/create-pubsub-resources.sh
+++ b/scripts/create-pubsub-resources.sh
@@ -6,19 +6,13 @@ REGION="eu01"
 BASE_URL="https://pubsub.api.qa.stackit.cloud/v1alpha"
 PUBLISHER_MAIL="pubsub-dataplane-sdk-44cqm3i8@sa.stackit.cloud"
 
-echo "Fetching fresh access token..."
-TOKEN=$(stackit auth activate-service-account --only-print-access-token --service-account-key-path "$SA_KEY_PATH" | tr -d '\r\n ')
-echo "SERVICE_ACCOUNT_TOKEN=$TOKEN" >> $GITHUB_ENV
-
-if [ -z "$TOKEN" ] || [ ${#TOKEN} -lt 20 ]; then
-  echo "::error file=scripts/create-pubsub-resources.sh::Retrieved token is empty or too short."
-  exit 1
-fi
+# Get token from stackit cli
+TOKEN=$(stackit auth get-access-token)
 
 #TOPIC
 echo "Creating Topic via curl (Targeting: $REGION)..."
 TOPICRESPONSE=$(curl -sk -w "\n%{http_code}" -X POST "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics" \
-  -H "Authorization: Bearer $TOKEN" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
 
@@ -37,7 +31,7 @@ echo "TOPIC_ID=$TOPIC_ID" >> $GITHUB_ENV
 
 echo "Waiting for topic to become active..."
 for i in {1..50}; do
-  STATUS=$(curl -sk -H "Authorization: Bearer $TOKEN" "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}" | jq -r '.state')
+  STATUS=$(curl -sk -H "Authorization: Bearer ${TOKEN}" "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}" | jq -r '.state')
   echo "Current topic status: $STATUS"
   if [ "$STATUS" == "active" ]; then
     break
@@ -52,7 +46,7 @@ done
 #SUBSCRIPTION
 echo "Creating Subscription via curl (Targeting: $REGION)..."
 SUBRESPONSE=$(curl -sk -w "\n%{http_code}" -X POST "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions" \
-  -H "Authorization: Bearer $TOKEN" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"displayName\": \"ci-topic-$(date +%s)\"}")
 
@@ -71,7 +65,7 @@ echo "SUBSCRIPTION_ID=$SUBSCRIPTION_ID" >> $GITHUB_ENV
 
 echo "Waiting for subscription to become active..."
 for i in {1..50}; do
-  STATUS=$(curl -sk -H "Authorization: Bearer $TOKEN" "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}" | jq -r '.state')
+  STATUS=$(curl -sk -H "Authorization: Bearer ${TOKEN}" "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}" | jq -r '.state')
   echo "Current subscription status: $STATUS"
   if [ "$STATUS" == "active" ]; then
     break
@@ -87,7 +81,7 @@ done
 #ACCESS
 echo "Granting Publisher Access via curl (Targeting: $REGION)..."
 PUBLISHER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/publishers" \
-  -H "Authorization: Bearer $TOKEN" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"emailAddress\": \"$PUBLISHER_MAIL\"}")
 
@@ -101,7 +95,7 @@ fi
 
 echo "Granting Subscriber Access via curl (Targeting: $REGION)..."
 SUBSCRIBER_RESPONSE=$(curl -sk -w "\n%{http_code}" -X PATCH "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}/subscribers" \
-  -H "Authorization: Bearer $TOKEN" \
+  -H "Authorization: Bearer ${TOKEN}" \
   -H "Content-Type: application/json" \
   -d "{\"emailAddress\": \"$PUBLISHER_MAIL\"}")
 
diff --git a/scripts/delete-pubsub-resources.sh b/scripts/delete-pubsub-resources.sh
index 73272bc..d965cfa 100755
--- a/scripts/delete-pubsub-resources.sh
+++ b/scripts/delete-pubsub-resources.sh
@@ -1,7 +1,7 @@
 #!/bin/bash
 
 # Configuration
-BASE_URL="https://pubsub.api.eu01.qa/v1alpha"
+BASE_URL="https://pubsub.api.qa.stackit.cloud/v1alpha"
 PROJECT_ID="0ec85b07-ecb2-4253-9ba8-25ae06db1b7a"
 REGION="eu01"
 
@@ -10,14 +10,21 @@ if [ -z "$TOPIC_ID" ]; then
     exit 0
 fi
 
-echo "Cleaning up Subscription: $SUBSCRIPTION_ID"
-curl -s --request DELETE \
-  --url "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}/subscriptions/${SUBSCRIPTION_ID}" \
-  --header "Authorization: Bearer ${STACKIT_SERVICE_ACCOUNT_TOKEN}"
+# Get token from stackit cli
+TOKEN=$(stackit auth get-access-token)
 
-echo "Cleaning up Topic: $TOPIC_ID"
-curl -s --request DELETE \
-  --url "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}" \
-  --header "Authorization: Bearer ${STACKIT_SERVICE_ACCOUNT_TOKEN}"
+# We use force=true query parameter on the topic deletion to perform a cascading delete of the topic
+# and all of its active subscriptions.
+echo "Cleaning up Topic: $TOPIC_ID with force=true"
+DELETE_RESPONSE=$(curl -sk -w "\n%{http_code}" -X DELETE \
+  -H "Authorization: Bearer ${TOKEN}" \
+  --url "${BASE_URL}/projects/${PROJECT_ID}/regions/${REGION}/topics/${TOPIC_ID}?force=true")
 
-echo "Cleanup complete."
\ No newline at end of file
+HTTP_STATUS=$(echo "$DELETE_RESPONSE" | tail -n 1)
+DELETE_BODY=$(echo "$DELETE_RESPONSE" | sed '$d')
+
+if [ "$HTTP_STATUS" -ne 202 ]; then
+  echo "::warning file=scripts/delete-pubsub-resources.sh::Failed to delete topic $TOPIC_ID (HTTP $HTTP_STATUS) - Response: $DELETE_BODY"
+else
+  echo "Successfully deleted topic $TOPIC_ID (HTTP $HTTP_STATUS)"
+fi

From 082251eab491baa2ef6beb4c1e0935097c91a187 Mon Sep 17 00:00:00 2001
From: selinafehn <144441353+selinafehn@users.noreply.github.com>
Date: Thu, 6 Aug 2026 10:59:17 +0200
Subject: [PATCH 10/10] fix-max-messages-amount (#22)

* fix: set max messages to max 32 instead 128

* fix: set max messages to max 32 instead 128 or 64

---------

Co-authored-by: Selina Fehn <selina.fehn@stackit.cloud>
---
 pkg/pubsub/subscriber.go      | 2 +-
 pkg/pubsub/subscriber_test.go | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/pkg/pubsub/subscriber.go b/pkg/pubsub/subscriber.go
index c8433e7..e4aea9b 100644
--- a/pkg/pubsub/subscriber.go
+++ b/pkg/pubsub/subscriber.go
@@ -131,7 +131,7 @@ func WithLongPullDuration(milliseconds int32) PullOption {
 
 func (s *Subscriber) Pull(ctx context.Context, opts ...PullOption) (PullMessages, error) {
 	cfg := &pullOptions{
-		maxMessages: 64,
+		maxMessages: 32,
 	}
 
 	for _, opt := range opts {
diff --git a/pkg/pubsub/subscriber_test.go b/pkg/pubsub/subscriber_test.go
index 0d17ff1..bf2de99 100644
--- a/pkg/pubsub/subscriber_test.go
+++ b/pkg/pubsub/subscriber_test.go
@@ -103,7 +103,7 @@ var _ = Describe("Pull messages", func() {
 
 		It("no error is occurring", func(ctx context.Context) {
 			subscriber := pubsub.NewSubscriber(topicId, subscriptionId, pubsub.WithHTTPRoundTripper(rt), pubsub.WithHost(environment))
-			resp, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(128))
+			resp, err := subscriber.Pull(ctx, pubsub.WithMaxMessages(32))
 			Expect(resp).ToNot(BeNil())
 			Expect(err).ToNot(HaveOccurred())
 		})