Update samples for better UX

README.md

@@ -83,19 +83,24 @@ 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")
 subscriber := pubsub.NewSubscriber(topicID, subscriptionID, pubsub.WithHTTPRoundTripper(roundTripper))
 
 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) {

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)

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)

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

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"

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
 }

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
+}

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 {

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