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:

// 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:

// 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:

// 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:

// 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:

// 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:

// Consumer: receives data queries from collector-queue
// Publisher: sends responses to chatbot-queue
// Automatic reconnection with exponential backoff

DataService

Handles AMQP request/response:

// Processes incoming query messages
// Coordinates repository calls
// Formats and publishes responses

Health Endpoint

// 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

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

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:

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 — claim your Tome and manage your conversions.