@@ -138,48 +194,187 @@
false
+
- com.googlecode.maven-download-plugin
- download-maven-plugin
- 1.13.0
+ maven-resources-plugin
+ 3.3.1
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+ 3.2.3
+
+
+ **/*Test.java
+
+
+ ${surefire.argLine}
+
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ 0.8.14
+
+
+ prepare-agent
+
+ prepare-agent
+
+
+ surefire.argLine
+
+
- swagger-ui
- generate-resources
+ prepare-agent-integration
- wget
+ prepare-agent-integration
- true
- https://github.com/swagger-api/swagger-ui/archive/master.tar.gz
- true
- ${project.build.directory}
+ failsafe.argLine
+
+
+
+ report
+ test
+
+ report
+
+
+
+ report-integration
+ verify
+
+ report-integration
+
+
+
+ check
+
+ check
+
+
+
+
+ BUNDLE
+
+
+ INSTRUCTION
+ COVEREDRATIO
+ 0.70
+
+
+ BRANCH
+ COVEREDRATIO
+ 0.60
+
+
+ CLASS
+ COVEREDRATIO
+ 0.80
+
+
+
+
+
+
+
+ merge-results
+ verify
+
+ merge
+
+
+
+
+ ${project.build.directory}
+
+ *.exec
+
+
+
+ ${project.build.directory}/jacoco/merged.exec
+
+
- maven-resources-plugin
- 3.3.1
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+ 3.2.3
- copy-swagger-resources
- process-resources
- copy-resources
+ integration-test
+ verify
-
- ${project.build.directory}/${project.artifactId}-${project.version}
-
-
- ${project.build.directory}/swagger-ui-master/dist
- true
-
- index.html
-
-
-
-
+
+
+
+
+ **/integration/*Test.java
+
+ ${failsafe.argLine} -Xmx4g -XX:+UseG1GC
+
+ ${project.basedir}/src/test/resources/logback-test.xml
+
+
+
+
+
+
+ com.diffplug.spotless
+ spotless-maven-plugin
+ 3.0.0
+
+
+
+ ${project.basedir}/spotless/java-formatter.xml
+
+
+
+
+
+
+
+ src/**/*.json
+ monitoring/**/*.json
+
+
+ 2.15.2
+
+ true
+
+
+
+
+
+
+ **/*.md
+
+
+ workshop/slides/**
+ workshop/WORKSHOP.md
+ workshop/analysis-checklist.md
+ workshop/exercises/**/*.md
+ workshop/recordings/**/*.md
+
+
+
+
+
+
+
+ check
+
+ compile
@@ -201,6 +396,88 @@
+
+
+ coverage
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ 0.8.12
+
+
+ prepare-agent
+
+ prepare-agent
+
+
+
+ report
+ test
+
+ report
+
+
+
+ XML
+ HTML
+
+
+
+
+ check
+
+ check
+
+
+
+
+ BUNDLE
+
+
+ INSTRUCTION
+ COVEREDRATIO
+ 0.15
+
+
+ CLASS
+ MISSEDCOUNT
+ 40
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ quick-test
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ **/*Test.java
+
+
+ **/integration/**
+ **/*LoadTest.java
+ **/*BenchmarkTest.java
+
+
+
+
+
+
diff --git a/postgres/init.sql b/postgres/init.sql
deleted file mode 100644
index d770781..0000000
--- a/postgres/init.sql
+++ /dev/null
@@ -1,62 +0,0 @@
-CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
-
--- Audit log table for security events
-CREATE TABLE audit_log
-(
- id VARCHAR(36) PRIMARY KEY,
- timestamp TIMESTAMP NOT NULL,
- username VARCHAR(100),
- ip_address VARCHAR(45),
- action VARCHAR(100) NOT NULL,
- resource VARCHAR(255),
- method_name VARCHAR(255),
- class_name VARCHAR(255),
- sensitivity_level VARCHAR(20),
- success BOOLEAN NOT NULL,
- duration_ms BIGINT,
- error_message VARCHAR(1000),
- user_agent VARCHAR(500),
- details TEXT
-);
-
--- Indexes for audit log queries
-CREATE INDEX idx_audit_username ON audit_log (username);
-CREATE INDEX idx_audit_timestamp ON audit_log (timestamp);
-CREATE INDEX idx_audit_action ON audit_log (action);
-CREATE INDEX idx_audit_level ON audit_log (sensitivity_level);
-
--- Patient table
-CREATE TABLE patient
-(
- id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
- first_name VARCHAR(50) NOT NULL,
- last_name VARCHAR(50) NOT NULL,
- date_of_birth DATE NOT NULL,
- ssn VARCHAR(255),
- email VARCHAR(255),
- phone VARCHAR(255),
- address VARCHAR(255),
- blood_type VARCHAR(10),
- allergies VARCHAR(500),
- medical_conditions VARCHAR(1000),
- assigned_doctor VARCHAR(100),
- department VARCHAR(50),
- created_at TIMESTAMP,
- updated_at TIMESTAMP
-);
-
-INSERT INTO patient (id, first_name, last_name, date_of_birth, ssn, email, phone, address, blood_type, allergies,
- medical_conditions, assigned_doctor, department, created_at, updated_at)
-VALUES ('a8b8b4f0-3e5b-48b0-8b4a-0e1b6d4b0b1b', 'Jane', 'Doe', '1985-05-15', '123-45-6789', 'jane.doe@email.com',
- '555-0101', '123 Main St, Springfield', 'O+', 'Penicillin', 'Hypertension', 'Dr. Smith', 'Cardiology', NOW(),
- NOW()),
- ('f8b8b4f0-3e5b-48b0-8b4a-0e1b6d4b0b1c', 'John', 'Smith', '1978-03-22', '987-65-4321', 'john.smith@email.com',
- '555-0102', '456 Oak Ave, Springfield', 'A+', 'None', 'Diabetes Type 2', 'Dr. Smith', 'Cardiology', NOW(),
- NOW()),
- ('e8b8b4f0-3e5b-48b0-8b4a-0e1b6d4b0b1d', 'Alice', 'Johnson', '1992-11-08', '456-78-9012', 'alice.j@email.com',
- '555-0103', '789 Pine Rd, Springfield', 'B-', 'Latex', 'Asthma', 'Dr. Williams', 'Emergency', NOW(), NOW()),
- ('d8b8b4f0-3e5b-48b0-8b4a-0e1b6d4b0b1e', 'Robert', 'Brown', '1965-07-30', '321-54-9876', 'rbrown@email.com',
- '555-0104', '321 Elm St, Springfield', 'AB+', 'Shellfish', 'Coronary artery disease', 'Dr. Smith', 'Cardiology',
- NOW(), NOW()),
- ('c8b8b4f0-3e5b-48b0-8b4a-0e1b6d4b0b1f', 'Emma', 'Davis', '1988-09-14', '654-32-1098', 'emma.d@email.com',
- '555-0105', '555 Maple Ln, Springfield', 'O-', 'None', 'None', 'Dr. Williams', 'Emergency', NOW(), NOW());
diff --git a/spotless/java-formatter.xml b/spotless/java-formatter.xml
new file mode 100644
index 0000000..71f63ff
--- /dev/null
+++ b/spotless/java-formatter.xml
@@ -0,0 +1,126 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/main/java/fish/payara/resource/HelloWorldResource.java b/src/main/java/fish/payara/resource/HelloWorldResource.java
index b82e4f5..0a7c26b 100644
--- a/src/main/java/fish/payara/resource/HelloWorldResource.java
+++ b/src/main/java/fish/payara/resource/HelloWorldResource.java
@@ -12,9 +12,9 @@
import org.eclipse.microprofile.metrics.annotation.Counted;
import org.eclipse.microprofile.metrics.annotation.Timed;
import org.eclipse.microprofile.openapi.annotations.Operation;
+import org.eclipse.microprofile.openapi.annotations.parameters.Parameter;
import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
import org.eclipse.microprofile.openapi.annotations.responses.APIResponses;
-import org.eclipse.microprofile.openapi.annotations.parameters.Parameter;
@Path("hello")
public class HelloWorldResource {
@@ -25,29 +25,22 @@ public class HelloWorldResource {
@GET
@Operation(summary = "Get a personalized greeting")
- @APIResponses(value = {
- @APIResponse(responseCode = "200", description = "Successful operation"),
- @APIResponse(responseCode = "400", description = "Invalid input")
- })
+ @APIResponses(value = {@APIResponse(responseCode = "200", description = "Successful operation"),
+ @APIResponse(responseCode = "400", description = "Invalid input")})
@Counted(name = "helloEndpointCount", description = "Count of calls to the hello endpoint")
@Timed(name = "helloEndpointTime", description = "Time taken to execute the hello endpoint")
- @Timeout(3000) // Timeout after 3 seconds
- @Retry(maxRetries = 3) // Retry the request up to 3 times on failure
+ @Timeout(3000)
+ @Retry(maxRetries = 3)
@Fallback(fallbackMethod = "fallbackMethod")
- public Response hello(@QueryParam("name") @Parameter(name = "name", description = "Name to include in the greeting", required = false, example = "John") String name) {
+ public Response hello(
+ @QueryParam("name") @Parameter(name = "name", description = "Name to include in the greeting", required = false, example = "John") String name) {
if ((name == null) || name.trim().isEmpty()) {
name = defaultName;
}
- return Response
- .ok(name)
- .build();
+ return Response.ok(name).build();
}
public Response fallbackMethod(@QueryParam("name") String name) {
- // Fallback logic when the hello method fails or exceeds retries
- return Response
- .ok("Fallback data")
- .build();
+ return Response.ok("Fallback data").build();
}
-
-}
\ No newline at end of file
+}
diff --git a/src/main/java/fish/payara/trader/aeron/AeronSubscriberBean.java b/src/main/java/fish/payara/trader/aeron/AeronSubscriberBean.java
index 192229d..8b38782 100644
--- a/src/main/java/fish/payara/trader/aeron/AeronSubscriberBean.java
+++ b/src/main/java/fish/payara/trader/aeron/AeronSubscriberBean.java
@@ -5,30 +5,24 @@
import io.aeron.Subscription;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
-import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
-import jakarta.ejb.Singleton;
-import jakarta.ejb.Startup;
import jakarta.enterprise.concurrent.ManagedExecutorService;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.context.Initialized;
import jakarta.enterprise.event.Observes;
import jakarta.inject.Inject;
-import org.agrona.CloseHelper;
-import org.agrona.concurrent.BackoffIdleStrategy;
-import org.agrona.concurrent.IdleStrategy;
-
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.agrona.CloseHelper;
+import org.agrona.concurrent.BackoffIdleStrategy;
+import org.agrona.concurrent.IdleStrategy;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
/**
- * Aeron Ingress Singleton Bean
- * Launches an embedded MediaDriver and subscribes to market data stream.
- * Uses SBE decoders for zero-copy message processing.
- * Runs in a dedicated thread to continuously poll for messages.
- * IMPORTANT: This must initialize BEFORE MarketDataPublisher
+ * Aeron Ingress Singleton Bean Launches an embedded MediaDriver and subscribes to market data stream. Uses SBE decoders for zero-copy message processing. Runs
+ * in a dedicated thread to continuously poll for messages. IMPORTANT: This must initialize BEFORE MarketDataPublisher
*/
@ApplicationScoped
public class AeronSubscriberBean {
@@ -47,39 +41,46 @@ public class AeronSubscriberBean {
@Inject
private MarketDataFragmentHandler fragmentHandler;
-
+
@Inject
@VirtualThreadExecutor
private ManagedExecutorService managedExecutorService;
+ @Inject
+ @ConfigProperty(name = "TRADER_INGESTION_MODE", defaultValue = "AERON")
+ private String ingestionMode;
+
void contextInitialized(@Observes @Initialized(ApplicationScoped.class) Object event) {
- init();
+ managedExecutorService.submit(this::init);
}
-
public void init() {
+ if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
+ LOGGER.info("Running in DIRECT mode - Skipping Aeron/MediaDriver initialization.");
+ return;
+ }
+
LOGGER.info("Initializing Aeron Subscriber Bean...");
try {
LOGGER.info("Launching embedded MediaDriver...");
- mediaDriver = MediaDriver.launchEmbedded(
- new MediaDriver.Context()
- .threadingMode(ThreadingMode.SHARED)
- .dirDeleteOnStart(true)
- .dirDeleteOnShutdown(true)
- );
+ // BackoffIdleStrategy on the shared MediaDriver thread ensures the
+ // driver eventually parks when idle, crossing a JVM safepoint poll.
+ // Without this, Zing/GPGC aborts under sustained load with
+ // "Checkpoint sync time longer than 200000 ms detected" because the
+ // Aeron driver thread never reaches a safepoint on its own.
+ final IdleStrategy driverIdleStrategy = new BackoffIdleStrategy(100, 10, TimeUnit.MICROSECONDS.toNanos(1), TimeUnit.MICROSECONDS.toNanos(100));
+ mediaDriver = MediaDriver.launchEmbedded(new MediaDriver.Context().threadingMode(ThreadingMode.SHARED)
+ .sharedIdleStrategy(driverIdleStrategy)
+ .dirDeleteOnStart(true)
+ .dirDeleteOnShutdown(true));
LOGGER.info("MediaDriver launched at: " + mediaDriver.aeronDirectoryName());
LOGGER.info("Connecting Aeron client...");
- aeron = Aeron.connect(
- new Aeron.Context()
- .aeronDirectoryName(mediaDriver.aeronDirectoryName())
- .errorHandler(this::onError)
- .availableImageHandler(image ->
- LOGGER.info("Available image: " + image.sourceIdentity()))
- .unavailableImageHandler(image ->
- LOGGER.info("Unavailable image: " + image.sourceIdentity()))
- );
+ aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(mediaDriver.aeronDirectoryName())
+ .errorHandler(this::onError)
+ .availableImageHandler(image -> LOGGER.info("Available image: " + image.sourceIdentity()))
+ .unavailableImageHandler(image -> LOGGER.info("Unavailable image: " + image.sourceIdentity())));
LOGGER.info("Adding subscription on channel: " + CHANNEL + ", stream: " + STREAM_ID);
subscription = aeron.addSubscription(CHANNEL, STREAM_ID);
startPolling();
@@ -92,19 +93,13 @@ public void init() {
}
}
- /**
- * Start background task to continuously poll for messages
- */
+ /** Start background task to continuously poll for messages */
private void startPolling() {
running = true;
pollingFuture = managedExecutorService.submit(() -> {
LOGGER.info("Aeron polling task started");
- final IdleStrategy idleStrategy = new BackoffIdleStrategy(
- 100, 10,
- TimeUnit.MICROSECONDS.toNanos(1),
- TimeUnit.MICROSECONDS.toNanos(100)
- );
+ final IdleStrategy idleStrategy = new BackoffIdleStrategy(100, 10, TimeUnit.MICROSECONDS.toNanos(1), TimeUnit.MICROSECONDS.toNanos(100));
while (running && !Thread.currentThread().isInterrupted()) {
try {
@@ -121,9 +116,7 @@ private void startPolling() {
});
}
- /**
- * Error handler for Aeron
- */
+ /** Error handler for Aeron */
private void onError(Throwable throwable) {
LOGGER.log(Level.SEVERE, "Aeron error occurred", throwable);
}
@@ -132,43 +125,32 @@ private void onError(Throwable throwable) {
public void shutdown() {
LOGGER.info("Shutting down Aeron Subscriber Bean...");
running = false;
-
+
if (pollingFuture != null) {
pollingFuture.cancel(true);
}
-
+
cleanup();
LOGGER.info("Aeron Subscriber Bean shut down");
}
- /**
- * Clean up all Aeron resources
- */
+ /** Clean up all Aeron resources */
private void cleanup() {
CloseHelper.quietClose(subscription);
CloseHelper.quietClose(aeron);
CloseHelper.quietClose(mediaDriver);
}
- /**
- * Get subscription statistics
- */
+ /** Get subscription statistics */
public String getStatus() {
if (subscription != null) {
- return String.format(
- "Channel: %s, Stream: %d, Images: %d, Running: %b",
- subscription.channel(),
- subscription.streamId(),
- subscription.imageCount(),
- running
- );
+ return String.format("Channel: %s, Stream: %d, Images: %d, Running: %b", subscription.channel(), subscription.streamId(), subscription.imageCount(),
+ running);
}
return "Not initialized";
}
- /**
- * Get the Aeron directory name for connecting publishers
- */
+ /** Get the Aeron directory name for connecting publishers */
public String getAeronDirectoryName() {
if (mediaDriver != null) {
return mediaDriver.aeronDirectoryName();
@@ -176,10 +158,8 @@ public String getAeronDirectoryName() {
return null;
}
- /**
- * Check if the MediaDriver is ready
- */
+ /** Check if the MediaDriver is ready */
public boolean isReady() {
return mediaDriver != null && aeron != null && subscription != null && running;
}
-}
\ No newline at end of file
+}
diff --git a/src/main/java/fish/payara/trader/aeron/MarketDataFragmentHandler.java b/src/main/java/fish/payara/trader/aeron/MarketDataFragmentHandler.java
index c3de352..577670b 100644
--- a/src/main/java/fish/payara/trader/aeron/MarketDataFragmentHandler.java
+++ b/src/main/java/fish/payara/trader/aeron/MarketDataFragmentHandler.java
@@ -1,53 +1,51 @@
package fish.payara.trader.aeron;
+import fish.payara.trader.jfr.MarketDataEvents;
import fish.payara.trader.sbe.*;
import fish.payara.trader.websocket.MarketDataBroadcaster;
import io.aeron.logbuffer.FragmentHandler;
import io.aeron.logbuffer.Header;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
-import org.agrona.DirectBuffer;
-
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.agrona.DirectBuffer;
/**
- * FragmentHandler that uses SBE Flyweights for zero-copy message decoding.
- * This handler processes Aeron fragments by:
- * 1. Decoding the SBE message header to identify message type
- * 2. Using the appropriate SBE decoder (flyweight pattern)
- * 3. Extracting data without object allocation
- * 4. Broadcasting as JSON (intentionally creating garbage for GC stress testing)
+ * FragmentHandler that uses SBE Flyweights for zero-copy message decoding. This handler processes Aeron fragments by: 1. Decoding the SBE message header to
+ * identify message type 2. Using the appropriate SBE decoder (flyweight pattern) 3. Extracting data without object allocation 4. Broadcasting as JSON
+ * (intentionally creating garbage for GC stress testing)
*/
@ApplicationScoped
public class MarketDataFragmentHandler implements FragmentHandler {
private static final Logger LOGGER = Logger.getLogger(MarketDataFragmentHandler.class.getName());
- // SBE Message Header Decoder (reusable flyweight)
private final MessageHeaderDecoder headerDecoder = new MessageHeaderDecoder();
- // SBE Message Decoders (reusable flyweights)
private final TradeDecoder tradeDecoder = new TradeDecoder();
private final QuoteDecoder quoteDecoder = new QuoteDecoder();
private final MarketDepthDecoder marketDepthDecoder = new MarketDepthDecoder();
private final OrderAckDecoder orderAckDecoder = new OrderAckDecoder();
private final HeartbeatDecoder heartbeatDecoder = new HeartbeatDecoder();
- // Statistics
private long messagesProcessed = 0;
private long messagesBroadcast = 0;
private long lastLogTime = System.currentTimeMillis();
- // Sampling: Only broadcast 1 in N messages to avoid overwhelming browser
private static final int SAMPLE_RATE = 50;
private long sampleCounter = 0;
@Inject
- private MarketDataBroadcaster broadcaster;
+ MarketDataBroadcaster broadcaster;
+
+ private final byte[] symbolBuffer = new byte[128];
+ private final StringBuilder sb = new StringBuilder(1024);
@Override
public void onFragment(DirectBuffer buffer, int offset, int length, Header header) {
+ long startTime = System.nanoTime();
+
try {
headerDecoder.wrap(buffer, offset);
@@ -61,31 +59,42 @@ public void onFragment(DirectBuffer buffer, int offset, int length, Header heade
final boolean shouldBroadcast = (sampleCounter % SAMPLE_RATE == 0);
switch (templateId) {
- case TradeDecoder.TEMPLATE_ID:
- processTrade(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
- break;
+ case TradeDecoder.TEMPLATE_ID :
+ processTrade(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
+ break;
- case QuoteDecoder.TEMPLATE_ID:
- processQuote(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
- break;
+ case QuoteDecoder.TEMPLATE_ID :
+ processQuote(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
+ break;
- case MarketDepthDecoder.TEMPLATE_ID:
- processMarketDepth(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
- break;
+ case MarketDepthDecoder.TEMPLATE_ID :
+ processMarketDepth(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
+ break;
- case OrderAckDecoder.TEMPLATE_ID:
- processOrderAck(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
- break;
+ case OrderAckDecoder.TEMPLATE_ID :
+ processOrderAck(buffer, offset, actingBlockLength, actingVersion, shouldBroadcast);
+ break;
- case HeartbeatDecoder.TEMPLATE_ID:
- processHeartbeat(buffer, offset, actingBlockLength, actingVersion);
- break;
+ case HeartbeatDecoder.TEMPLATE_ID :
+ processHeartbeat(buffer, offset, actingBlockLength, actingVersion);
+ break;
- default:
- LOGGER.warning("Unknown message template ID: " + templateId);
+ default :
+ LOGGER.warning("Unknown message template ID: " + templateId);
}
messagesProcessed++;
+
+ // JFR event for batch processing
+ long processingTime = System.nanoTime() - startTime;
+ MarketDataEvents.BatchProcessed batchEvent = new MarketDataEvents.BatchProcessed();
+ if (batchEvent.isEnabled()) {
+ batchEvent.messageCount = 1;
+ batchEvent.processingTimeNanos = processingTime;
+ batchEvent.source = "AERON";
+ batchEvent.commit();
+ }
+
if (shouldBroadcast) {
messagesBroadcast++;
}
@@ -96,43 +105,61 @@ public void onFragment(DirectBuffer buffer, int offset, int length, Header heade
}
}
- /**
- * Process Trade message using SBE decoder (zero-copy)
- */
+ /** Process Trade message using SBE decoder (zero-copy) */
private void processTrade(DirectBuffer buffer, int offset, int blockLength, int version, boolean shouldBroadcast) {
+ long decodeStart = System.nanoTime();
+
tradeDecoder.wrap(buffer, offset, blockLength, version);
- // Extract fields from SBE decoder
final long timestamp = tradeDecoder.timestamp();
final long tradeId = tradeDecoder.tradeId();
final long price = tradeDecoder.price();
final long quantity = tradeDecoder.quantity();
final Side side = tradeDecoder.side();
- // Extract variable-length symbol string
final int symbolLength = tradeDecoder.symbolLength();
- final byte[] symbolBytes = new byte[symbolLength];
- tradeDecoder.getSymbol(symbolBytes, 0, symbolLength);
- final String symbol = new String(symbolBytes);
-
- // Only broadcast sampled messages to avoid overwhelming browser
- if (shouldBroadcast) {
- // Convert to JSON (intentionally creating garbage for GC testing)
- String json = String.format(
- "{\"type\":\"trade\",\"timestamp\":%d,\"tradeId\":%d,\"symbol\":\"%s\"," +
- "\"price\":%.4f,\"quantity\":%d,\"side\":\"%s\"}",
- timestamp, tradeId, symbol,
- price / 10000.0, quantity, side
- );
-
- broadcaster.broadcast(json);
+ tradeDecoder.getSymbol(symbolBuffer, 0, symbolLength);
+ final String symbol = new String(symbolBuffer, 0, symbolLength);
+
+ // JFR event for SBE decoding performance
+ long decodeTime = System.nanoTime() - decodeStart;
+ MarketDataEvents.SbeDecode decodeEvent = new MarketDataEvents.SbeDecode();
+ if (decodeEvent.isEnabled()) {
+ decodeEvent.messageType = "Trade";
+ decodeEvent.decodedBytes = blockLength;
+ decodeEvent.decodeTimeNanos = decodeTime;
+ decodeEvent.commit();
+ }
+
+ if (!shouldBroadcast) {
+ return;
}
+
+ sb.setLength(0);
+ sb.append("{\"type\":\"trade\",\"timestamp\":")
+ .append(timestamp)
+ .append(",\"tradeId\":")
+ .append(tradeId)
+ .append(",\"symbol\":\"")
+ .append(symbol)
+ .append("\"")
+ .append(",\"price\":")
+ .append(price / 10000.0)
+ .append(",\"quantity\":")
+ .append(quantity)
+ .append(",\"side\":\"")
+ .append(side)
+ .append("\"}");
+
+ broadcaster.broadcast(sb.toString());
}
- /**
- * Process Quote message using SBE decoder (zero-copy)
- */
+ /** Process Quote message using SBE decoder (zero-copy) */
private void processQuote(DirectBuffer buffer, int offset, int blockLength, int version, boolean shouldBroadcast) {
+ if (!shouldBroadcast) {
+ return;
+ }
+
quoteDecoder.wrap(buffer, offset, blockLength, version);
final long timestamp = quoteDecoder.timestamp();
@@ -142,84 +169,79 @@ private void processQuote(DirectBuffer buffer, int offset, int blockLength, int
final long askSize = quoteDecoder.askSize();
final int symbolLength = quoteDecoder.symbolLength();
- final byte[] symbolBytes = new byte[symbolLength];
- quoteDecoder.getSymbol(symbolBytes, 0, symbolLength);
- final String symbol = new String(symbolBytes);
-
- // Only broadcast sampled messages
- if (shouldBroadcast) {
- String json = String.format(
- "{\"type\":\"quote\",\"timestamp\":%d,\"symbol\":\"%s\"," +
- "\"bid\":{\"price\":%.4f,\"size\":%d},\"ask\":{\"price\":%.4f,\"size\":%d}}",
- timestamp, symbol,
- bidPrice / 10000.0, bidSize,
- askPrice / 10000.0, askSize
- );
-
- broadcaster.broadcast(json);
- }
+ quoteDecoder.getSymbol(symbolBuffer, 0, symbolLength);
+ final String symbol = new String(symbolBuffer, 0, symbolLength);
+
+ sb.setLength(0);
+ sb.append("{\"type\":\"quote\",\"timestamp\":")
+ .append(timestamp)
+ .append(",\"symbol\":\"")
+ .append(symbol)
+ .append("\"")
+ .append(",\"bid\":{\"price\":")
+ .append(bidPrice / 10000.0)
+ .append(",\"size\":")
+ .append(bidSize)
+ .append("}")
+ .append(",\"ask\":{\"price\":")
+ .append(askPrice / 10000.0)
+ .append(",\"size\":")
+ .append(askSize)
+ .append("}}");
+
+ broadcaster.broadcast(sb.toString());
}
- /**
- * Process MarketDepth message with repeating groups (zero-copy)
- */
+ /** Process MarketDepth message with repeating groups (zero-copy) */
private void processMarketDepth(DirectBuffer buffer, int offset, int blockLength, int version, boolean shouldBroadcast) {
+ if (!shouldBroadcast) {
+ return;
+ }
+
marketDepthDecoder.wrap(buffer, offset, blockLength, version);
final long timestamp = marketDepthDecoder.timestamp();
final long sequenceNumber = marketDepthDecoder.sequenceNumber();
- // Only broadcast sampled messages (but always decode for GC stress)
- if (shouldBroadcast) {
- // Build JSON for bids
- StringBuilder bidsJson = new StringBuilder("[");
- MarketDepthDecoder.BidsDecoder bids = marketDepthDecoder.bids();
- int bidCount = 0;
- while (bids.hasNext()) {
- bids.next();
- if (bidCount > 0) bidsJson.append(",");
- bidsJson.append(String.format(
- "{\"price\":%.4f,\"quantity\":%d}",
- bids.price() / 10000.0, bids.quantity()
- ));
- bidCount++;
- }
- bidsJson.append("]");
-
- // Build JSON for asks
- StringBuilder asksJson = new StringBuilder("[");
- MarketDepthDecoder.AsksDecoder asks = marketDepthDecoder.asks();
- int askCount = 0;
- while (asks.hasNext()) {
- asks.next();
- if (askCount > 0) asksJson.append(",");
- asksJson.append(String.format(
- "{\"price\":%.4f,\"quantity\":%d}",
- asks.price() / 10000.0, asks.quantity()
- ));
- askCount++;
- }
- asksJson.append("]");
+ sb.setLength(0);
+ sb.append("{\"type\":\"depth\",\"timestamp\":").append(timestamp).append(",\"sequence\":").append(sequenceNumber).append(",\"bids\":[");
+
+ MarketDepthDecoder.BidsDecoder bids = marketDepthDecoder.bids();
+ int bidCount = 0;
+ while (bids.hasNext()) {
+ bids.next();
+ if (bidCount > 0)
+ sb.append(",");
+ sb.append("{\"price\":").append(bids.price() / 10000.0).append(",\"quantity\":").append(bids.quantity()).append("}");
+ bidCount++;
+ }
+ sb.append("],\"asks\":[");
+
+ MarketDepthDecoder.AsksDecoder asks = marketDepthDecoder.asks();
+ int askCount = 0;
+ while (asks.hasNext()) {
+ asks.next();
+ if (askCount > 0)
+ sb.append(",");
+ sb.append("{\"price\":").append(asks.price() / 10000.0).append(",\"quantity\":").append(asks.quantity()).append("}");
+ askCount++;
+ }
+ sb.append("]");
- final int symbolLength = marketDepthDecoder.symbolLength();
- final byte[] symbolBytes = new byte[symbolLength];
- marketDepthDecoder.getSymbol(symbolBytes, 0, symbolLength);
- final String symbol = new String(symbolBytes);
+ final int symbolLength = marketDepthDecoder.symbolLength();
+ marketDepthDecoder.getSymbol(symbolBuffer, 0, symbolLength);
+ final String symbol = new String(symbolBuffer, 0, symbolLength);
- String json = String.format(
- "{\"type\":\"depth\",\"timestamp\":%d,\"symbol\":\"%s\",\"sequence\":%d," +
- "\"bids\":%s,\"asks\":%s}",
- timestamp, symbol, sequenceNumber, bidsJson, asksJson
- );
+ sb.append(",\"symbol\":\"").append(symbol).append("\"}");
- broadcaster.broadcast(json);
- }
+ broadcaster.broadcast(sb.toString());
}
- /**
- * Process OrderAck message
- */
private void processOrderAck(DirectBuffer buffer, int offset, int blockLength, int version, boolean shouldBroadcast) {
+ if (!shouldBroadcast) {
+ return;
+ }
+
orderAckDecoder.wrap(buffer, offset, blockLength, version);
final long timestamp = orderAckDecoder.timestamp();
@@ -234,54 +256,58 @@ private void processOrderAck(DirectBuffer buffer, int offset, int blockLength, i
final long cumQty = orderAckDecoder.cumQty();
final int symbolLength = orderAckDecoder.symbolLength();
- final byte[] symbolBytes = new byte[symbolLength];
- orderAckDecoder.getSymbol(symbolBytes, 0, symbolLength);
- final String symbol = new String(symbolBytes);
-
- // Only broadcast sampled messages
- if (shouldBroadcast) {
- String json = String.format(
- "{\"type\":\"orderAck\",\"timestamp\":%d,\"orderId\":%d,\"clientOrderId\":%d," +
- "\"symbol\":\"%s\",\"side\":\"%s\",\"orderType\":\"%s\",\"price\":%.4f," +
- "\"quantity\":%d,\"execType\":\"%s\",\"leavesQty\":%d,\"cumQty\":%d}",
- timestamp, orderId, clientOrderId, symbol, side, orderType,
- price / 10000.0, quantity, execType, leavesQty, cumQty
- );
-
- broadcaster.broadcast(json);
- }
+ orderAckDecoder.getSymbol(symbolBuffer, 0, symbolLength);
+ final String symbol = new String(symbolBuffer, 0, symbolLength);
+
+ sb.setLength(0);
+ sb.append("{\"type\":\"orderAck\",\"timestamp\":")
+ .append(timestamp)
+ .append(",\"orderId\":")
+ .append(orderId)
+ .append(",\"clientOrderId\":")
+ .append(clientOrderId)
+ .append(",\"symbol\":\"")
+ .append(symbol)
+ .append("\"")
+ .append(",\"side\":\"")
+ .append(side)
+ .append("\"")
+ .append(",\"orderType\":\"")
+ .append(orderType)
+ .append("\"")
+ .append(",\"price\":")
+ .append(price / 10000.0)
+ .append(",\"quantity\":")
+ .append(quantity)
+ .append(",\"execType\":\"")
+ .append(execType)
+ .append("\"")
+ .append(",\"leavesQty\":")
+ .append(leavesQty)
+ .append(",\"cumQty\":")
+ .append(cumQty)
+ .append("}");
+
+ broadcaster.broadcast(sb.toString());
}
- /**
- * Process Heartbeat message
- */
private void processHeartbeat(DirectBuffer buffer, int offset, int blockLength, int version) {
heartbeatDecoder.wrap(buffer, offset, blockLength, version);
final long timestamp = heartbeatDecoder.timestamp();
final long sequenceNumber = heartbeatDecoder.sequenceNumber();
- // Log heartbeats but don't broadcast them
if (LOGGER.isLoggable(Level.FINE)) {
LOGGER.fine("Heartbeat: timestamp=" + timestamp + ", seq=" + sequenceNumber);
}
}
- /**
- * Log statistics periodically
- */
private void logStatistics() {
long now = System.currentTimeMillis();
- if (now - lastLogTime > 5000) { // Log every 5 seconds
+ if (now - lastLogTime > 5000) {
double elapsedSeconds = (now - lastLogTime) / 1000.0;
- LOGGER.info(String.format(
- "FragmentHandler Stats - Processed: %,d (%.0f msg/sec) | Broadcast to UI: %,d (%.0f msg/sec) | Sample rate: 1 in %d",
- messagesProcessed,
- messagesProcessed / elapsedSeconds,
- messagesBroadcast,
- messagesBroadcast / elapsedSeconds,
- SAMPLE_RATE
- ));
+ LOGGER.info(String.format("FragmentHandler Stats - Processed: %,d (%.0f msg/sec) | Broadcast to UI: %,d (%.0f msg/sec) | Sample rate: 1 in %d",
+ messagesProcessed, messagesProcessed / elapsedSeconds, messagesBroadcast, messagesBroadcast / elapsedSeconds, SAMPLE_RATE));
lastLogTime = now;
messagesProcessed = 0;
messagesBroadcast = 0;
diff --git a/src/main/java/fish/payara/trader/aeron/MarketDataPublisher.java b/src/main/java/fish/payara/trader/aeron/MarketDataPublisher.java
index 1cf1345..6047ce4 100644
--- a/src/main/java/fish/payara/trader/aeron/MarketDataPublisher.java
+++ b/src/main/java/fish/payara/trader/aeron/MarketDataPublisher.java
@@ -1,38 +1,35 @@
package fish.payara.trader.aeron;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.HazelcastInstanceNotActiveException;
+import com.hazelcast.cp.IAtomicLong;
import fish.payara.trader.concurrency.VirtualThreadExecutor;
+import fish.payara.trader.jfr.MarketDataEvents;
import fish.payara.trader.sbe.*;
import fish.payara.trader.websocket.MarketDataBroadcaster;
import io.aeron.Aeron;
import io.aeron.Publication;
-import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
-import jakarta.ejb.DependsOn;
-import jakarta.ejb.Singleton;
-import jakarta.ejb.Startup;
-import jakarta.enterprise.concurrent.ManagedExecutorDefinition;
import jakarta.enterprise.concurrent.ManagedExecutorService;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.context.Initialized;
import jakarta.enterprise.event.Observes;
import jakarta.inject.Inject;
-import org.agrona.concurrent.UnsafeBuffer;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
-
import java.nio.ByteBuffer;
-import java.util.Random;
import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.agrona.concurrent.UnsafeBuffer;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
/**
- * Market Data Publisher Simulator
- * Generates synthetic market data and publishes via Aeron using SBE encoding.
- * This simulates a high-frequency data feed for testing the ingestion pipeline.
- * IMPORTANT: Depends on AeronSubscriberBean to initialize first (provides MediaDriver)
+ * Market Data Publisher Simulator Generates synthetic market data and publishes via Aeron using SBE encoding. This simulates a high-frequency data feed for
+ * testing the ingestion pipeline. IMPORTANT: Depends on AeronSubscriberBean to initialize first (provides MediaDriver)
*/
@ApplicationScoped
public class MarketDataPublisher {
@@ -42,32 +39,33 @@ public class MarketDataPublisher {
private static final String CHANNEL = "aeron:ipc";
private static final int STREAM_ID = 1001;
private static final int BUFFER_SIZE = 4096;
- private static final int SAMPLE_RATE = 50; // Broadcast 1 in 50 messages to prevent flooding
+ private static final int SAMPLE_RATE = 50;
private static final String[] SYMBOLS = {"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "NVDA", "META", "NFLX"};
private Aeron aeron;
private Publication publication;
- // SBE encoders (reusable flyweights)
private final MessageHeaderEncoder headerEncoder = new MessageHeaderEncoder();
private final TradeEncoder tradeEncoder = new TradeEncoder();
private final QuoteEncoder quoteEncoder = new QuoteEncoder();
private final MarketDepthEncoder marketDepthEncoder = new MarketDepthEncoder();
private final HeartbeatEncoder heartbeatEncoder = new HeartbeatEncoder();
- // Buffer for encoding messages
- private final UnsafeBuffer buffer = new UnsafeBuffer(ByteBuffer.allocateDirect(BUFFER_SIZE));
+ private UnsafeBuffer buffer;
- private final Random random = new Random();
private final AtomicLong sequenceNumber = new AtomicLong(0);
private final AtomicLong tradeIdGenerator = new AtomicLong(1000);
private final AtomicLong messagesPublished = new AtomicLong(0);
+ private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
+ private static final int MAX_CONSECUTIVE_FAILURES = 50;
private long sampleCounter = 0;
- private volatile boolean initialized = false;
private volatile boolean running = false;
-
+ private boolean isDirectMode;
+ private long lastWarningLogTime = 0;
+ private static final long WARNING_LOG_INTERVAL_MS = 5000;
+
private Future> publisherFuture;
private Future> statsFuture;
@@ -78,49 +76,73 @@ public class MarketDataPublisher {
@ConfigProperty(name = "TRADER_INGESTION_MODE", defaultValue = "AERON")
private String ingestionMode;
+ @Inject
+ @ConfigProperty(name = "ENABLE_PUBLISHER", defaultValue = "true")
+ String enablePublisherEnv;
+
@Inject
private MarketDataBroadcaster broadcaster;
-
+
+ @Inject
+ private HazelcastInstance hazelcastInstance;
+
@Inject
@VirtualThreadExecutor
private ManagedExecutorService managedExecutorService;
+ private IAtomicLong clusterMessageCounter;
+
void contextInitialized(@Observes @Initialized(ApplicationScoped.class) Object event) {
init();
}
public void init() {
- LOGGER.info("Initializing Market Data Publisher. Mode: " + ingestionMode);
+ if (hazelcastInstance != null) {
+ try {
+ clusterMessageCounter = hazelcastInstance.getCPSubsystem().getAtomicLong("cluster-message-count");
+ LOGGER.info("Initialized cluster-wide message counter");
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failed to initialize cluster message counter, using local counter only", e);
+ }
+ }
+
+ if (enablePublisherEnv != null && !"true".equalsIgnoreCase(enablePublisherEnv)) {
+ LOGGER.info("Market Data Publisher DISABLED on this instance (ENABLE_PUBLISHER=" + enablePublisherEnv + ")");
+ LOGGER.info("This instance will only consume messages from the cluster topic via Hazelcast");
+ return;
+ }
+
+ LOGGER.info("Initializing Market Data Publisher (ENABLE_PUBLISHER=" + enablePublisherEnv + "). Mode: " + ingestionMode);
if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
LOGGER.info("Running in DIRECT mode - Bypassing Aeron/SBE setup.");
- initialized = true;
+ isDirectMode = true;
+
startPublishing();
return;
}
+ this.buffer = new UnsafeBuffer(ByteBuffer.allocateDirect(BUFFER_SIZE));
+
try {
- // Wait for AeronSubscriberBean to be ready (both observers fire at roughly same time)
LOGGER.info("Waiting for AeronSubscriberBean to be ready...");
int waitAttempts = 0;
while (!subscriberBean.isReady() && waitAttempts < 60) {
Thread.sleep(500);
waitAttempts++;
+ LOGGER.info("Waiting for AeronSubscriberBean to be ready... " + waitAttempts + " wait count");
}
if (!subscriberBean.isReady()) {
- LOGGER.severe("AeronSubscriberBean did not become ready in time");
+ LOGGER.severe("AeronSubscriberBean did not become ready in time after " + waitAttempts + " wait count");
return;
}
String aeronDir = subscriberBean.getAeronDirectoryName();
LOGGER.info("Connecting to embedded MediaDriver at: " + aeronDir);
- aeron = Aeron.connect(new Aeron.Context()
- .aeronDirectoryName(aeronDir)
- .errorHandler(throwable ->
- LOGGER.log(Level.SEVERE, "Aeron publisher error", throwable))
- );
+ aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(aeronDir)
+ .errorHandler(throwable -> LOGGER.log(Level.SEVERE, "Aeron publisher error", throwable)));
LOGGER.info("Connected to Aeron. Adding publication...");
@@ -135,7 +157,7 @@ public void init() {
if (publication.isConnected()) {
LOGGER.info("Market Data Publisher initialized successfully");
- initialized = true;
+
startPublishing();
} else {
LOGGER.warning("Publisher not connected after waiting");
@@ -147,19 +169,42 @@ public void init() {
}
/**
- * Start background thread to continuously publish market data at high throughput
+ * Start background thread to continuously publish market data at high throughput.
+ *
+ *
+ * Burst Pattern: Each burst publishes 1,500 messages (500 iterations of Trade + Quote + MarketDepth), followed by a 5μs park. This yields a
+ * theoretical upper limit of ~300M messages/sec, but actual throughput is limited by Aeron backpressure and SBE encoding overhead.
+ *
+ *
+ * Message Counting: In AERON mode, messages are counted AFTER successful publication via {@code offer()} (i.e., "delivered" count). In DIRECT mode,
+ * messages are counted immediately upon generation (i.e., "attempted" count).
+ *
+ *
+ * Burst Multiplier: Time-based multiplier simulates market events: 1x normal, 5x during news events (seconds 20-25 of each minute), 3x during market
+ * close (seconds 45-50).
*/
private void startPublishing() {
running = true;
publisherFuture = managedExecutorService.submit(() -> {
- LOGGER.info("Market data publisher task started - targeting 50k-100k messages/sec");
+ Thread.currentThread().setName("market-data-publisher");
+ LOGGER.info("Market data publisher task started - burst pattern: 1,500 messages per 5μs (rate limited by Aeron backpressure)");
- final int BURST_SIZE = 500;
- final long PARK_NANOS = 5_000; // 5 microseconds = ~100k messages/sec
+ final int BASE_BURST_SIZE = 500;
+ final long PARK_NANOS = 5_000;
while (running && !Thread.currentThread().isInterrupted()) {
try {
- for (int i = 0; i < BURST_SIZE && running; i++) {
+ int burstMultiplier = getBurstMultiplier();
+ int adjustedBurstSize = BASE_BURST_SIZE * burstMultiplier;
+
+ if (burstMultiplier > 1) {
+ long secondOfMinute = (System.currentTimeMillis() / 1000) % 60;
+ if (secondOfMinute == 20 || secondOfMinute == 45) {
+ LOGGER.info("BURST MODE: " + burstMultiplier + "x allocation spike started");
+ }
+ }
+
+ for (int i = 0; i < adjustedBurstSize && running; i++) {
publishTrade();
publishQuote();
publishMarketDepth();
@@ -171,8 +216,8 @@ private void startPublishing() {
LockSupport.parkNanos(PARK_NANOS);
- } catch (Exception e) {
- LOGGER.log(Level.WARNING, "Error in publisher loop", e);
+ } catch (Throwable e) {
+ LOGGER.log(Level.SEVERE, "Critical error in publisher loop", e);
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
}
}
@@ -181,6 +226,7 @@ private void startPublishing() {
});
statsFuture = managedExecutorService.submit(() -> {
+ Thread.currentThread().setName("market-data-stats");
long lastCount = 0;
long lastTime = System.currentTimeMillis();
@@ -194,12 +240,15 @@ private void startPublishing() {
double elapsedSeconds = (currentTime - lastTime) / 1000.0;
double messagesPerSecond = messagesSinceLastLog / elapsedSeconds;
- LOGGER.info(String.format(
- "Publisher Stats - Total: %,d | Last 5s: %,d (%.0f msg/sec)",
- currentCount,
- messagesSinceLastLog,
- messagesPerSecond
- ));
+ int currentMultiplier = getBurstMultiplier();
+ String burstStatus = currentMultiplier > 1 ? " [BURST: " + currentMultiplier + "x]" : "";
+
+ LOGGER.info(String.format("Publisher Stats - Total: %,d | Last 5s: %,d (%.0f msg/sec)%s", currentCount, messagesSinceLastLog,
+ messagesPerSecond, burstStatus));
+
+ String statsJson = String.format("{\"type\":\"stats\",\"total\":%d,\"rate\":%.0f,\"burstMultiplier\":%d}", currentCount, messagesPerSecond,
+ currentMultiplier);
+ broadcaster.broadcast(statsJson);
lastCount = currentCount;
lastTime = currentTime;
@@ -212,168 +261,265 @@ private void startPublishing() {
}
/**
- * Publish a Trade message
+ * Calculate burst multiplier based on time pattern simulating market events
+ *
+ *
+ * Pattern (60-second cycle): - 00-20s: Normal trading (1x) - 20-25s: News event burst (5x allocation spike) - 25-45s: Normal trading (1x) - 45-50s: Market
+ * close spike (3x allocation) - 50-60s: Normal trading (1x)
*/
+ private int getBurstMultiplier() {
+ long secondOfMinute = (System.currentTimeMillis() / 1000) % 60;
+
+ if (secondOfMinute >= 20 && secondOfMinute < 25) {
+ MarketDataEvents.BurstModeActivated event = new MarketDataEvents.BurstModeActivated();
+ if (event.isEnabled()) {
+ event.multiplier = 5;
+ event.reason = "News Event";
+ event.secondOfMinute = secondOfMinute;
+ event.commit();
+ }
+ return 5;
+ } else if (secondOfMinute >= 45 && secondOfMinute < 50) {
+ MarketDataEvents.BurstModeActivated event = new MarketDataEvents.BurstModeActivated();
+ if (event.isEnabled()) {
+ event.multiplier = 3;
+ event.reason = "Market Close";
+ event.secondOfMinute = secondOfMinute;
+ event.commit();
+ }
+ return 3;
+ }
+
+ return 1;
+ }
+
+ /** Publish a Trade message */
private void publishTrade() {
- if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
- final double price = 100.0 + random.nextDouble() * 400.0;
- final int quantity = random.nextInt(1000) + 100;
- final String side = random.nextBoolean() ? "BUY" : "SELL";
-
- String json = String.format(
- "{\"type\":\"trade\",\"timestamp\":%d,\"tradeId\":%d,\"symbol\":\"%s\",\"price\":%.4f,\"quantity\":%d,\"side\":\"%s\"}",
- System.currentTimeMillis(),
- tradeIdGenerator.incrementAndGet(),
- symbol,
- price,
- quantity,
- side
- );
+ final ThreadLocalRandom currentRandom = ThreadLocalRandom.current();
+ if (isDirectMode) {
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
+ final double price = 100.0 + currentRandom.nextDouble() * 400.0;
+ final int quantity = currentRandom.nextInt(1000) + 100;
+ final String side = currentRandom.nextBoolean() ? "BUY" : "SELL";
+
+ String json = String.format("{\"type\":\"trade\",\"timestamp\":%d,\"tradeId\":%d,\"symbol\":\"%s\",\"price\":%.4f,\"quantity\":%d,\"side\":\"%s\"}",
+ System.currentTimeMillis(), tradeIdGenerator.incrementAndGet(), symbol, price, quantity, side);
+
+ MarketDataEvents.TradePublished tradeEvent = new MarketDataEvents.TradePublished();
+ if (tradeEvent.isEnabled()) {
+ tradeEvent.symbol = symbol;
+ tradeEvent.price = (long) (price * 10000);
+ tradeEvent.quantity = quantity;
+ tradeEvent.side = side;
+ tradeEvent.commit();
+ }
+
if (++sampleCounter % SAMPLE_RATE == 0) {
broadcaster.broadcastWithArtificialLoad(json);
}
messagesPublished.incrementAndGet();
+ if (clusterMessageCounter != null) {
+ try {
+ clusterMessageCounter.incrementAndGet();
+ } catch (Exception e) {
+ }
+ }
return;
}
+ long encodeStart = System.nanoTime();
int bufferOffset = 0;
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
headerEncoder.wrap(buffer, bufferOffset)
- .blockLength(tradeEncoder.sbeBlockLength())
- .templateId(tradeEncoder.sbeTemplateId())
- .schemaId(tradeEncoder.sbeSchemaId())
- .version(tradeEncoder.sbeSchemaVersion());
+ .blockLength(tradeEncoder.sbeBlockLength())
+ .templateId(tradeEncoder.sbeTemplateId())
+ .schemaId(tradeEncoder.sbeSchemaId())
+ .version(tradeEncoder.sbeSchemaVersion());
bufferOffset += headerEncoder.encodedLength();
+ final long encodedPrice = (long) ((100.0 + currentRandom.nextDouble() * 400.0) * 10000);
+ final int encodedQuantity = currentRandom.nextInt(1000) + 100;
+ final Side encodedSide = currentRandom.nextBoolean() ? Side.BUY : Side.SELL;
+
tradeEncoder.wrap(buffer, bufferOffset)
- .timestamp(System.currentTimeMillis())
- .tradeId(tradeIdGenerator.incrementAndGet())
- .price((long) ((100.0 + random.nextDouble() * 400.0) * 10000)) // $100-$500
- .quantity(random.nextInt(1000) + 100)
- .side(random.nextBoolean() ? Side.BUY : Side.SELL)
- .symbol(symbol);
+ .timestamp(System.currentTimeMillis())
+ .tradeId(tradeIdGenerator.incrementAndGet())
+ .price(encodedPrice)
+ .quantity(encodedQuantity)
+ .side(encodedSide)
+ .symbol(symbol);
final int length = headerEncoder.encodedLength() + tradeEncoder.encodedLength();
- offer(buffer, 0, length, "Trade");
+
+ MarketDataEvents.SbeEncode encodeEvent = new MarketDataEvents.SbeEncode();
+ if (encodeEvent.isEnabled()) {
+ encodeEvent.messageType = "Trade";
+ encodeEvent.encodedBytes = length;
+ encodeEvent.encodeTimeNanos = System.nanoTime() - encodeStart;
+ encodeEvent.commit();
+ }
+
+ offer(buffer, 0, length, "Trade", symbol, encodedPrice, encodedQuantity, encodedSide.name());
}
- /**
- * Publish a Quote message
- */
+ /** Publish a Quote message */
private void publishQuote() {
- if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
- final double basePrice = 100.0 + random.nextDouble() * 400.0;
+ final ThreadLocalRandom currentRandom = ThreadLocalRandom.current();
+ if (isDirectMode) {
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
+ final double basePrice = 100.0 + currentRandom.nextDouble() * 400.0;
final double bidPrice = basePrice - 0.01;
final double askPrice = basePrice + 0.01;
- final int bidSize = random.nextInt(10000) + 100;
- final int askSize = random.nextInt(10000) + 100;
+ final int bidSize = currentRandom.nextInt(10000) + 100;
+ final int askSize = currentRandom.nextInt(10000) + 100;
+
+ MarketDataEvents.QuotePublished quoteEvent = new MarketDataEvents.QuotePublished();
+ if (quoteEvent.isEnabled()) {
+ quoteEvent.symbol = symbol;
+ quoteEvent.bidPrice = (long) (bidPrice * 10000);
+ quoteEvent.askPrice = (long) (askPrice * 10000);
+ quoteEvent.bidSize = bidSize;
+ quoteEvent.askSize = askSize;
+ quoteEvent.commit();
+ }
String json = String.format(
- "{\"type\":\"quote\",\"timestamp\":%d,\"symbol\":\"%s\",\"bid\":{\"price\":%.4f,\"size\":%d},\"ask\":{\"price\":%.4f,\"size\":%d}}",
- System.currentTimeMillis(), symbol,
- bidPrice, bidSize,
- askPrice, askSize
- );
+ "{\"type\":\"quote\",\"timestamp\":%d,\"symbol\":\"%s\",\"bid\":{\"price\":%.4f,\"size\":%d},\"ask\":{\"price\":%.4f,\"size\":%d}}",
+ System.currentTimeMillis(), symbol, bidPrice, bidSize, askPrice, askSize);
if (++sampleCounter % SAMPLE_RATE == 0) {
broadcaster.broadcastWithArtificialLoad(json);
}
messagesPublished.incrementAndGet();
+ if (clusterMessageCounter != null) {
+ try {
+ clusterMessageCounter.incrementAndGet();
+ } catch (Exception e) {
+ }
+ }
return;
}
+ long encodeStart = System.nanoTime();
int bufferOffset = 0;
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
- final double basePrice = 100.0 + random.nextDouble() * 400.0;
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
+ final double basePrice = 100.0 + currentRandom.nextDouble() * 400.0;
headerEncoder.wrap(buffer, bufferOffset)
- .blockLength(quoteEncoder.sbeBlockLength())
- .templateId(quoteEncoder.sbeTemplateId())
- .schemaId(quoteEncoder.sbeSchemaId())
- .version(quoteEncoder.sbeSchemaVersion());
+ .blockLength(quoteEncoder.sbeBlockLength())
+ .templateId(quoteEncoder.sbeTemplateId())
+ .schemaId(quoteEncoder.sbeSchemaId())
+ .version(quoteEncoder.sbeSchemaVersion());
bufferOffset += headerEncoder.encodedLength();
+ final long bidPrice = (long) ((basePrice - 0.01) * 10000);
+ final long askPrice = (long) ((basePrice + 0.01) * 10000);
+ final int bidSize = currentRandom.nextInt(10000) + 100;
+ final int askSize = currentRandom.nextInt(10000) + 100;
+
quoteEncoder.wrap(buffer, bufferOffset)
- .timestamp(System.currentTimeMillis())
- .bidPrice((long) ((basePrice - 0.01) * 10000))
- .bidSize(random.nextInt(10000) + 100)
- .askPrice((long) ((basePrice + 0.01) * 10000))
- .askSize(random.nextInt(10000) + 100)
- .symbol(symbol);
+ .timestamp(System.currentTimeMillis())
+ .bidPrice(bidPrice)
+ .bidSize(bidSize)
+ .askPrice(askPrice)
+ .askSize(askSize)
+ .symbol(symbol);
final int length = headerEncoder.encodedLength() + quoteEncoder.encodedLength();
+
+ MarketDataEvents.SbeEncode encodeEvent = new MarketDataEvents.SbeEncode();
+ if (encodeEvent.isEnabled()) {
+ encodeEvent.messageType = "Quote";
+ encodeEvent.encodedBytes = length;
+ encodeEvent.encodeTimeNanos = System.nanoTime() - encodeStart;
+ encodeEvent.commit();
+ }
+
offer(buffer, 0, length, "Quote");
+
+ MarketDataEvents.QuotePublished quoteEvent = new MarketDataEvents.QuotePublished();
+ if (quoteEvent.isEnabled()) {
+ quoteEvent.symbol = symbol;
+ quoteEvent.bidPrice = bidPrice;
+ quoteEvent.askPrice = askPrice;
+ quoteEvent.bidSize = bidSize;
+ quoteEvent.askSize = askSize;
+ quoteEvent.commit();
+ }
}
- /**
- * Publish a MarketDepth message
- */
+ /** Publish a MarketDepth message */
private void publishMarketDepth() {
- if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
- final double basePrice = 100.0 + random.nextDouble() * 400.0;
+ final ThreadLocalRandom currentRandom = ThreadLocalRandom.current();
+ if (isDirectMode) {
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
+ final double basePrice = 100.0 + currentRandom.nextDouble() * 400.0;
final long timestamp = System.currentTimeMillis();
final long seq = sequenceNumber.incrementAndGet();
+ MarketDataEvents.MarketDepthPublished depthEvent = new MarketDataEvents.MarketDepthPublished();
+ if (depthEvent.isEnabled()) {
+ depthEvent.symbol = symbol;
+ depthEvent.depthLevels = 5;
+ depthEvent.sequenceNumber = seq;
+ depthEvent.commit();
+ }
+
StringBuilder bidsJson = new StringBuilder("[");
for (int i = 0; i < 5; i++) {
- if (i > 0) bidsJson.append(",");
- bidsJson.append(String.format("{\"price\":%.4f,\"quantity\":%d}",
- basePrice - (i + 1) * 0.01, random.nextInt(5000) + 100));
+ if (i > 0)
+ bidsJson.append(",");
+ bidsJson.append(String.format("{\"price\":%.4f,\"quantity\":%d}", basePrice - (i + 1) * 0.01, currentRandom.nextInt(5000) + 100));
}
bidsJson.append("]");
StringBuilder asksJson = new StringBuilder("[");
for (int i = 0; i < 5; i++) {
- if (i > 0) asksJson.append(",");
- asksJson.append(String.format("{\"price\":%.4f,\"quantity\":%d}",
- basePrice + (i + 1) * 0.01, random.nextInt(5000) + 100));
+ if (i > 0)
+ asksJson.append(",");
+ asksJson.append(String.format("{\"price\":%.4f,\"quantity\":%d}", basePrice + (i + 1) * 0.01, currentRandom.nextInt(5000) + 100));
}
asksJson.append("]");
- String json = String.format(
- "{\"type\":\"depth\",\"timestamp\":%d,\"symbol\":\"%s\",\"sequence\":%d,\"bids\":%s,\"asks\":%s}",
- timestamp, symbol, seq, bidsJson, asksJson
- );
+ String json = String.format("{\"type\":\"depth\",\"timestamp\":%d,\"symbol\":\"%s\",\"sequence\":%d,\"bids\":%s,\"asks\":%s}", timestamp, symbol,
+ seq, bidsJson, asksJson);
if (++sampleCounter % SAMPLE_RATE == 0) {
broadcaster.broadcastWithArtificialLoad(json);
}
messagesPublished.incrementAndGet();
+ if (clusterMessageCounter != null) {
+ try {
+ clusterMessageCounter.incrementAndGet();
+ } catch (Exception e) {
+ }
+ }
return;
}
int bufferOffset = 0;
- final String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
- final double basePrice = 100.0 + random.nextDouble() * 400.0;
+ final String symbol = SYMBOLS[currentRandom.nextInt(SYMBOLS.length)];
+ final double basePrice = 100.0 + currentRandom.nextDouble() * 400.0;
headerEncoder.wrap(buffer, bufferOffset)
- .blockLength(marketDepthEncoder.sbeBlockLength())
- .templateId(marketDepthEncoder.sbeTemplateId())
- .schemaId(marketDepthEncoder.sbeSchemaId())
- .version(marketDepthEncoder.sbeSchemaVersion());
+ .blockLength(marketDepthEncoder.sbeBlockLength())
+ .templateId(marketDepthEncoder.sbeTemplateId())
+ .schemaId(marketDepthEncoder.sbeSchemaId())
+ .version(marketDepthEncoder.sbeSchemaVersion());
bufferOffset += headerEncoder.encodedLength();
- marketDepthEncoder.wrap(buffer, bufferOffset)
- .timestamp(System.currentTimeMillis())
- .sequenceNumber(sequenceNumber.incrementAndGet());
+ marketDepthEncoder.wrap(buffer, bufferOffset).timestamp(System.currentTimeMillis()).sequenceNumber(sequenceNumber.incrementAndGet());
MarketDepthEncoder.BidsEncoder bidsEncoder = marketDepthEncoder.bidsCount(5);
for (int i = 0; i < 5; i++) {
- bidsEncoder.next()
- .price((long) ((basePrice - (i + 1) * 0.01) * 10000))
- .quantity(random.nextInt(5000) + 100);
+ bidsEncoder.next().price((long) ((basePrice - (i + 1) * 0.01) * 10000)).quantity(currentRandom.nextInt(5000) + 100);
}
MarketDepthEncoder.AsksEncoder asksEncoder = marketDepthEncoder.asksCount(5);
for (int i = 0; i < 5; i++) {
- asksEncoder.next()
- .price((long) ((basePrice + (i + 1) * 0.01) * 10000))
- .quantity(random.nextInt(5000) + 100);
+ asksEncoder.next().price((long) ((basePrice + (i + 1) * 0.01) * 10000)).quantity(currentRandom.nextInt(5000) + 100);
}
marketDepthEncoder.symbol(symbol);
@@ -382,13 +528,9 @@ private void publishMarketDepth() {
offer(buffer, 0, length, "MarketDepth");
}
- /**
- * Publish a Heartbeat message
- */
+ /** Publish a Heartbeat message */
private void publishHeartbeat() {
- if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
- // In DIRECT mode, we just increment counter but don't broadcast heartbeats to UI
- // as they are mainly for system health checks in the Aeron log
+ if (isDirectMode) {
messagesPublished.incrementAndGet();
return;
}
@@ -396,53 +538,138 @@ private void publishHeartbeat() {
int bufferOffset = 0;
headerEncoder.wrap(buffer, bufferOffset)
- .blockLength(heartbeatEncoder.sbeBlockLength())
- .templateId(heartbeatEncoder.sbeTemplateId())
- .schemaId(heartbeatEncoder.sbeSchemaId())
- .version(heartbeatEncoder.sbeSchemaVersion());
+ .blockLength(heartbeatEncoder.sbeBlockLength())
+ .templateId(heartbeatEncoder.sbeTemplateId())
+ .schemaId(heartbeatEncoder.sbeSchemaId())
+ .version(heartbeatEncoder.sbeSchemaVersion());
bufferOffset += headerEncoder.encodedLength();
- heartbeatEncoder.wrap(buffer, bufferOffset)
- .timestamp(System.currentTimeMillis())
- .sequenceNumber(sequenceNumber.get());
+ heartbeatEncoder.wrap(buffer, bufferOffset).timestamp(System.currentTimeMillis()).sequenceNumber(sequenceNumber.get());
final int length = headerEncoder.encodedLength() + heartbeatEncoder.encodedLength();
offer(buffer, 0, length, "Heartbeat");
}
- /**
- * Offer buffer to Aeron publication with retry logic
- */
+ /** Offer buffer to Aeron publication with retry logic */
private void offer(UnsafeBuffer buffer, int offset, int length, String messageType) {
long result;
- int retries = 3;
- while (retries > 0) {
+ for (int retries = 3; retries > 0; retries--) {
result = publication.offer(buffer, offset, length);
if (result > 0) {
messagesPublished.incrementAndGet();
+ if (clusterMessageCounter != null) {
+ try {
+ clusterMessageCounter.incrementAndGet();
+ } catch (Exception e) {
+ }
+ }
+ consecutiveFailures.set(0);
return;
} else if (result == Publication.BACK_PRESSURED) {
- LOGGER.fine("Back pressured on " + messageType);
- retries--;
- try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return;
+ MarketDataEvents.BackpressureEvent bpEvent = new MarketDataEvents.BackpressureEvent();
+ if (bpEvent.isEnabled()) {
+ bpEvent.messageType = messageType;
+ bpEvent.consecutiveFailures = consecutiveFailures.get() + 1;
+ bpEvent.result = "BACK_PRESSURED";
+ bpEvent.commit();
+ }
+ if (retries > 1) {
+ continue;
}
} else if (result == Publication.NOT_CONNECTED) {
- LOGGER.warning("Publication not connected");
+ logWarningRateLimited("Publication not connected");
+ emitBackpressureEvent(messageType, result);
+ handlePublishFailure(messageType);
return;
} else {
- LOGGER.warning("Offer failed for " + messageType + ": " + result);
+ logWarningRateLimited("Offer failed for " + messageType + ": " + result);
+ emitBackpressureEvent(messageType, result);
+ handlePublishFailure(messageType);
return;
}
}
- LOGGER.warning("Failed to publish " + messageType + " after retries");
+ logWarningRateLimited("Failed to publish " + messageType + " after retries");
+ handlePublishFailure(messageType);
+ }
+
+ /** Offer Trade message with JFR event emission */
+ private void offer(UnsafeBuffer buffer, int offset, int length, String messageType, String symbol, long price, int quantity, String side) {
+ long result;
+
+ for (int retries = 3; retries > 0; retries--) {
+ result = publication.offer(buffer, offset, length);
+
+ if (result > 0) {
+ messagesPublished.incrementAndGet();
+ if (clusterMessageCounter != null) {
+ try {
+ clusterMessageCounter.incrementAndGet();
+ } catch (Exception e) {
+ }
+ }
+ consecutiveFailures.set(0);
+
+ MarketDataEvents.TradePublished event = new MarketDataEvents.TradePublished();
+ if (event.isEnabled()) {
+ event.symbol = symbol;
+ event.price = price;
+ event.quantity = quantity;
+ event.side = side;
+ event.commit();
+ }
+ return;
+ } else if (result == Publication.BACK_PRESSURED) {
+ emitBackpressureEvent("Trade", result);
+ if (retries > 1) {
+ continue;
+ }
+ } else if (result == Publication.NOT_CONNECTED) {
+ logWarningRateLimited("Publication not connected");
+ emitBackpressureEvent("Trade", result);
+ handlePublishFailure("Trade");
+ return;
+ } else {
+ logWarningRateLimited("Offer failed for Trade: " + result);
+ emitBackpressureEvent("Trade", result);
+ handlePublishFailure("Trade");
+ return;
+ }
+ }
+
+ logWarningRateLimited("Failed to publish Trade after retries");
+ handlePublishFailure("Trade");
+ }
+
+ private void emitBackpressureEvent(String messageType, long resultCode) {
+ MarketDataEvents.BackpressureEvent event = new MarketDataEvents.BackpressureEvent();
+ if (event.isEnabled()) {
+ event.messageType = messageType;
+ event.consecutiveFailures = consecutiveFailures.get() + 1;
+ event.result = resultCode == Publication.BACK_PRESSURED ? "BACK_PRESSURED"
+ : resultCode == Publication.NOT_CONNECTED ? "NOT_CONNECTED" : String.valueOf(resultCode);
+ event.commit();
+ }
+ }
+
+ private void logWarningRateLimited(String message) {
+ long now = System.currentTimeMillis();
+ if (now - lastWarningLogTime >= WARNING_LOG_INTERVAL_MS) {
+ LOGGER.warning(message);
+ lastWarningLogTime = now;
+ }
+ }
+
+ private void handlePublishFailure(String messageType) {
+ int failures = consecutiveFailures.incrementAndGet();
+ if (failures >= MAX_CONSECUTIVE_FAILURES) {
+ LOGGER.severe(String.format("Circuit breaker triggered: %d consecutive publish failures (last attempted: %s). Stopping message generation.",
+ failures, messageType));
+ running = false;
+ }
}
@PreDestroy
@@ -450,15 +677,21 @@ public void shutdown() {
LOGGER.info("Shutting down Market Data Publisher...");
running = false;
-
+
if (publisherFuture != null) {
publisherFuture.cancel(true);
}
-
+
if (statsFuture != null) {
statsFuture.cancel(true);
}
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
if (publication != null) {
publication.close();
}
@@ -472,4 +705,42 @@ public void shutdown() {
public long getMessagesPublished() {
return messagesPublished.get();
}
+
+ public long getClusterMessagesPublished() {
+ if (clusterMessageCounter != null) {
+ try {
+ return clusterMessageCounter.get();
+ } catch (HazelcastInstanceNotActiveException e) {
+ LOGGER.log(Level.FINE, "Cluster message counter unavailable (Hazelcast shutting down)");
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failed to read cluster message counter", e);
+ }
+ }
+ return 0;
+ }
+
+ private volatile long lastMessagesPublishedAtRateCheck = 0;
+ private volatile long lastRateCheckTime = 0;
+
+ public long getMessageRatePerSecond() {
+ long now = System.currentTimeMillis();
+ long currentCount = messagesPublished.get();
+ long elapsedMs = now - lastRateCheckTime;
+
+ if (elapsedMs < 1000) {
+ return lastMessagesPublishedAtRateCheck;
+ }
+
+ long rate = (currentCount - lastMessagesPublishedAtRateCheck) * 1000 / elapsedMs;
+ lastMessagesPublishedAtRateCheck = currentCount;
+ lastRateCheckTime = now;
+ return Math.max(rate, 0);
+ }
+
+ /**
+ * Check if the publisher is actively running and publishing messages. Used by health check endpoints to verify system readiness.
+ */
+ public boolean isRunning() {
+ return running;
+ }
}
diff --git a/src/main/java/fish/payara/trader/analysis/BarAggregator.java b/src/main/java/fish/payara/trader/analysis/BarAggregator.java
new file mode 100644
index 0000000..cfcb014
--- /dev/null
+++ b/src/main/java/fish/payara/trader/analysis/BarAggregator.java
@@ -0,0 +1,206 @@
+package fish.payara.trader.analysis;
+
+import fish.payara.trader.concurrency.VirtualThreadExecutor;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.concurrent.ManagedExecutorService;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.ta4j.core.Bar;
+import org.ta4j.core.BarSeries;
+import org.ta4j.core.BaseBarSeriesBuilder;
+import org.ta4j.core.bars.TimeBarBuilder;
+
+/**
+ * Aggregates raw ticks (trades and quotes) into OHLCV bars for ta4j. One series per symbol, capped at maxBars. A background task closes completed bars and
+ * opens new ones on the configured interval.
+ */
+@ApplicationScoped
+public class BarAggregator {
+
+ private static final Logger LOGGER = Logger.getLogger(BarAggregator.class.getName());
+
+ private final ConcurrentHashMap series = new ConcurrentHashMap<>();
+
+ private record MutableBar(double open, double high, double low, double close, double volume, Instant endTime) {
+ }
+
+ private final ConcurrentHashMap mutableBars = new ConcurrentHashMap<>();
+
+ @Inject
+ private IndicatorConfig config;
+
+ @Inject
+ @VirtualThreadExecutor
+ private ManagedExecutorService executorService;
+
+ private ScheduledExecutorService barScheduler;
+ private Future> barTask;
+
+ @PostConstruct
+ public void init() {
+ barScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "bar-aggregator");
+ t.setDaemon(true);
+ return t;
+ });
+
+ long intervalMs = config.barDurationSeconds() * 1000L;
+ barTask = barScheduler.scheduleAtFixedRate(this::closeAllBars, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ LOGGER.info("BarAggregator initialized with %d-second bars, max %d bars".formatted(config.barDurationSeconds(), config.maxBars()));
+ }
+
+ @PreDestroy
+ public void shutdown() {
+ if (barTask != null) {
+ barTask.cancel(false);
+ }
+ if (barScheduler != null) {
+ barScheduler.shutdown();
+ }
+ }
+
+ /**
+ * Feed a trade into the current bar for the given symbol.
+ */
+ public void onTrade(String symbol, double price, long quantity, long timestamp) {
+ executorService.submit(() -> updateBar(symbol, price, quantity, timestamp));
+ }
+
+ /**
+ * Feed a quote (bid/ask) into the current bar using the mid-price.
+ */
+ public void onQuote(String symbol, double bidPrice, double askPrice, long timestamp) {
+ executorService.submit(() -> {
+ double midPrice = (bidPrice + askPrice) / 2.0;
+ updateBar(symbol, midPrice, 0, timestamp);
+ });
+ }
+
+ private void updateBar(String symbol, double price, long quantity, long timestamp) {
+ Instant endTime = Instant.ofEpochMilli(timestamp);
+
+ MutableBar current = mutableBars.compute(symbol, (key, existing) -> {
+ if (existing == null) {
+ getOrCreateSeries(symbol);
+ Instant barEnd = alignToEndOfBar(endTime, config.barDurationSeconds());
+ return new MutableBar(price, price, price, price, quantity, barEnd);
+ }
+
+ Instant barEnd = existing.endTime();
+ Instant barStart = barEnd.minus(Duration.ofSeconds(config.barDurationSeconds()));
+
+ if (!timestampIsInBar(endTime, barStart, barEnd)) {
+ closeBar(symbol);
+ Instant newBarEnd = alignToEndOfBar(endTime, config.barDurationSeconds());
+ return new MutableBar(price, price, price, price, quantity, newBarEnd);
+ }
+
+ double newHigh = Math.max(existing.high(), price);
+ double newLow = Math.min(existing.low(), price);
+ double newVolume = existing.volume() + quantity;
+ return new MutableBar(existing.open(), newHigh, newLow, price, newVolume, barEnd);
+ });
+ }
+
+ private Instant alignToEndOfBar(Instant timestamp, int barDurationSeconds) {
+ long epochSeconds = timestamp.getEpochSecond();
+ long alignedEnd = ((epochSeconds / barDurationSeconds) + 1) * barDurationSeconds;
+ return Instant.ofEpochSecond(alignedEnd);
+ }
+
+ private boolean timestampIsInBar(Instant timestamp, Instant barStart, Instant barEnd) {
+ return !timestamp.isBefore(barStart) && timestamp.isBefore(barEnd);
+ }
+
+ private BarSeries getOrCreateSeries(String symbol) {
+ return series.computeIfAbsent(symbol, k -> {
+ var srs = new BaseBarSeriesBuilder().withName(k).build();
+ srs.setMaximumBarCount(config.maxBars());
+ return srs;
+ });
+ }
+
+ private void closeBar(String symbol) {
+ MutableBar mutable = mutableBars.get(symbol);
+ if (mutable == null) {
+ return;
+ }
+
+ BarSeries srs = getOrCreateSeries(symbol);
+ try {
+ var builder = new TimeBarBuilder().timePeriod(Duration.ofSeconds(config.barDurationSeconds()))
+ .endTime(mutable.endTime())
+ .openPrice(mutable.open())
+ .highPrice(mutable.high())
+ .lowPrice(mutable.low())
+ .closePrice(mutable.close())
+ .volume(mutable.volume());
+
+ srs.addBar(builder.build());
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failed to close bar for " + symbol, e);
+ }
+ }
+
+ /**
+ * Called by the scheduled task to close all open bars and start fresh ones.
+ */
+ private void closeAllBars() {
+ for (String symbol : mutableBars.keySet()) {
+ try {
+ closeBar(symbol);
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Error closing bar for " + symbol, e);
+ }
+ }
+ mutableBars.clear();
+ }
+
+ /**
+ * Returns the BarSeries for the given symbol, or an empty series if none exists.
+ */
+ public BarSeries getSeries(String symbol) {
+ return series.getOrDefault(symbol, new BaseBarSeriesBuilder().withName(symbol).build());
+ }
+
+ /**
+ * Returns the current (in-progress) bar if one exists.
+ */
+ public Optional getCurrentBar(String symbol) {
+ MutableBar mutable = mutableBars.get(symbol);
+ if (mutable == null) {
+ return Optional.empty();
+ }
+ try {
+ var builder = new TimeBarBuilder().timePeriod(Duration.ofSeconds(config.barDurationSeconds()))
+ .endTime(mutable.endTime())
+ .openPrice(mutable.open())
+ .highPrice(mutable.high())
+ .lowPrice(mutable.low())
+ .closePrice(mutable.close())
+ .volume(mutable.volume());
+ return Optional.of(builder.build());
+ } catch (Exception e) {
+ return Optional.empty();
+ }
+ }
+
+ /**
+ * Returns the number of completed bars in the series.
+ */
+ public int getBarCount(String symbol) {
+ BarSeries srs = series.get(symbol);
+ return srs != null ? srs.getBarCount() : 0;
+ }
+}
diff --git a/src/main/java/fish/payara/trader/analysis/IndicatorConfig.java b/src/main/java/fish/payara/trader/analysis/IndicatorConfig.java
new file mode 100644
index 0000000..8930eb1
--- /dev/null
+++ b/src/main/java/fish/payara/trader/analysis/IndicatorConfig.java
@@ -0,0 +1,108 @@
+package fish.payara.trader.analysis;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+/**
+ * MicroProfile configuration for technical analysis indicator parameters. All values configurable via environment variables or microprofile-config.properties.
+ */
+@ApplicationScoped
+public class IndicatorConfig {
+
+ @Inject
+ @ConfigProperty(name = "analysis.sma.period", defaultValue = "20")
+ private int smaPeriod;
+
+ @Inject
+ @ConfigProperty(name = "analysis.ema.period", defaultValue = "12")
+ private int emaPeriod;
+
+ @Inject
+ @ConfigProperty(name = "analysis.rsi.period", defaultValue = "14")
+ private int rsiPeriod;
+
+ @Inject
+ @ConfigProperty(name = "analysis.macd.fast", defaultValue = "12")
+ private int macdFast;
+
+ @Inject
+ @ConfigProperty(name = "analysis.macd.slow", defaultValue = "26")
+ private int macdSlow;
+
+ @Inject
+ @ConfigProperty(name = "analysis.macd.signal", defaultValue = "9")
+ private int macdSignal;
+
+ @Inject
+ @ConfigProperty(name = "analysis.bb.period", defaultValue = "20")
+ private int bbPeriod;
+
+ @Inject
+ @ConfigProperty(name = "analysis.bb.stddev", defaultValue = "2.0")
+ private double bbStdDevMultiplier;
+
+ @Inject
+ @ConfigProperty(name = "analysis.atr.period", defaultValue = "14")
+ private int atrPeriod;
+
+ @Inject
+ @ConfigProperty(name = "analysis.bar.duration.seconds", defaultValue = "60")
+ private int barDurationSeconds;
+
+ @Inject
+ @ConfigProperty(name = "analysis.max.bars", defaultValue = "500")
+ private int maxBars;
+
+ @Inject
+ @ConfigProperty(name = "analysis.broadcast.interval.ms", defaultValue = "1000")
+ private long broadcastIntervalMs;
+
+ public int smaPeriod() {
+ return smaPeriod;
+ }
+
+ public int emaPeriod() {
+ return emaPeriod;
+ }
+
+ public int rsiPeriod() {
+ return rsiPeriod;
+ }
+
+ public int macdFast() {
+ return macdFast;
+ }
+
+ public int macdSlow() {
+ return macdSlow;
+ }
+
+ public int macdSignal() {
+ return macdSignal;
+ }
+
+ public int bbPeriod() {
+ return bbPeriod;
+ }
+
+ public double bbStdDevMultiplier() {
+ return bbStdDevMultiplier;
+ }
+
+ public int atrPeriod() {
+ return atrPeriod;
+ }
+
+ public int barDurationSeconds() {
+ return barDurationSeconds;
+ }
+
+ public int maxBars() {
+ return maxBars;
+ }
+
+ public long broadcastIntervalMs() {
+ return broadcastIntervalMs;
+ }
+}
diff --git a/src/main/java/fish/payara/trader/analysis/IndicatorService.java b/src/main/java/fish/payara/trader/analysis/IndicatorService.java
new file mode 100644
index 0000000..c866266
--- /dev/null
+++ b/src/main/java/fish/payara/trader/analysis/IndicatorService.java
@@ -0,0 +1,149 @@
+package fish.payara.trader.analysis;
+
+import fish.payara.trader.analysis.model.IndicatorSnapshot;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.ta4j.core.BarSeries;
+import org.ta4j.core.indicators.ATRIndicator;
+import org.ta4j.core.indicators.MACDIndicator;
+import org.ta4j.core.indicators.RSIIndicator;
+import org.ta4j.core.indicators.averages.EMAIndicator;
+import org.ta4j.core.indicators.averages.SMAIndicator;
+import org.ta4j.core.indicators.bollinger.BollingerBandsLowerIndicator;
+import org.ta4j.core.indicators.bollinger.BollingerBandsMiddleIndicator;
+import org.ta4j.core.indicators.bollinger.BollingerBandsUpperIndicator;
+import org.ta4j.core.indicators.helpers.ClosePriceIndicator;
+import org.ta4j.core.indicators.statistics.StandardDeviationIndicator;
+import org.ta4j.core.num.Num;
+
+/**
+ * Computes technical analysis indicators over aggregated bar data. All indicator families evaluated: SMA, EMA, RSI, MACD, Bollinger Bands (lower/middle/upper),
+ * and ATR.
+ */
+@ApplicationScoped
+public class IndicatorService {
+
+ private static final Logger LOGGER = Logger.getLogger(IndicatorService.class.getName());
+
+ @Inject
+ private BarAggregator barAggregator;
+
+ @Inject
+ private IndicatorConfig config;
+
+ /**
+ * Returns a snapshot of all indicators for the given symbol. Returns null if the series has insufficient bars.
+ */
+ public IndicatorSnapshot getSnapshot(String symbol) {
+ Map values = calculateIndicators(symbol);
+ if (values.isEmpty()) {
+ return null;
+ }
+
+ BarSeries series = barAggregator.getSeries(symbol);
+ double lastPrice = 0.0;
+ if (!series.isEmpty()) {
+ lastPrice = series.getLastBar().getClosePrice().doubleValue();
+ }
+
+ return new IndicatorSnapshot(symbol, System.currentTimeMillis(), lastPrice, values);
+ }
+
+ /**
+ * Walks the series backward computing indicators at each bar.
+ */
+ public List getHistoricalSnapshots(String symbol, int barCount) {
+ BarSeries series = barAggregator.getSeries(symbol);
+ if (series.isEmpty()) {
+ return List.of();
+ }
+
+ int requiredBars = requiredBarCount();
+ if (series.getEndIndex() < requiredBars) {
+ return List.of();
+ }
+
+ List snapshots = new ArrayList<>();
+ int endIndex = series.getEndIndex();
+ int startIndex = Math.max(requiredBars, endIndex - barCount + 1);
+
+ for (int i = startIndex; i <= endIndex; i++) {
+ BarSeries subSeries = series.getSubSeries(0, i);
+ Map values = calculateIndicatorsForSeries(subSeries);
+ if (values.isEmpty()) {
+ continue;
+ }
+
+ double lastPrice = subSeries.getLastBar().getClosePrice().doubleValue();
+ long timestamp = subSeries.getLastBar().getEndTime().toEpochMilli();
+
+ snapshots.add(new IndicatorSnapshot(symbol, timestamp, lastPrice, values));
+ }
+
+ return snapshots;
+ }
+
+ /**
+ * Calculates all indicator families for the given symbol.
+ */
+ public Map calculateIndicators(String symbol) {
+ BarSeries series = barAggregator.getSeries(symbol);
+ if (series.isEmpty() || series.getEndIndex() < requiredBarCount()) {
+ return Map.of();
+ }
+ return calculateIndicatorsForSeries(series);
+ }
+
+ private Map calculateIndicatorsForSeries(BarSeries series) {
+ ClosePriceIndicator closePrice = new ClosePriceIndicator(series);
+ Map values = new LinkedHashMap<>();
+ int last = series.getEndIndex();
+
+ int maxRequired = requiredBarCount();
+ if (last < maxRequired) {
+ return Map.of();
+ }
+
+ try {
+ SMAIndicator sma = new SMAIndicator(closePrice, config.smaPeriod());
+ values.put("SMA(%d)".formatted(config.smaPeriod()), sma.getValue(last).doubleValue());
+
+ EMAIndicator ema = new EMAIndicator(closePrice, config.emaPeriod());
+ values.put("EMA(%d)".formatted(config.emaPeriod()), ema.getValue(last).doubleValue());
+
+ RSIIndicator rsi = new RSIIndicator(closePrice, config.rsiPeriod());
+ values.put("RSI(%d)".formatted(config.rsiPeriod()), rsi.getValue(last).doubleValue());
+
+ MACDIndicator macd = new MACDIndicator(closePrice, config.macdFast(), config.macdSlow());
+ values.put("MACD(%d,%d)".formatted(config.macdFast(), config.macdSlow()), macd.getValue(last).doubleValue());
+
+ BollingerBandsMiddleIndicator bbMiddle = new BollingerBandsMiddleIndicator(sma);
+ StandardDeviationIndicator sd = new StandardDeviationIndicator(closePrice, config.bbPeriod());
+ Num k = series.numFactory().numOf(config.bbStdDevMultiplier());
+ BollingerBandsLowerIndicator bbLower = new BollingerBandsLowerIndicator(bbMiddle, sd, k);
+ BollingerBandsUpperIndicator bbUpper = new BollingerBandsUpperIndicator(bbMiddle, sd, k);
+
+ values.put("BB_Lower(%d)".formatted(config.bbPeriod()), bbLower.getValue(last).doubleValue());
+ values.put("BB_Middle(%d)".formatted(config.bbPeriod()), bbMiddle.getValue(last).doubleValue());
+ values.put("BB_Upper(%d)".formatted(config.bbPeriod()), bbUpper.getValue(last).doubleValue());
+
+ ATRIndicator atr = new ATRIndicator(series, config.atrPeriod());
+ values.put("ATR(%d)".formatted(config.atrPeriod()), atr.getValue(last).doubleValue());
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failed to compute indicators: " + e.getMessage(), e);
+ return Map.of();
+ }
+
+ return values;
+ }
+
+ private int requiredBarCount() {
+ return Math.max(config.macdSlow(), Math.max(config.bbPeriod(), config.atrPeriod())) + 1;
+ }
+}
diff --git a/src/main/java/fish/payara/trader/analysis/model/IndicatorSnapshot.java b/src/main/java/fish/payara/trader/analysis/model/IndicatorSnapshot.java
new file mode 100644
index 0000000..3294c0c
--- /dev/null
+++ b/src/main/java/fish/payara/trader/analysis/model/IndicatorSnapshot.java
@@ -0,0 +1,16 @@
+package fish.payara.trader.analysis.model;
+
+import java.util.Map;
+
+/**
+ * Snapshot of computed indicator values for a symbol at a point in time. Values are keyed by indicator label (e.g. "SMA(20)", "RSI(14)").
+ */
+public record IndicatorSnapshot(String symbol, long timestamp, double lastPrice, Map values) {
+
+ /**
+ * Returns the indicator value for the given name, or Double.NaN if absent.
+ */
+ public double getValue(String name) {
+ return values.getOrDefault(name, Double.NaN);
+ }
+}
diff --git a/src/main/java/fish/payara/trader/concurrency/ConcurrencyConfig.java b/src/main/java/fish/payara/trader/concurrency/ConcurrencyConfig.java
index a9079aa..caab85b 100644
--- a/src/main/java/fish/payara/trader/concurrency/ConcurrencyConfig.java
+++ b/src/main/java/fish/payara/trader/concurrency/ConcurrencyConfig.java
@@ -4,14 +4,9 @@
import jakarta.enterprise.context.ApplicationScoped;
/**
- * Configuration for Jakarta Concurrency resources.
- * Defines a ManagedExecutorService that uses Virtual Threads (Project Loom).
+ * Configuration for Jakarta Concurrency resources. Defines a ManagedExecutorService that uses Virtual Threads (Project Loom).
*/
@ApplicationScoped
-@ManagedExecutorDefinition(
- name = "java:module/concurrent/VirtualThreadExecutor",
- virtual = true,
- qualifiers = {VirtualThreadExecutor.class}
-)
+@ManagedExecutorDefinition(name = "java:module/concurrent/VirtualThreadExecutor", virtual = true, qualifiers = {VirtualThreadExecutor.class})
public class ConcurrencyConfig {
}
diff --git a/src/main/java/fish/payara/trader/concurrency/VirtualThreadExecutor.java b/src/main/java/fish/payara/trader/concurrency/VirtualThreadExecutor.java
index 10c85c6..6cdf766 100644
--- a/src/main/java/fish/payara/trader/concurrency/VirtualThreadExecutor.java
+++ b/src/main/java/fish/payara/trader/concurrency/VirtualThreadExecutor.java
@@ -6,14 +6,11 @@
import static java.lang.annotation.ElementType.TYPE;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
+import jakarta.inject.Qualifier;
import java.lang.annotation.Retention;
import java.lang.annotation.Target;
-import jakarta.inject.Qualifier;
-
-/**
- * Qualifier for Virtual Thread ManagedExecutorService
- */
+/** Qualifier for Virtual Thread ManagedExecutorService */
@Qualifier
@Retention(RUNTIME)
@Target({METHOD, FIELD, PARAMETER, TYPE})
diff --git a/src/main/java/fish/payara/trader/demo/DemoPresetConfig.java b/src/main/java/fish/payara/trader/demo/DemoPresetConfig.java
new file mode 100644
index 0000000..5fb4037
--- /dev/null
+++ b/src/main/java/fish/payara/trader/demo/DemoPresetConfig.java
@@ -0,0 +1,114 @@
+package fish.payara.trader.demo;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.snakeyaml.engine.v2.api.Load;
+import org.snakeyaml.engine.v2.api.LoadSettings;
+import org.snakeyaml.engine.v2.exceptions.YamlEngineException;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+
+/**
+ * Loads and provides access to demo preset configurations from demo-presets.yml.
+ */
+@ApplicationScoped
+public class DemoPresetConfig {
+
+ private static final Logger LOGGER = Logger.getLogger(DemoPresetConfig.class.getName());
+ private static final String CONFIG_PATH = "/demo-presets.yml";
+
+ private List presets;
+
+ @PostConstruct
+ public void init() {
+ this.presets = loadPresets();
+ LOGGER.info("Loaded " + presets.size() + " demo presets from configuration");
+ }
+
+ /**
+ * Loads demo presets from YAML configuration file.
+ */
+ private List loadPresets() {
+ try (InputStream input = getClass().getResourceAsStream(CONFIG_PATH)) {
+ if (input == null) {
+ LOGGER.warning("Demo presets configuration file not found: " + CONFIG_PATH);
+ return List.of();
+ }
+
+ LoadSettings settings = LoadSettings.builder().build();
+ Load yaml = new Load(settings);
+
+ @SuppressWarnings("unchecked")
+ Map root = (Map) yaml.loadFromInputStream(input);
+ List