java-microservice

## What I Do Provide guidance for developing and maintaining the crypto-scout-collector microservice, a Java 25 Maven application that consumes crypto market data from RabbitMQ Streams and persists to TimescaleDB. ## Core Components ### Modules ActiveJ DI modules for separation of concerns: ```java // CoreModule - reactor and virtual-thread executor // WebModule - HTTP server with /health endpoint // CollectorModule - DI wiring for repositories and services ``` ### StreamService Consumes from RabbitMQ Streams with database-backed offset management: ```java // Subscribes to: crypto-scout-stream, bybit-stream // External offset tracking in crypto_scout.stream_offsets // Dispatches to: CryptoScoutService, BybitStreamService, AnalystService ``` ### BybitStreamService Processes Bybit market data with batching: ```java // Spot data: PMST source - klines, tickers, trades, order books // Linear data: PML source - klines, tickers, trades, order books, liquidations // Batching: Configurable batch size and flush interval // Exactly-once: Transactional data+offset writes ``` ### CryptoScoutService Processes CMC data with batching: ```java // Fear & Greed Index (FGI source) // BTC/USD daily klines (BTC_USD_1D source) // BTC/USD weekly klines (BTC_USD_1W source) // Batching and transactional offset updates ``` ### Repository Pattern JDBC/HikariCP repositories for database access: ```java // BybitSpotRepository - spot market data // BybitLinearRepository - linear/perps market data // CryptoScoutRepository - CMC FGI and klines // StreamOffsetsRepository - offset management // AnalystRepository - technical analysis data ``` ### AmqpConsumer/AmqpPublisher AMQP integration for control messages: ```java // Consumer: receives data queries from collector-queue // Publisher: sends responses to chatbot-queue // Automatic reconnection with exponential backoff ``` ### DataService Handles AMQP request/response: ```java // Processes incoming query messages // Coordinates repository calls // Formats and publishes responses ``` ### Health Endpoint ```java // GET /health - returns JSON status // {"status":"UP","database":{"status":"UP"},"amqp":{"status":"UP"}} // HTTP 200 when healthy, HTTP 503 when degraded ``` ## Architecture Patterns ### Batching Service Pattern ```java public final class ExampleService extends AbstractReactive implements ReactiveService { private final Queue<OffsetPayload<Map<String, Object>>> buffer = new ConcurrentLinkedQueue<>(); private final AtomicBoolean flushInProgress = new AtomicBoolean(false); private final AtomicBoolean running = new AtomicBoolean(false); @Override public Promise<Void> start() { running.set(true); reactor.delayBackground(flushIntervalMs, this::scheduledFlush); return Promise.complete(); } public Promise<Void> save(final Payload<Map<String, Object>> payload, final long offset) { buffer.add(OffsetPayload.of(payload, offset)); if (buffer.size() >= batchSize) { return flush(); } return Promise.complete(); } private Promise<Void> flush() { // Drain buffer, batch insert, update offset transactionally } } ``` ### Repository Pattern ```java public final class ExampleRepository { private final DataSource dataSource; public int saveBatch(final List<Map<String, Object>> items, final long offset) throws SQLException { try (final var conn = dataSource.getConnection()) { conn.setAutoCommit(false); try { // Insert data batch final var count = insertBatch(conn, items); // Update offset upsertOffset(conn, stream, offset); conn.commit(); return count; } catch (final SQLException e) { conn.rollback(); throw e; } } } } ``` ## Configuration All settings via system properties or environment variables: | Property | Default | Description | |----------|---------|-------------| | `server.port` | `8081` | HTTP server port | | `amqp.rabbitmq.host` | `localhost` | RabbitMQ host | | `amqp.stream.port` | `5552` | RabbitMQ Streams port | | `amqp.rabbitmq.username` | `crypto_scout_mq` | RabbitMQ user | | `amqp.rabbitmq.password` | (empty) | RabbitMQ password (required) | | `amqp.bybit.stream` | `bybit-stream` | Bybit data stream name | | `amqp.crypto.scout.stream` | `crypto-scout-stream` | CMC data stream name | | `jdbc.datasource.url` | `jdbc:postgresql://localhost:5432/crypto_scout` | Database URL | | `jdbc.datasource.username` | `crypto_scout_db` | Database user | | `jdbc.datasource.password` | (empty) | Database password (required) | | `jdbc.bybit.batch-size` | `1000` | Bybit batch size (1-10000) | | `jdbc.bybit.flush-interval-ms` | `1000` | Bybit flush interval | | `jdbc.crypto.scout.batch-size` | `100` | CMC batch size (1-10000) | | `jdbc.crypto.scout.flush-interval-ms` | `1000` | CMC flush interval | ## When to Use Me Use this skill when: - Implementing new stream consumers or services - Creating new database repositories - Understanding the microservice architecture - Working with RabbitMQ Streams consumption - Implementing batch processing patterns - Adding health checks or observability features - Configuring database connection pooling - Understanding offset management patterns --- > Converted and distributed by [TomeVault](https://tomevault.io/claim/akarazhev) — claim your Tome and manage your conversions. <!-- tomevault:4.0:skill_md:2026-04-15 -->