Skip to content

Commit 07264bc

Browse files
pedanticdevLuqman Saeed
pedanticdev
and
Luqman Saeed
authored
Develop (#17)
* Revert HTF simulation strategy * implement alternative gc stress tests (#15) Co-authored-by: Luqman Saeed <[email protected]> * Fix C4 name identification * Use bundled eclipse formatter with spotless * copy spotless config file. Update presentation * Amend reame --------- Co-authored-by: Luqman Saeed <[email protected]>
1 parent 8a8eecf commit 07264bc

41 files changed

Lines changed: 4784 additions & 4776 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Dockerfile‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ WORKDIR /app
77
# Copy Maven wrapper and pom.xml first for better layer caching
88
COPY mvnw .
99
COPY mvnw.cmd .
10+
COPY spotless ./spotless
1011
COPY .mvn .mvn
1112
COPY pom.xml .
1213

‎Dockerfile.scale‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ COPY mvnw .
99
COPY mvnw.cmd .
1010
COPY .mvn .mvn
1111
COPY pom.xml .
12+
COPY spotless ./spotless
1213

1314
RUN ./mvnw dependency:go-offline -B
1415

‎Dockerfile.scale.standard‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ COPY mvnw .
99
COPY mvnw.cmd .
1010
COPY .mvn .mvn
1111
COPY pom.xml .
12+
COPY spotless ./spotless
1213

1314
RUN ./mvnw dependency:go-offline -B
1415

‎Dockerfile.standard‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ COPY mvnw .
99
COPY mvnw.cmd .
1010
COPY .mvn .mvn
1111
COPY pom.xml .
12+
COPY spotless ./spotless
1213

1314
RUN ./mvnw dependency:go-offline -B
1415

‎README.md‎

Lines changed: 47 additions & 36 deletions
Large diffs are not rendered by default.

‎pom.xml‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -324,10 +324,9 @@
324324
<version>3.0.0</version>
325325
<configuration>
326326
<java>
327-
<googleJavaFormat>
328-
<version>1.17.0</version>
329-
<style>GOOGLE</style>
330-
</googleJavaFormat>
327+
<eclipse>
328+
<file>${project.basedir}/spotless/java-formatter.xml</file>
329+
</eclipse>
331330
<removeUnusedImports/>
332331
<trimTrailingWhitespace/>
333332
<endWithNewline/>

‎src/main/java/fish/payara/resource/HelloWorldResource.java‎

Lines changed: 21 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -19,37 +19,28 @@
1919
@Path("hello")
2020
public class HelloWorldResource {
2121

22-
@Inject
23-
@ConfigProperty(name = "defaultName", defaultValue = "world")
24-
private String defaultName;
22+
@Inject
23+
@ConfigProperty(name = "defaultName", defaultValue = "world")
24+
private String defaultName;
2525

26-
@GET
27-
@Operation(summary = "Get a personalized greeting")
28-
@APIResponses(
29-
value = {
30-
@APIResponse(responseCode = "200", description = "Successful operation"),
31-
@APIResponse(responseCode = "400", description = "Invalid input")
32-
})
33-
@Counted(name = "helloEndpointCount", description = "Count of calls to the hello endpoint")
34-
@Timed(name = "helloEndpointTime", description = "Time taken to execute the hello endpoint")
35-
@Timeout(3000)
36-
@Retry(maxRetries = 3)
37-
@Fallback(fallbackMethod = "fallbackMethod")
38-
public Response hello(
39-
@QueryParam("name")
40-
@Parameter(
41-
name = "name",
42-
description = "Name to include in the greeting",
43-
required = false,
44-
example = "John")
45-
String name) {
46-
if ((name == null) || name.trim().isEmpty()) {
47-
name = defaultName;
26+
@GET
27+
@Operation(summary = "Get a personalized greeting")
28+
@APIResponses(value = {@APIResponse(responseCode = "200", description = "Successful operation"),
29+
@APIResponse(responseCode = "400", description = "Invalid input")})
30+
@Counted(name = "helloEndpointCount", description = "Count of calls to the hello endpoint")
31+
@Timed(name = "helloEndpointTime", description = "Time taken to execute the hello endpoint")
32+
@Timeout(3000)
33+
@Retry(maxRetries = 3)
34+
@Fallback(fallbackMethod = "fallbackMethod")
35+
public Response hello(
36+
@QueryParam("name") @Parameter(name = "name", description = "Name to include in the greeting", required = false, example = "John") String name) {
37+
if ((name == null) || name.trim().isEmpty()) {
38+
name = defaultName;
39+
}
40+
return Response.ok(name).build();
4841
}
49-
return Response.ok(name).build();
50-
}
5142

52-
public Response fallbackMethod(@QueryParam("name") String name) {
53-
return Response.ok("Fallback data").build();
54-
}
43+
public Response fallbackMethod(@QueryParam("name") String name) {
44+
return Response.ok("Fallback data").build();
45+
}
5546
}

‎src/main/java/fish/payara/trader/aeron/AeronSubscriberBean.java‎

Lines changed: 105 additions & 120 deletions
Original file line numberDiff line numberDiff line change
@@ -21,152 +21,137 @@
2121
import org.eclipse.microprofile.config.inject.ConfigProperty;
2222

2323
/**
24-
* Aeron Ingress Singleton Bean Launches an embedded MediaDriver and subscribes to market data
25-
* stream. Uses SBE decoders for zero-copy message processing. Runs in a dedicated thread to
26-
* continuously poll for messages. IMPORTANT: This must initialize BEFORE MarketDataPublisher
24+
* Aeron Ingress Singleton Bean Launches an embedded MediaDriver and subscribes to market data stream. Uses SBE decoders for zero-copy message processing. Runs
25+
* in a dedicated thread to continuously poll for messages. IMPORTANT: This must initialize BEFORE MarketDataPublisher
2726
*/
2827
@ApplicationScoped
2928
public class AeronSubscriberBean {
3029

31-
private static final Logger LOGGER = Logger.getLogger(AeronSubscriberBean.class.getName());
30+
private static final Logger LOGGER = Logger.getLogger(AeronSubscriberBean.class.getName());
3231

33-
private static final String CHANNEL = "aeron:ipc";
34-
private static final int STREAM_ID = 1001;
35-
private static final int FRAGMENT_LIMIT = 10;
32+
private static final String CHANNEL = "aeron:ipc";
33+
private static final int STREAM_ID = 1001;
34+
private static final int FRAGMENT_LIMIT = 10;
3635

37-
private MediaDriver mediaDriver;
38-
private Aeron aeron;
39-
private Subscription subscription;
40-
private volatile boolean running = false;
41-
private Future<?> pollingFuture;
36+
private MediaDriver mediaDriver;
37+
private Aeron aeron;
38+
private Subscription subscription;
39+
private volatile boolean running = false;
40+
private Future<?> pollingFuture;
4241

43-
@Inject private MarketDataFragmentHandler fragmentHandler;
42+
@Inject
43+
private MarketDataFragmentHandler fragmentHandler;
4444

45-
@Inject @VirtualThreadExecutor private ManagedExecutorService managedExecutorService;
45+
@Inject
46+
@VirtualThreadExecutor
47+
private ManagedExecutorService managedExecutorService;
4648

47-
@Inject
48-
@ConfigProperty(name = "TRADER_INGESTION_MODE", defaultValue = "AERON")
49-
private String ingestionMode;
49+
@Inject
50+
@ConfigProperty(name = "TRADER_INGESTION_MODE", defaultValue = "AERON")
51+
private String ingestionMode;
5052

51-
void contextInitialized(@Observes @Initialized(ApplicationScoped.class) Object event) {
52-
managedExecutorService.submit(this::init);
53-
}
54-
55-
public void init() {
56-
if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
57-
LOGGER.info("Running in DIRECT mode - Skipping Aeron/MediaDriver initialization.");
58-
return;
53+
void contextInitialized(@Observes @Initialized(ApplicationScoped.class) Object event) {
54+
managedExecutorService.submit(this::init);
5955
}
6056

61-
LOGGER.info("Initializing Aeron Subscriber Bean...");
62-
63-
try {
64-
LOGGER.info("Launching embedded MediaDriver...");
65-
mediaDriver =
66-
MediaDriver.launchEmbedded(
67-
new MediaDriver.Context()
68-
.threadingMode(ThreadingMode.SHARED)
69-
.dirDeleteOnStart(true)
70-
.dirDeleteOnShutdown(true));
71-
72-
LOGGER.info("MediaDriver launched at: " + mediaDriver.aeronDirectoryName());
73-
LOGGER.info("Connecting Aeron client...");
74-
aeron =
75-
Aeron.connect(
76-
new Aeron.Context()
77-
.aeronDirectoryName(mediaDriver.aeronDirectoryName())
78-
.errorHandler(this::onError)
79-
.availableImageHandler(
80-
image -> LOGGER.info("Available image: " + image.sourceIdentity()))
81-
.unavailableImageHandler(
82-
image -> LOGGER.info("Unavailable image: " + image.sourceIdentity())));
83-
LOGGER.info("Adding subscription on channel: " + CHANNEL + ", stream: " + STREAM_ID);
84-
subscription = aeron.addSubscription(CHANNEL, STREAM_ID);
85-
startPolling();
86-
LOGGER.info("Aeron Subscriber Bean initialized successfully");
87-
88-
} catch (Exception e) {
89-
LOGGER.log(Level.SEVERE, "Failed to initialize Aeron Subscriber Bean", e);
90-
cleanup();
91-
throw new RuntimeException("Failed to initialize Aeron", e);
57+
public void init() {
58+
if ("DIRECT".equalsIgnoreCase(ingestionMode)) {
59+
LOGGER.info("Running in DIRECT mode - Skipping Aeron/MediaDriver initialization.");
60+
return;
61+
}
62+
63+
LOGGER.info("Initializing Aeron Subscriber Bean...");
64+
65+
try {
66+
LOGGER.info("Launching embedded MediaDriver...");
67+
mediaDriver = MediaDriver
68+
.launchEmbedded(new MediaDriver.Context().threadingMode(ThreadingMode.SHARED).dirDeleteOnStart(true).dirDeleteOnShutdown(true));
69+
70+
LOGGER.info("MediaDriver launched at: " + mediaDriver.aeronDirectoryName());
71+
LOGGER.info("Connecting Aeron client...");
72+
aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(mediaDriver.aeronDirectoryName())
73+
.errorHandler(this::onError)
74+
.availableImageHandler(image -> LOGGER.info("Available image: " + image.sourceIdentity()))
75+
.unavailableImageHandler(image -> LOGGER.info("Unavailable image: " + image.sourceIdentity())));
76+
LOGGER.info("Adding subscription on channel: " + CHANNEL + ", stream: " + STREAM_ID);
77+
subscription = aeron.addSubscription(CHANNEL, STREAM_ID);
78+
startPolling();
79+
LOGGER.info("Aeron Subscriber Bean initialized successfully");
80+
81+
} catch (Exception e) {
82+
LOGGER.log(Level.SEVERE, "Failed to initialize Aeron Subscriber Bean", e);
83+
cleanup();
84+
throw new RuntimeException("Failed to initialize Aeron", e);
85+
}
9286
}
93-
}
94-
95-
/** Start background task to continuously poll for messages */
96-
private void startPolling() {
97-
running = true;
98-
pollingFuture =
99-
managedExecutorService.submit(
100-
() -> {
101-
LOGGER.info("Aeron polling task started");
102-
103-
final IdleStrategy idleStrategy =
104-
new BackoffIdleStrategy(
105-
100,
106-
10,
107-
TimeUnit.MICROSECONDS.toNanos(1),
108-
TimeUnit.MICROSECONDS.toNanos(100));
109-
110-
while (running && !Thread.currentThread().isInterrupted()) {
87+
88+
/** Start background task to continuously poll for messages */
89+
private void startPolling() {
90+
running = true;
91+
pollingFuture = managedExecutorService.submit(() -> {
92+
LOGGER.info("Aeron polling task started");
93+
94+
final IdleStrategy idleStrategy = new BackoffIdleStrategy(100, 10, TimeUnit.MICROSECONDS.toNanos(1), TimeUnit.MICROSECONDS.toNanos(100));
95+
96+
while (running && !Thread.currentThread().isInterrupted()) {
11197
try {
112-
final int fragmentsRead = subscription.poll(fragmentHandler, FRAGMENT_LIMIT);
98+
final int fragmentsRead = subscription.poll(fragmentHandler, FRAGMENT_LIMIT);
11399

114-
idleStrategy.idle(fragmentsRead);
100+
idleStrategy.idle(fragmentsRead);
115101

116102
} catch (Exception e) {
117-
LOGGER.log(Level.SEVERE, "Error polling subscription", e);
103+
LOGGER.log(Level.SEVERE, "Error polling subscription", e);
118104
}
119-
}
105+
}
106+
107+
LOGGER.info("Aeron polling task stopped");
108+
});
109+
}
120110

121-
LOGGER.info("Aeron polling task stopped");
122-
});
123-
}
111+
/** Error handler for Aeron */
112+
private void onError(Throwable throwable) {
113+
LOGGER.log(Level.SEVERE, "Aeron error occurred", throwable);
114+
}
124115

125-
/** Error handler for Aeron */
126-
private void onError(Throwable throwable) {
127-
LOGGER.log(Level.SEVERE, "Aeron error occurred", throwable);
128-
}
116+
@PreDestroy
117+
public void shutdown() {
118+
LOGGER.info("Shutting down Aeron Subscriber Bean...");
119+
running = false;
129120

130-
@PreDestroy
131-
public void shutdown() {
132-
LOGGER.info("Shutting down Aeron Subscriber Bean...");
133-
running = false;
121+
if (pollingFuture != null) {
122+
pollingFuture.cancel(true);
123+
}
134124

135-
if (pollingFuture != null) {
136-
pollingFuture.cancel(true);
125+
cleanup();
126+
LOGGER.info("Aeron Subscriber Bean shut down");
137127
}
138128

139-
cleanup();
140-
LOGGER.info("Aeron Subscriber Bean shut down");
141-
}
142-
143-
/** Clean up all Aeron resources */
144-
private void cleanup() {
145-
CloseHelper.quietClose(subscription);
146-
CloseHelper.quietClose(aeron);
147-
CloseHelper.quietClose(mediaDriver);
148-
}
149-
150-
/** Get subscription statistics */
151-
public String getStatus() {
152-
if (subscription != null) {
153-
return String.format(
154-
"Channel: %s, Stream: %d, Images: %d, Running: %b",
155-
subscription.channel(), subscription.streamId(), subscription.imageCount(), running);
129+
/** Clean up all Aeron resources */
130+
private void cleanup() {
131+
CloseHelper.quietClose(subscription);
132+
CloseHelper.quietClose(aeron);
133+
CloseHelper.quietClose(mediaDriver);
156134
}
157-
return "Not initialized";
158-
}
159135

160-
/** Get the Aeron directory name for connecting publishers */
161-
public String getAeronDirectoryName() {
162-
if (mediaDriver != null) {
163-
return mediaDriver.aeronDirectoryName();
136+
/** Get subscription statistics */
137+
public String getStatus() {
138+
if (subscription != null) {
139+
return String.format("Channel: %s, Stream: %d, Images: %d, Running: %b", subscription.channel(), subscription.streamId(), subscription.imageCount(),
140+
running);
141+
}
142+
return "Not initialized";
164143
}
165-
return null;
166-
}
167144

168-
/** Check if the MediaDriver is ready */
169-
public boolean isReady() {
170-
return mediaDriver != null && aeron != null && subscription != null && running;
171-
}
145+
/** Get the Aeron directory name for connecting publishers */
146+
public String getAeronDirectoryName() {
147+
if (mediaDriver != null) {
148+
return mediaDriver.aeronDirectoryName();
149+
}
150+
return null;
151+
}
152+
153+
/** Check if the MediaDriver is ready */
154+
public boolean isReady() {
155+
return mediaDriver != null && aeron != null && subscription != null && running;
156+
}
172157
}

0 commit comments

Comments
 (0)