From a7346166ba02ded43bc35e913989de0cae96a0d3 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Mon, 29 Jun 2026 15:38:26 +0200 Subject: [PATCH 1/5] fix. updated samples --- example/example_publish.go | 7 ++----- example/example_pull.go | 3 +++ go.mod | 2 +- mise.toml | 2 +- pkg/pubsub/helper.go | 9 +++++---- pkg/pubsub/message.go | 25 +++++++++++++++++++++++-- pkg/pubsub/publisher.go | 4 ++++ pkg/pubsub/publisher_test.go | 6 ++---- pkg/pubsub/subscriber_test.go | 22 +++++++++------------- 9 files changed, 50 insertions(+), 30 deletions(-) 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..7bed8f9 100644 --- a/example/example_pull.go +++ b/example/example_pull.go @@ -38,6 +38,9 @@ func pull() { } log.Printf("Successfully pulled message: %v", pulledMessages) + for i := 0; i < len(pulledMessages); i++ { + log.Printf("Message [%d]: %s:", i, pulledMessages[0].Data) + } // Get your AckIDs and acknowledge them ackIDs := pulledMessages.GetAckIDs() 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..1a69482 100644 --- a/pkg/pubsub/helper.go +++ b/pkg/pubsub/helper.go @@ -5,7 +5,7 @@ import ( "fmt" ) -func StringsToBase64(strings ...string) [][]byte { +func stringsToBase64(strings ...string) [][]byte { result := make([][]byte, len(strings)) for i, string := range strings { @@ -15,15 +15,16 @@ func StringsToBase64(strings ...string) [][]byte { return result } -func Base64ToStrings(encodedStrings ...string) ([]string, error) { +func base64ToStrings(encodedStrings ...[]byte) ([]string, error) { result := make([]string, len(encodedStrings)) for i, s := range encodedStrings { - decodedBytes, err := base64.StdEncoding.DecodeString(s) + decodedBytes := make([]byte, base64.StdEncoding.DecodedLen(len(s))) + n, err := base64.StdEncoding.Decode(decodedBytes, s) if err != nil { return nil, fmt.Errorf("failed to decode string at index %d: %w", i, err) } - result[i] = string(decodedBytes) + result[i] = string(decodedBytes[:n]) } return result, nil diff --git a/pkg/pubsub/message.go b/pkg/pubsub/message.go index 1f860c7..53fbb22 100644 --- a/pkg/pubsub/message.go +++ b/pkg/pubsub/message.go @@ -14,8 +14,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}) } @@ -31,3 +29,26 @@ func (m PullMessages) GetAckIDs() []string { } return ids } + +func (m *PullMessage) AsString() (string, error) { + str, err := base64ToStrings(m.Data) + if err != nil { + return "", err + } + return str[0], nil +} + +func (m *PullMessage) AsBytes() []byte { + return m.Data +} + +type PullMessages []PullMessage + +func (m PullMessages) AsStrings() ([]string, error) { + strings := make([]string, 0, len(m)) + for _, msg := range m { + str, _ := msg.AsString() + strings = append(strings, str) + } + return strings, nil +} diff --git a/pkg/pubsub/publisher.go b/pkg/pubsub/publisher.go index adaf3de..32cd9dc 100644 --- a/pkg/pubsub/publisher.go +++ b/pkg/pubsub/publisher.go @@ -49,6 +49,10 @@ func NewPublisher(topicId uuid.UUID, opts ...Option) *Publisher { return publisher } +func (p *Publisher) PublishStrings(ctx context.Context, messages ...string) ([]uint64, error) { + return p.Publish(ctx, stringsToBase64(messages...)) +} + func (p *Publisher) Publish(ctx context.Context, messages [][]byte) ([]uint64, error) { messagesToPublish := make([]api.PublishMessage, len(messages)) for i, msg := range messages { 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_test.go b/pkg/pubsub/subscriber_test.go index 6ac7d23..6d65762 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)) @@ -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,9 +151,9 @@ 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].AsString() Expect(err).ToNot(HaveOccurred()) - Expect(decodedStrings[0]).To(Equal("testMessage")) + Expect(decodedString).To(Equal("testMessage")) err = subscriber.Ack(ctx, receivedMessages.GetAckIDs()) Expect(err).ToNot(HaveOccurred()) @@ -181,9 +177,9 @@ 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].AsString() Expect(err).ToNot(HaveOccurred()) - Expect(decodedStrings[0]).To(Equal("testMessage")) + Expect(decodedString).To(Equal("testMessage")) err = subscriber.Ack(ctx, receivedMessages.GetAckIDs()) Expect(err).ToNot(HaveOccurred()) @@ -203,9 +199,9 @@ 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].AsString() Expect(err).ToNot(HaveOccurred()) - Expect(decoded[0]).To(Equal("testMessage")) + Expect(decoded).To(Equal("testMessage")) err = subscriber.Ack(ctx, messages.GetAckIDs()) Expect(err).ToNot(HaveOccurred()) callbackInvoked.Store(true) From 6383a92877cb32d21114c1f18580b1048aa6083f Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Mon, 29 Jun 2026 15:45:53 +0200 Subject: [PATCH 2/5] feat: update readme for new methods --- README.md | 15 ++++++++++----- example/example_pull.go | 6 +++++- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 9d8fafc..6acb2dc 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,6 +91,14 @@ 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[0].AsString() + if err != nil { + log.Fatalf("Error converting message to string: %v", err) + } + log.Printf("Message [%d]: %s:", i, msg) +} + // Acknowledge ackIDs := pulledMessages.GetAckIDs() 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_pull.go b/example/example_pull.go index 7bed8f9..4843de5 100644 --- a/example/example_pull.go +++ b/example/example_pull.go @@ -39,7 +39,11 @@ func pull() { log.Printf("Successfully pulled message: %v", pulledMessages) for i := 0; i < len(pulledMessages); i++ { - log.Printf("Message [%d]: %s:", i, pulledMessages[0].Data) + msg, err := pulledMessages[0].AsString() + 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 From b6bf96e6871b5844fd99e6f989d26e18d2f16e16 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Mon, 29 Jun 2026 15:54:27 +0200 Subject: [PATCH 3/5] fix 0 -> i --- README.md | 2 +- example/example_pull.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 6acb2dc..9bc7de0 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ 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[0].AsString() + msg, err := pulledMessages[i].AsString() if err != nil { log.Fatalf("Error converting message to string: %v", err) } diff --git a/example/example_pull.go b/example/example_pull.go index 4843de5..010928c 100644 --- a/example/example_pull.go +++ b/example/example_pull.go @@ -39,7 +39,7 @@ func pull() { log.Printf("Successfully pulled message: %v", pulledMessages) for i := 0; i < len(pulledMessages); i++ { - msg, err := pulledMessages[0].AsString() + msg, err := pulledMessages[i].AsString() if err != nil { log.Fatalf("Error converting message to string: %v", err) } From 2e9bdab1eb66cda8d100e98cead0300eabb69b11 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Tue, 30 Jun 2026 13:30:42 +0200 Subject: [PATCH 4/5] save --- README.md | 4 +- example/example_pull.go | 6 +-- pkg/pubsub/helper.go | 33 ++++++--------- pkg/pubsub/message.go | 41 +++++++++++------- pkg/pubsub/publisher.go | 59 +++++++++++++++----------- pkg/pubsub/subscriber.go | 80 +++++++++++++++-------------------- pkg/pubsub/subscriber_test.go | 14 +++--- 7 files changed, 118 insertions(+), 119 deletions(-) diff --git a/README.md b/README.md index 9bc7de0..a4acf8c 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ 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].AsString() + msg, err := pulledMessages[i].Text() if err != nil { log.Fatalf("Error converting message to string: %v", err) } @@ -100,7 +100,7 @@ for i := 0; i < len(pulledMessages); i++ { } // Acknowledge -ackIDs := pulledMessages.GetAckIDs() +ackIDs := pulledMessages.AckIDs() err = subscriber.Ack(ctx, ackIDs) ``` diff --git a/example/example_pull.go b/example/example_pull.go index 010928c..e3d173e 100644 --- a/example/example_pull.go +++ b/example/example_pull.go @@ -39,7 +39,7 @@ func pull() { log.Printf("Successfully pulled message: %v", pulledMessages) for i := 0; i < len(pulledMessages); i++ { - msg, err := pulledMessages[i].AsString() + msg, err := pulledMessages[i].Text() if err != nil { log.Fatalf("Error converting message to string: %v", err) } @@ -47,14 +47,14 @@ func pull() { } // 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/pkg/pubsub/helper.go b/pkg/pubsub/helper.go index 1a69482..d6c848f 100644 --- a/pkg/pubsub/helper.go +++ b/pkg/pubsub/helper.go @@ -2,30 +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 ...[]byte) ([]string, error) { - result := make([]string, len(encodedStrings)) - - for i, s := range encodedStrings { - decodedBytes := make([]byte, base64.StdEncoding.DecodedLen(len(s))) - n, err := base64.StdEncoding.Decode(decodedBytes, s) - if err != nil { - return nil, fmt.Errorf("failed to decode string at index %d: %w", i, err) - } - result[i] = string(decodedBytes[:n]) +// 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 53fbb22..ef4edf1 100644 --- a/pkg/pubsub/message.go +++ b/pkg/pubsub/message.go @@ -2,6 +2,7 @@ package pubsub import ( "context" + "fmt" "time" ) @@ -22,7 +23,10 @@ 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 @@ -30,25 +34,30 @@ func (m PullMessages) GetAckIDs() []string { return ids } -func (m *PullMessage) AsString() (string, error) { - str, err := base64ToStrings(m.Data) +// 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 "", err + return "", fmt.Errorf("failed to decode message data: %w", err) } - return str[0], nil -} - -func (m *PullMessage) AsBytes() []byte { - return m.Data + return string(decoded), nil } -type PullMessages []PullMessage - -func (m PullMessages) AsStrings() ([]string, error) { - strings := make([]string, 0, len(m)) - for _, msg := range m { - str, _ := msg.AsString() - strings = append(strings, str) +// 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 32cd9dc..3370921 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, error) { cfg := &clientConfig{ httpClient: http.DefaultClient, host: "pubsub.eu01.onstackit.cloud", @@ -32,30 +34,44 @@ 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)} + + dataplane, err := api.NewClientWithResponses( + topicURL.String(), + api.WithHTTPClient(cfg.httpClient), + ) + if err != nil { + return nil, NewConfigurationError("failed to initialize api dataplane client", err) + } 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 + return publisher, nil } +// 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) { - return p.Publish(ctx, stringsToBase64(messages...)) + 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) { - messagesToPublish := make([]api.PublishMessage, len(messages)) - for i, msg := range messages { + encodedMessages := bytesToBase64(messages...) + + messagesToPublish := make([]api.PublishMessage, len(encodedMessages)) + for i, msg := range encodedMessages { messagesToPublish[i] = api.PublishMessage{ Data: msg, } @@ -68,10 +84,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 { @@ -88,16 +101,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/subscriber.go b/pkg/pubsub/subscriber.go index bed8b47..bd35412 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, error) { cfg := &clientConfig{ httpClient: http.DefaultClient, host: "pubsub.eu01.onstackit.cloud", @@ -34,21 +36,25 @@ 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)} + + dataplane, err := api.NewClientWithResponses( + topicURL.String(), + api.WithHTTPClient(cfg.httpClient), + ) + if err != nil { + return nil, NewConfigurationError("failed to initialize api dataplane client", err) + } 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 + return subscriber, nil } func (s *Subscriber) Ack(ctx context.Context, ids []string) error { @@ -56,25 +62,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 +82,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 +140,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 +156,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 6d65762..5e01d27 100644 --- a/pkg/pubsub/subscriber_test.go +++ b/pkg/pubsub/subscriber_test.go @@ -26,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() { @@ -151,10 +151,10 @@ var _ = Describe("PullJob", func() { Eventually(jobChan, "5s").Should(Receive(&receivedMessages)) Expect(receivedMessages).To(HaveLen(1)) - decodedString, err := receivedMessages[0].AsString() + decodedString, err := receivedMessages[0].Text() Expect(err).ToNot(HaveOccurred()) Expect(decodedString).To(Equal("testMessage")) - err = subscriber.Ack(ctx, receivedMessages.GetAckIDs()) + err = subscriber.Ack(ctx, receivedMessages.AckIDs()) Expect(err).ToNot(HaveOccurred()) cancel() @@ -177,10 +177,10 @@ var _ = Describe("PullJob", func() { Eventually(jobChan, "10s").Should(Receive(&receivedMessages)) Expect(receivedMessages).To(HaveLen(1)) - decodedString, err := receivedMessages[0].AsString() + decodedString, err := receivedMessages[0].Text() Expect(err).ToNot(HaveOccurred()) Expect(decodedString).To(Equal("testMessage")) - err = subscriber.Ack(ctx, receivedMessages.GetAckIDs()) + err = subscriber.Ack(ctx, receivedMessages.AckIDs()) Expect(err).ToNot(HaveOccurred()) cancel() @@ -199,10 +199,10 @@ var _ = Describe("PullJob", func() { callback := func(ctx context.Context, messages pubsub.PullMessages) { defer GinkgoRecover() Expect(messages).To(HaveLen(1)) - decoded, err := messages[0].AsString() + decoded, err := messages[0].Text() Expect(err).ToNot(HaveOccurred()) Expect(decoded).To(Equal("testMessage")) - err = subscriber.Ack(ctx, messages.GetAckIDs()) + err = subscriber.Ack(ctx, messages.AckIDs()) Expect(err).ToNot(HaveOccurred()) callbackInvoked.Store(true) cancel() From 1119188aa98ae81ae6256ccbfe6a4c1804939a22 Mon Sep 17 00:00:00 2001 From: Jennifer Linnenberg Date: Tue, 30 Jun 2026 13:46:25 +0200 Subject: [PATCH 5/5] feat: renamed funcs --- README.md | 2 +- example/example_pull.go | 2 +- pkg/pubsub/publisher.go | 10 ++++------ pkg/pubsub/subscriber.go | 10 ++++------ pkg/pubsub/subscriber_test.go | 6 +++--- 5 files changed, 13 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index a4acf8c..127c485 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ 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].Text() + msg, err := pulledMessages[i].DecodeString() if err != nil { log.Fatalf("Error converting message to string: %v", err) } diff --git a/example/example_pull.go b/example/example_pull.go index e3d173e..79e80e5 100644 --- a/example/example_pull.go +++ b/example/example_pull.go @@ -39,7 +39,7 @@ func pull() { log.Printf("Successfully pulled message: %v", pulledMessages) for i := 0; i < len(pulledMessages); i++ { - msg, err := pulledMessages[i].Text() + msg, err := pulledMessages[i].DecodeString() if err != nil { log.Fatalf("Error converting message to string: %v", err) } diff --git a/pkg/pubsub/publisher.go b/pkg/pubsub/publisher.go index 3370921..09a259c 100644 --- a/pkg/pubsub/publisher.go +++ b/pkg/pubsub/publisher.go @@ -23,7 +23,7 @@ type Publisher struct { // 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, error) { +func NewPublisher(topicID uuid.UUID, opts ...Option) *Publisher { cfg := &clientConfig{ httpClient: http.DefaultClient, host: "pubsub.eu01.onstackit.cloud", @@ -36,13 +36,11 @@ func NewPublisher(topicID uuid.UUID, opts ...Option) (*Publisher, error) { topicURL := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicID.String(), cfg.host)} - dataplane, err := api.NewClientWithResponses( + // 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), ) - if err != nil { - return nil, NewConfigurationError("failed to initialize api dataplane client", err) - } publisher := &Publisher{ TopicID: topicID, @@ -52,7 +50,7 @@ func NewPublisher(topicID uuid.UUID, opts ...Option) (*Publisher, error) { dataplane: dataplane, } - return publisher, nil + return publisher } // PublishStrings acts as a lightweight adapter converting string slice to byte slices, diff --git a/pkg/pubsub/subscriber.go b/pkg/pubsub/subscriber.go index bd35412..c8433e7 100644 --- a/pkg/pubsub/subscriber.go +++ b/pkg/pubsub/subscriber.go @@ -25,7 +25,7 @@ type Subscriber struct { // 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, error) { +func NewSubscriber(topicID uuid.UUID, subscriptionID uuid.UUID, opts ...Option) *Subscriber { cfg := &clientConfig{ httpClient: http.DefaultClient, host: "pubsub.eu01.onstackit.cloud", @@ -38,13 +38,11 @@ func NewSubscriber(topicID uuid.UUID, subscriptionID uuid.UUID, opts ...Option) topicURL := url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s", topicID.String(), cfg.host)} - dataplane, err := api.NewClientWithResponses( + // 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), ) - if err != nil { - return nil, NewConfigurationError("failed to initialize api dataplane client", err) - } subscriber := &Subscriber{ SubscriptionID: subscriptionID, @@ -54,7 +52,7 @@ func NewSubscriber(topicID uuid.UUID, subscriptionID uuid.UUID, opts ...Option) dataplane: dataplane, } - return subscriber, nil + return subscriber } func (s *Subscriber) Ack(ctx context.Context, ids []string) error { diff --git a/pkg/pubsub/subscriber_test.go b/pkg/pubsub/subscriber_test.go index 5e01d27..0d17ff1 100644 --- a/pkg/pubsub/subscriber_test.go +++ b/pkg/pubsub/subscriber_test.go @@ -151,7 +151,7 @@ var _ = Describe("PullJob", func() { Eventually(jobChan, "5s").Should(Receive(&receivedMessages)) Expect(receivedMessages).To(HaveLen(1)) - decodedString, err := receivedMessages[0].Text() + decodedString, err := receivedMessages[0].DecodeString() Expect(err).ToNot(HaveOccurred()) Expect(decodedString).To(Equal("testMessage")) err = subscriber.Ack(ctx, receivedMessages.AckIDs()) @@ -177,7 +177,7 @@ var _ = Describe("PullJob", func() { Eventually(jobChan, "10s").Should(Receive(&receivedMessages)) Expect(receivedMessages).To(HaveLen(1)) - decodedString, err := receivedMessages[0].Text() + decodedString, err := receivedMessages[0].DecodeString() Expect(err).ToNot(HaveOccurred()) Expect(decodedString).To(Equal("testMessage")) err = subscriber.Ack(ctx, receivedMessages.AckIDs()) @@ -199,7 +199,7 @@ var _ = Describe("PullJob", func() { callback := func(ctx context.Context, messages pubsub.PullMessages) { defer GinkgoRecover() Expect(messages).To(HaveLen(1)) - decoded, err := messages[0].Text() + decoded, err := messages[0].DecodeString() Expect(err).ToNot(HaveOccurred()) Expect(decoded).To(Equal("testMessage")) err = subscriber.Ack(ctx, messages.AckIDs())