diff --git a/README.md b/README.md index 46be87b..d8bb649 100644 --- a/README.md +++ b/README.md @@ -1,25 +1,27 @@ # pubsub-sdk-java -Kotlin SDK for the **STACKIT PubSub Data Plane API**. +[![Maven Central](https://img.shields.io/maven-central/v/cloud.stackit.pubsub.sdk/pubsub-sdk-java)](https://central.sonatype.com/artifact/cloud.stackit.pubsub.sdk/pubsub-sdk-java) + +Kotlin/Java SDK for the **STACKIT PubSub Data Plane API**. This SDK covers the **data plane only** (publish, pull, ack, nack). Creating/deleting topics and subscriptions, and granting publisher/subscriber access, is done via the separate STACKIT PubSub **control plane** REST API or the STACKIT Portal. ## Install -Every GitHub Release publishes `pubsub-sdk-java` to Maven Central automatically (see `.github/workflows/release.yml`). Once a release is cut: +The SDK is published to [Maven Central](https://central.sonatype.com/artifact/cloud.stackit.pubsub.sdk/pubsub-sdk-java) as `cloud.stackit.pubsub.sdk:pubsub-sdk-java`. Every GitHub Release publishes a new version automatically (see `.github/workflows/release.yml`); check the badge above or the [releases page](https://github.com/stackitcloud/pubsub-sdk-java/releases) for the latest version. **Maven** (`pom.xml`): ```xml cloud.stackit.pubsub.sdk pubsub-sdk-java - 1.0.0 + 0.1.1 ``` **Gradle** (`build.gradle.kts`): ```kotlin -implementation("cloud.stackit.pubsub.sdk:pubsub-sdk-java:1.0.0") +implementation("cloud.stackit.pubsub.sdk:pubsub-sdk-java:0.1.1") ``` Requires JDK 21+. Runtime dependencies: `cloud.stackit.sdk:core`, `java-jwt`, `okhttp`, `gson`. @@ -128,3 +130,10 @@ Standalone example programs demonstrating common patterns are in [`examples/`](e - `BatchExample.java` — batch publishing - `PurgeExample.java` — topic purging - `RetryExample.java` — custom retry configuration + +The examples depend on the published SDK version set in `examples/pom.xml` (`pubsub-sdk.version`). To run them against your local checkout instead: + +```sh +./gradlew publishToMavenLocal -PreleaseVersion=0.0.0-local +cd examples && mvn compile -Dpubsub-sdk.version=0.0.0-local +``` diff --git a/examples/pom.xml b/examples/pom.xml index d22938d..d74ea8a 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -29,13 +29,14 @@ 21 21 UTF-8 + 0.1.1 - cloud.stackit.sdk + cloud.stackit.pubsub.sdk pubsub-sdk-java - 1.0-SNAPSHOT + ${pubsub-sdk.version} cloud.stackit.sdk diff --git a/examples/src/main/java/cloud/stackit/pubsub/examples/BasicExample.java b/examples/src/main/java/cloud/stackit/pubsub/examples/BasicExample.java index 3d7a39e..2fb3d0e 100644 --- a/examples/src/main/java/cloud/stackit/pubsub/examples/BasicExample.java +++ b/examples/src/main/java/cloud/stackit/pubsub/examples/BasicExample.java @@ -18,7 +18,7 @@ import cloud.stackit.pubsub.sdk.auth.AuthConfig; import cloud.stackit.pubsub.sdk.client.PubSubClient; -import cloud.stackit.pubsub.sdk.errors.PubSubAuthError; +import cloud.stackit.pubsub.sdk.errors.PubSubError; import cloud.stackit.pubsub.sdk.models.Message; import cloud.stackit.pubsub.sdk.models.PublishMessage; @@ -40,18 +40,10 @@ public static void main(String[] args) { try { // Initialize auth config with service account key file path - AuthConfig authConfig = new AuthConfig( - null, - "sa-key.json", - null, - "https://authorization.stackit.cloud/auth/token" - ); + AuthConfig authConfig = new AuthConfig(null, "sa-key.json"); // Create the client with topic ID and auth config - PubSubClient client = new PubSubClient( - TOPIC_ID, - authConfig - ); + PubSubClient client = new PubSubClient(TOPIC_ID, authConfig); // Publish a message String payload = "Hello, PubSub!"; @@ -80,7 +72,7 @@ public static void main(String[] args) { System.out.println("Subscriber ack calls: " + subscriber.getAckCalls()); - } catch (PubSubAuthError e) { + } catch (PubSubError e) { System.err.println("PubSub error: " + e.getMessage()); if (e.getCause() != null) { System.err.println(" Caused by: " + e.getCause().getMessage()); diff --git a/examples/src/main/java/cloud/stackit/pubsub/examples/BatchExample.java b/examples/src/main/java/cloud/stackit/pubsub/examples/BatchExample.java index 8e7c6a6..24f60e3 100644 --- a/examples/src/main/java/cloud/stackit/pubsub/examples/BatchExample.java +++ b/examples/src/main/java/cloud/stackit/pubsub/examples/BatchExample.java @@ -18,7 +18,7 @@ import cloud.stackit.pubsub.sdk.auth.AuthConfig; import cloud.stackit.pubsub.sdk.client.PubSubClient; -import cloud.stackit.pubsub.sdk.errors.PubSubAuthError; +import cloud.stackit.pubsub.sdk.errors.PubSubError; import cloud.stackit.pubsub.sdk.models.PublishMessage; import java.util.Base64; @@ -37,12 +37,7 @@ public static void main(String[] args) { System.out.println("--- Example 3: Batch Publishing ---"); try { - AuthConfig authConfig = new AuthConfig( - null, - "sa-key.json", - null, - "https://authorization.stackit.cloud/auth/token" - ); + AuthConfig authConfig = new AuthConfig(null, "sa-key.json"); PubSubClient client = new PubSubClient(TOPIC_ID, authConfig); var publisher = client.getPublisher(); @@ -58,7 +53,7 @@ public static void main(String[] args) { System.out.println("Total messages sent: " + publisher.getMessagesSent()); System.out.println("Total publish calls: " + publisher.getPublishCalls()); - } catch (PubSubAuthError e) { + } catch (PubSubError e) { System.err.println("PubSub error: " + e.getMessage()); } } diff --git a/examples/src/main/java/cloud/stackit/pubsub/examples/ErrorHandlingExample.java b/examples/src/main/java/cloud/stackit/pubsub/examples/ErrorHandlingExample.java index d4230bd..052c659 100644 --- a/examples/src/main/java/cloud/stackit/pubsub/examples/ErrorHandlingExample.java +++ b/examples/src/main/java/cloud/stackit/pubsub/examples/ErrorHandlingExample.java @@ -18,7 +18,7 @@ import cloud.stackit.pubsub.sdk.auth.AuthConfig; import cloud.stackit.pubsub.sdk.client.PubSubClient; -import cloud.stackit.pubsub.sdk.errors.PubSubAuthError; +import cloud.stackit.pubsub.sdk.errors.PubSubError; import cloud.stackit.pubsub.sdk.models.PublishMessage; import java.util.Base64; @@ -37,12 +37,7 @@ public static void main(String[] args) { System.out.println("--- Example 2: Error Handling ---"); try { - AuthConfig authConfig = new AuthConfig( - null, - "sa-key.json", - null, - null - ); + AuthConfig authConfig = new AuthConfig(null, "sa-key.json"); PubSubClient client = new PubSubClient(TOPIC_ID, authConfig); var publisher = client.getPublisher(); @@ -51,14 +46,17 @@ public static void main(String[] args) { PublishMessage message = new PublishMessage(Base64.getEncoder().encodeToString("test".getBytes())); publisher.publish(List.of(message)); - } catch (PubSubAuthError.MissingCredentials e) { + } catch (PubSubError.AuthError.MissingCredentials e) { System.err.println("Missing credentials - please configure a service account key"); - } catch (PubSubAuthError.TokenAcquisitionFailed e) { + } catch (PubSubError.AuthError.TokenAcquisitionFailed e) { System.err.println("Token acquisition failed: " + e.getMessage()); // This catches HTTP errors (4xx, 5xx) and network errors if (e.getCause() != null) { System.err.println(" Caused by: " + e.getCause().getClass().getSimpleName()); } + } catch (PubSubError.HttpError e) { + // Thrown for non-retryable API responses (400, 401, 403, 404, 413) and exhausted retries on 5xx + System.err.println("API error (" + e.getClass().getSimpleName() + "): " + e.getMessage()); } catch (Exception e) { System.err.println("Unexpected error: " + e.getMessage()); } diff --git a/examples/src/main/java/cloud/stackit/pubsub/examples/PurgeExample.java b/examples/src/main/java/cloud/stackit/pubsub/examples/PurgeExample.java index f8a996f..1ef88dd 100644 --- a/examples/src/main/java/cloud/stackit/pubsub/examples/PurgeExample.java +++ b/examples/src/main/java/cloud/stackit/pubsub/examples/PurgeExample.java @@ -18,7 +18,7 @@ import cloud.stackit.pubsub.sdk.auth.AuthConfig; import cloud.stackit.pubsub.sdk.client.PubSubClient; -import cloud.stackit.pubsub.sdk.errors.PubSubAuthError; +import cloud.stackit.pubsub.sdk.errors.PubSubError; import cloud.stackit.pubsub.sdk.models.PublishMessage; import java.util.Base64; @@ -37,12 +37,7 @@ public static void main(String[] args) { System.out.println("--- Example 4: Topic Purging ---"); try { - AuthConfig authConfig = new AuthConfig( - null, - "sa-key.json", - null, - null - ); + AuthConfig authConfig = new AuthConfig(null, "sa-key.json"); PubSubClient client = new PubSubClient(TOPIC_ID, authConfig); var publisher = client.getPublisher(); @@ -52,7 +47,7 @@ public static void main(String[] args) { System.out.println("Topic purged successfully"); System.out.println("Purge calls: " + publisher.getPurgeCalls()); - } catch (PubSubAuthError e) { + } catch (PubSubError e) { System.err.println("PubSub error: " + e.getMessage()); } } diff --git a/examples/src/main/java/cloud/stackit/pubsub/examples/RetryExample.java b/examples/src/main/java/cloud/stackit/pubsub/examples/RetryExample.java index b6bdf19..39d8319 100644 --- a/examples/src/main/java/cloud/stackit/pubsub/examples/RetryExample.java +++ b/examples/src/main/java/cloud/stackit/pubsub/examples/RetryExample.java @@ -18,7 +18,7 @@ import cloud.stackit.pubsub.sdk.auth.AuthConfig; import cloud.stackit.pubsub.sdk.client.PubSubClient; -import cloud.stackit.pubsub.sdk.errors.PubSubAuthError; +import cloud.stackit.pubsub.sdk.errors.PubSubError; import cloud.stackit.pubsub.sdk.models.PublishMessage; import cloud.stackit.pubsub.sdk.retry.RetryConfig; @@ -38,12 +38,7 @@ public static void main(String[] args) { System.out.println("--- Example 5: Custom Retry Configuration ---"); try { - AuthConfig authConfig = new AuthConfig( - null, - "sa-key.json", - null, - null - ); + AuthConfig authConfig = new AuthConfig(null, "sa-key.json"); PubSubClient client = new PubSubClient(TOPIC_ID, authConfig); @@ -70,7 +65,7 @@ public static void main(String[] args) { System.out.println("Published with custom retry config"); System.out.println("Messages sent: " + publisher.getMessagesSent()); - } catch (PubSubAuthError e) { + } catch (PubSubError e) { System.err.println("PubSub error after retries: " + e.getMessage()); // Note: After all retries are exhausted, the error includes retry information if (e.getMessage().contains("retries")) { diff --git a/src/main/kotlin/cloud/stackit/pubsub/sdk/auth/AuthConfig.kt b/src/main/kotlin/cloud/stackit/pubsub/sdk/auth/AuthConfig.kt index 32be198..2e2b5e5 100644 --- a/src/main/kotlin/cloud/stackit/pubsub/sdk/auth/AuthConfig.kt +++ b/src/main/kotlin/cloud/stackit/pubsub/sdk/auth/AuthConfig.kt @@ -18,7 +18,7 @@ package cloud.stackit.pubsub.sdk.auth import cloud.stackit.sdk.core.config.CoreConfiguration -data class AuthConfig( +data class AuthConfig @JvmOverloads constructor( val serviceAccountKey: String? = null, val serviceAccountKeyPath: String? = null, val privateKeyPath: String? = null, diff --git a/src/main/kotlin/cloud/stackit/pubsub/sdk/client/PubSubClient.kt b/src/main/kotlin/cloud/stackit/pubsub/sdk/client/PubSubClient.kt index 7b9326f..0d31377 100644 --- a/src/main/kotlin/cloud/stackit/pubsub/sdk/client/PubSubClient.kt +++ b/src/main/kotlin/cloud/stackit/pubsub/sdk/client/PubSubClient.kt @@ -39,7 +39,7 @@ import java.util.UUID * @param region API region (default: EU01). * @param environment STACKIT environment (default: PROD). */ -class PubSubClient( +class PubSubClient @JvmOverloads constructor( val topicId: UUID, val authConfig: AuthConfig? = null, val region: Region = Region.EU01, @@ -61,6 +61,7 @@ class PubSubClient( * @param retryConfig Optional retry configuration. * @return Configured PubSubPublisher. */ + @JvmOverloads fun getPublisher(retryConfig: RetryConfig? = null): PubSubPublisher { return PubSubPublisher( topicId = topicId, @@ -80,6 +81,7 @@ class PubSubClient( * @param longPullDurationMs Long pull timeout in ms (100-5000). * @return Configured PubSubSubscriber. */ + @JvmOverloads fun getSubscriber( subscriptionId: UUID, retryConfig: RetryConfig? = null, diff --git a/src/main/kotlin/cloud/stackit/pubsub/sdk/publisher/PubSubPublisher.kt b/src/main/kotlin/cloud/stackit/pubsub/sdk/publisher/PubSubPublisher.kt index db0f0d3..5b087b2 100644 --- a/src/main/kotlin/cloud/stackit/pubsub/sdk/publisher/PubSubPublisher.kt +++ b/src/main/kotlin/cloud/stackit/pubsub/sdk/publisher/PubSubPublisher.kt @@ -46,7 +46,7 @@ import java.util.concurrent.atomic.AtomicInteger * @param publishCalls Total count of publish calls made. * @param purgeCalls Total count of purge calls made. */ -class PubSubPublisher( +class PubSubPublisher @JvmOverloads constructor( private val topicId: UUID, private val baseDataplaneUrl: String, private val authProvider: AuthProvider, @@ -103,6 +103,7 @@ class PubSubPublisher( * @param messages All messages to publish. * @param batchSize Maximum messages per batch (default: 100). */ + @JvmOverloads fun publishBatch(messages: List, batchSize: Int = 100) { messages.chunked(batchSize).forEach { batch -> publish(batch) diff --git a/src/main/kotlin/cloud/stackit/pubsub/sdk/retry/RetryConfig.kt b/src/main/kotlin/cloud/stackit/pubsub/sdk/retry/RetryConfig.kt index 337c5fc..a840c0b 100644 --- a/src/main/kotlin/cloud/stackit/pubsub/sdk/retry/RetryConfig.kt +++ b/src/main/kotlin/cloud/stackit/pubsub/sdk/retry/RetryConfig.kt @@ -25,7 +25,7 @@ package cloud.stackit.pubsub.sdk.retry * @param multiplier Multiplier applied to delay after each retry. Default: 2.0 * @param jitterFactor Randomization factor (0.0-1.0) to avoid thundering herd. Default: 0.25 */ -data class RetryConfig( +data class RetryConfig @JvmOverloads constructor( val maxRetries: Int = 3, val baseDelay: Long = 500, val maxDelay: Long = 30_000, @@ -33,6 +33,7 @@ data class RetryConfig( val jitterFactor: Double = 0.25 ) { companion object { + @JvmField val Default = RetryConfig() } } diff --git a/src/main/kotlin/cloud/stackit/pubsub/sdk/subscriber/PubSubSubscriber.kt b/src/main/kotlin/cloud/stackit/pubsub/sdk/subscriber/PubSubSubscriber.kt index 12d3ef1..c1a8fd8 100644 --- a/src/main/kotlin/cloud/stackit/pubsub/sdk/subscriber/PubSubSubscriber.kt +++ b/src/main/kotlin/cloud/stackit/pubsub/sdk/subscriber/PubSubSubscriber.kt @@ -41,6 +41,7 @@ import cloud.stackit.pubsub.sdk.retry.isRetryableStatusCode import java.io.IOException import java.time.Duration import java.util.UUID +import java.util.concurrent.atomic.AtomicInteger /** * Subscriber interface for receiving messages from a STACKIT PubSub subscription. @@ -52,8 +53,11 @@ import java.util.UUID * @param maxMessages Maximum messages to pull per request. * @param longPullDurationMs Long pull timeout in milliseconds. * @param httpClient Shared HTTP client for connection reuse. + * @param pullCalls Total count of pull calls made. + * @param ackCalls Total count of ack calls made. + * @param nackCalls Total count of nack calls made. */ -class PubSubSubscriber( +class PubSubSubscriber @JvmOverloads constructor( private val subscriptionId: UUID, private val baseDataplaneUrl: String, private val authProvider: AuthProvider, @@ -81,6 +85,12 @@ class PubSubSubscriber( .writeTimeout(Duration.ofSeconds(10)) .build() + val pullCalls: AtomicInteger = AtomicInteger(0) + + val ackCalls: AtomicInteger = AtomicInteger(0) + + val nackCalls: AtomicInteger = AtomicInteger(0) + init { require(maxMessages in 1..32) { "maxMessages must be between 1 and 32" } require(longPullDurationMs in 100..5000) { "longPullDurationMs must be between 100 and 5000" } @@ -93,12 +103,15 @@ class PubSubSubscriber( * @param longPullDurationMs Override long pull duration (default: 5000ms). * @return List of messages. */ + @JvmOverloads fun pull( maxMessages: Int = this.maxMessages, longPullDurationMs: Int = this.longPullDurationMs ): List { require(maxMessages in 1..32) { "maxMessages must be between 1 and 32" } + pullCalls.incrementAndGet() + val request = Request.Builder() .url("$baseDataplaneUrl/v1alpha/subscriptions/$subscriptionId/pull") .post("{}".toRequestBody(jsonMediaType)) @@ -127,6 +140,8 @@ class PubSubSubscriber( fun ack(ackIds: List) { require(ackIds.isNotEmpty()) { "ackIds must not be empty" } + ackCalls.incrementAndGet() + val request = Request.Builder() .url("$baseDataplaneUrl/v1alpha/subscriptions/$subscriptionId/ack") .post(gson.toJson(AckMessagesRequest(ackIds)).toRequestBody(jsonMediaType)) @@ -150,6 +165,8 @@ class PubSubSubscriber( fun nack(nackIds: List) { require(nackIds.isNotEmpty()) { "nackIds must not be empty" } + nackCalls.incrementAndGet() + val request = Request.Builder() .url("$baseDataplaneUrl/v1alpha/subscriptions/$subscriptionId/nack") .post(gson.toJson(NackMessagesRequest(nackIds)).toRequestBody(jsonMediaType)) diff --git a/src/test/kotlin/cloud/stackit/pubsub/sdk/subscriber/TestSubscriber.kt b/src/test/kotlin/cloud/stackit/pubsub/sdk/subscriber/TestSubscriber.kt index bf45281..d5d5f18 100644 --- a/src/test/kotlin/cloud/stackit/pubsub/sdk/subscriber/TestSubscriber.kt +++ b/src/test/kotlin/cloud/stackit/pubsub/sdk/subscriber/TestSubscriber.kt @@ -89,6 +89,7 @@ class TestSubscriberPull { assertEquals(2, messages.size) assertEquals("ack-1", messages[0].ackId) assertEquals("Hello, World!", messages[0].text) + assertEquals(1, subscriber.pullCalls.get()) } @Test @@ -270,6 +271,9 @@ class TestSubscriberAck { val subscriber = createSubscriber(response) subscriber.ack(listOf("ack-1", "ack-2")) + + assertEquals(1, subscriber.ackCalls.get()) + assertEquals(0, subscriber.nackCalls.get()) } @Test @@ -333,6 +337,8 @@ class TestSubscriberNack { val subscriber = createSubscriber(response) subscriber.nack(listOf("nack-1", "nack-2")) + + assertEquals(1, subscriber.nackCalls.get()) } @Test