Skip to content

Commit b68005f

Browse files
mvallimmvallim
committed
chore: update documentation
Signed-off-by: Marcos Tischer Vallim <tischer@gmail.com>
1 parent a2b2b40 commit b68005f

6 files changed

Lines changed: 787 additions & 33 deletions

File tree

‎.opencode/skills/documentation/SKILL.md‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
---
2+
name: "GenerateMarkdownDocs"
3+
description: "Skill that analyzes code and generates technical documentation in Markdown."
4+
scope: "workspace"
5+
commands:
6+
- "generate docs"
7+
- "document project"
8+
author: "mvallim"
9+
version: "1.0.0"
10+
---
11+
12+
## Objective
13+
Analyze the codebase and generate README.md files or technical guides.
14+
15+
## Instructions
16+
- Use hierarchical headings (#, ##, ###).
17+
- List dependencies in code blocks.
18+
- Include usage examples.

‎GUIDE.md‎

Lines changed: 285 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,285 @@
1+
# Amazon SNS Java Messaging Library — Technical Guide
2+
3+
## Architecture Overview
4+
5+
The library provides an asynchronous, batched messaging client for Amazon SNS, supporting both AWS SDK v1 and v2. It is organized as a multi-module Maven project:
6+
7+
| Module | Artifact | Purpose |
8+
|---|---|---|
9+
| `amazon-sns-java-messaging-lib-template` | *(internal)* | SDK-agnostic core: batching, queuing, threading, metrics |
10+
| `amazon-sns-java-messaging-lib-v1` | `amazon-sns-java-messaging-lib-v1` | AWS SDK v1 implementation (`AmazonSNS` client) |
11+
| `amazon-sns-java-messaging-lib-v2` | `amazon-sns-java-messaging-lib-v2` | AWS SDK v2 implementation (`SnsClient`) |
12+
13+
### Core Components
14+
15+
```
16+
┌──────────────────────────────────────────────────────────┐
17+
│ AmazonSnsTemplate<E> │
18+
├──────────────────────────────────────────────────────────┤
19+
│ ┌──────────────────────┐ ┌────────────────────────┐ │
20+
│ │ AmazonSnsProducer<E> │ │ AmazonSnsConsumer<R,O> │ │
21+
│ │ (AbstractProducer) │ │ (AbstractConsumer) │ │
22+
│ │ │ │ │ │
23+
│ │ - BlockingQueue │ │ - ScheduledExecutor │ │
24+
│ │ - PendingRequests │ │ - Batching Logic │ │
25+
│ └──────────┬───────────┘ └───────────┬────────────┘ │
26+
│ │ │ │
27+
│ send(E) publishBatch(...) │
28+
└─────────────┼────────────────────────────┼───────────────┘
29+
│ │
30+
▼ ▼
31+
┌──────────────────────────────────────────┐
32+
│ Amazon SNS (v1/v2) │
33+
└──────────────────────────────────────────┘
34+
```
35+
36+
- **`AmazonSnsTemplate`** — Main entry point. Created via a fluent builder (`AmazonSnsTemplate.builder(snsClient, topicProperty)`).
37+
- **`AmazonSnsProducer`** — Accepts messages into a `BlockingQueue`, tracks pending futures, and returns `ListenableFuture` results.
38+
- **`AmazonSnsConsumer`** — Scheduled drainer that pulls messages from the queue at `linger` intervals, batches them (respecting count and 256KB size limits), and publishes via the SDK's `publishBatch` API.
39+
40+
---
41+
42+
## Batching Behavior
43+
44+
Messages are accumulated in a `BlockingQueue` and drained periodically by a scheduled executor.
45+
46+
- **Linger**: Time (ms) to wait before flushing the batch. Resets on each new message arrival.
47+
- **Max batch size**: Maximum number of messages per `publishBatch` call.
48+
- **256KB limit**: Each batch request must not exceed the SNS payload limit. Messages exceeding 256KB individually throw `MaximumAllowedMessageException`.
49+
- **Memory**: The buffer stores up to `maximumPoolSize × maxBatchSize` messages internally (backed by the `BlockingQueue`).
50+
51+
For **FIFO** topics, messages are published **synchronously** on a single-threaded executor to preserve ordering. For **standard** topics, publishing is **asynchronous** via a multi-threaded executor.
52+
53+
---
54+
55+
## Message Flow
56+
57+
1. User calls `template.send(RequestEntry<E>)`
58+
2. Producer serializes the message payload to JSON (via Jackson `ObjectMapper`)
59+
3. Producer enqueues the serialized entry into a `BlockingQueue` and registers a `ListenableFuture`
60+
4. Consumer's scheduled task drains the queue at `linger` intervals, building a `PublishBatchRequest`
61+
5. Consumer calls `publishBatch()` on the SNS client
62+
6. On success: individual `ResponseSuccessEntry` results are matched back to futures by message ID
63+
7. On failure: `ResponseFailEntry` objects complete the corresponding futures with error details
64+
65+
---
66+
67+
## Dependencies
68+
69+
### Template Module (shared)
70+
71+
```xml
72+
<dependencies>
73+
<dependency>org.slf4j:slf4j-api:2.0.6</dependency>
74+
<dependency>org.apache.commons:commons-collections4:4.5.0</dependency>
75+
<dependency>org.apache.commons:commons-lang3:3.20.0</dependency>
76+
<dependency>com.fasterxml.jackson.core:jackson-databind:2.16.1</dependency>
77+
<dependency>io.micrometer:micrometer-core:1.16.3</dependency>
78+
<dependency>org.projectlombok:lombok:1.18.42 (provided)</dependency>
79+
</dependencies>
80+
```
81+
82+
### AWS SDK v1 Module
83+
84+
```xml
85+
<dependency>
86+
<groupId>com.amazonaws</groupId>
87+
<artifactId>aws-java-sdk-sns</artifactId>
88+
<version>1.12.661</version>
89+
</dependency>
90+
```
91+
92+
### AWS SDK v2 Module
93+
94+
```xml
95+
<dependency>
96+
<groupId>software.amazon.awssdk</groupId>
97+
<artifactId>sns</artifactId>
98+
<version>2.20.162</version>
99+
</dependency>
100+
```
101+
102+
### Test Dependencies
103+
104+
```xml
105+
<dependency>org.junit.jupiter:junit-jupiter:5.10.2 (test)</dependency>
106+
<dependency>org.mockito:mockito-core:4.11.0 (test)</dependency>
107+
<dependency>org.awaitility:awaitility:4.2.2 (test)</dependency>
108+
<dependency>org.assertj:assertj-core:3.24.2 (test)</dependency>
109+
<dependency>org.testcontainers:localstack:1.20.4 (test)</dependency>
110+
```
111+
112+
---
113+
114+
## Configuration Reference
115+
116+
### TopicProperty
117+
118+
| Property | Type | Default | Description |
119+
|---|---|---|---|
120+
| `fifo` | `boolean` | `false` | Whether the SNS topic is FIFO |
121+
| `topicArn` | `String` | — | The SNS topic ARN |
122+
| `maximumPoolSize` | `int` | — | Max threads for the producer pool |
123+
| `maxBatchSize` | `int` | — | Max messages per batch request |
124+
| `linger` | `long` (ms) | — | Time to wait before flushing a batch |
125+
126+
**Note**: The in-memory buffer size = `maximumPoolSize × maxBatchSize`. Large values consume proportionally more memory.
127+
128+
---
129+
130+
## Usage Examples
131+
132+
### 1. Setup with Builder (Recommended)
133+
134+
```java
135+
// For AWS SDK v1 — AmazonSNS client
136+
// For AWS SDK v2 — SnsClient
137+
138+
TopicProperty topicProperty = TopicProperty.builder()
139+
.fifo(false)
140+
.linger(100L)
141+
.maxBatchSize(10)
142+
.maximumPoolSize(5)
143+
.topicArn("arn:aws:sns:us-east-2:000000000000:topic")
144+
.build();
145+
146+
AmazonSnsTemplate<MyMessage> template = AmazonSnsTemplate.builder(snsClient, topicProperty)
147+
.meterRegistry(new SimpleMeterRegistry())
148+
.topicRequests(new RingBufferBlockingQueue<>(1024))
149+
.build();
150+
```
151+
152+
### 2. Sending a Standard Message
153+
154+
```java
155+
template.send(
156+
RequestEntry.<MyMessage>builder()
157+
.withValue(new MyMessage("hello"))
158+
.withMessageHeaders(Map.of("source", "app-1"))
159+
.build()
160+
);
161+
```
162+
163+
### 3. Sending a FIFO Message
164+
165+
```java
166+
template.send(
167+
RequestEntry.<MyMessage>builder()
168+
.withValue(new MyMessage("ordered-msg"))
169+
.withGroupId("my-group-id")
170+
.withDeduplicationId(UUID.randomUUID().toString())
171+
.build()
172+
);
173+
```
174+
175+
### 4. Async Callbacks
176+
177+
```java
178+
template.send(requestEntry)
179+
.addCallback(
180+
success -> log.info("Sent: {}", success.getMessageId()),
181+
failure -> log.error("Failed: {} [{}]", failure.getMessage(), failure.getCode())
182+
);
183+
```
184+
185+
### 5. Await Completion and Shutdown
186+
187+
```java
188+
template.send(requestEntry);
189+
template.await().thenRun(template::shutdown).join();
190+
```
191+
192+
### 6. Custom ObjectMapper and BlockingQueue
193+
194+
```java
195+
AmazonSnsTemplate<MyMessage> template = new AmazonSnsTemplate<>(
196+
amazonSNS,
197+
topicProperty,
198+
new LinkedBlockingQueue<>(100),
199+
new ObjectMapper()
200+
);
201+
```
202+
203+
---
204+
205+
## Metrics (Micrometer)
206+
207+
The library integrates with Micrometer and records the following metrics when a `MeterRegistry` is provided via the builder's `.meterRegistry(registry)`.
208+
209+
### SNS Publish Metrics
210+
211+
Tags: `topic` = `<topicArn>`
212+
213+
| Metric Name | Type | Description | Config |
214+
|---|---|---|---|
215+
| `sns.publish.attempts` | `Counter` | Total number of SNS PublishBatch calls attempted | — |
216+
| `sns.publish.success` | `Counter` | Individual SNS messages acknowledged as successful | — |
217+
| `sns.publish.failure` | `Counter` | Individual SNS messages that failed | Dynamic tags: `error_code`, `error_type` |
218+
| `sns.publish.duration` | `Timer` | End-to-end latency of SNS PublishBatch calls | Percentiles: 0.5, 0.95, 0.99 |
219+
| `sns.publish.batch.size` | `DistributionSummary` | Number of entries per SNS PublishBatch request | — |
220+
| `sns.publish.inflight` | `Gauge` | PublishBatches currently in progress | Backed by `AtomicInteger` |
221+
222+
The `sns.publish.failure` counter is created dynamically with additional `error_code` (AWS error code string) and `error_type` (amazon_service_exception or unknown) tags.
223+
224+
### Blocking Queue Metrics
225+
226+
Tags: `name` = `<queueName>`
227+
228+
| Metric Name | Type | Description | Config |
229+
|---|---|---|---|
230+
| `blocking.queue.puts.total` | `Counter` | Total number of successful put operations | — |
231+
| `blocking.queue.puts.failed` | `Counter` | Total number of put operations that threw an exception | — |
232+
| `blocking.queue.put.duration` | `Timer` | Latency of put operations (including wait time when queue is full) | Percentile histogram |
233+
| `blocking.queue.takes.total` | `Counter` | Total number of successful take operations | — |
234+
| `blocking.queue.takes.failed` | `Counter` | Total number of take operations that threw an exception | — |
235+
| `blocking.queue.take.duration` | `Timer` | Latency of take operations (including wait time when queue is empty) | Percentile histogram |
236+
| `blocking.queue.size` | `Gauge` | Current number of elements in the queue | Calls `delegate.size()` |
237+
238+
### Executor Metrics
239+
240+
Tags: `name` = `<executorName>`
241+
242+
| Metric Name | Type | Description | Config |
243+
|---|---|---|---|
244+
| `executor.active` | `Gauge` | Number of tasks currently being executed by pool threads | Backed by `AtomicInteger` |
245+
| `executor.tasks.succeeded` | `Counter` | Total number of tasks that completed without throwing an exception | — |
246+
| `executor.tasks.failed` | `Counter` | Total number of tasks that completed by throwing an exception | — |
247+
| `executor.task.duration` | `Timer` | Wall-clock duration of each task execution | — |
248+
249+
---
250+
251+
## Threading Model
252+
253+
- **Standard topics**: Uses `AmazonSnsThreadPoolExecutor` with `SynchronousQueue`, zero core threads, and `BlockingSubmissionPolicy` (30s blocking timeout). Threads are created on demand.
254+
- **FIFO topics**: Single-threaded executor to guarantee message ordering.
255+
- **Consumer scheduler**: A `ScheduledExecutorService` with a single daemon thread drains the queue at each `linger` interval.
256+
257+
---
258+
259+
## Exception Handling
260+
261+
| Exception | Condition |
262+
|---|---|
263+
| `MaximumAllowedMessageException` | A single message exceeds the 256KB SNS payload limit |
264+
| SDK exceptions (`AmazonServiceException` / `AwsServiceException`) | Service-side errors during `publishBatch` |
265+
| SDK exceptions (`AmazonClientException` / `AwsClientException`) | Client-side errors (network, serialization) |
266+
267+
Failed messages are delivered to the failure callback with:
268+
- `messageId` — the original request ID
269+
- `code` — the error code
270+
- `message` — error description
271+
- `senderFault` — whether the error is a client or server fault
272+
273+
---
274+
275+
## Testing
276+
277+
```bash
278+
# Run unit tests
279+
mvn test
280+
281+
# Run integration tests (requires Docker for LocalStack)
282+
mvn verify -P integration-test
283+
```
284+
285+
The integration tests use [Testcontainers](https://testcontainers.com) with `localstack/localstack:3.4.0` to spin up real SNS and SQS services. Messages are verified by subscribing an SQS queue to the SNS topic and polling for delivery.

0 commit comments

Comments
 (0)