r35157/com_r35157_nenjim-hubd-impl_ref
0
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

61: Introduce PriceSource and PriceSink architecture for AssetAZ Ticker

Browse Source
This commit is contained in:
minimons
2026-08-07 00:14:27 +02:00
parent 0db74b3296
commit ef1b5e0adc
13 changed files with 918 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/tjava/com/r35157/assetaz/core/service/ticker/PriceObservation.tjava
+3 -1
View File
@@ -8,10 +8,12 @@ import java.util.Objects;
public record PriceObservation(
@NotNull AssetPrice price,
@NotNull Instant observedAt
@NotNull Instant observedAt,
@NotNull ΩPriceSourceNameΩ sourceName
) {
public PriceObservation {
Objects.requireNonNull(price, "price");
Objects.requireNonNull(observedAt, "observedAt");
Objects.requireNonNull(sourceName, "sourceName");
}
}
src/main/tjava/com/r35157/assetaz/core/service/ticker/PriceSink.tjava
+19
View File
@@ -0,0 +1,19 @@
package com.r35157.assetaz.core.service.ticker;
import org.jetbrains.annotations.NotNull;
import java.math.BigDecimal;
import java.time.Instant;
/**
* Receives raw typed observations from price sources.
*/
@FunctionalInterface
public interface PriceSink {
void announce(
@NotNull PriceSource source,
@NotNull ΩPriceΩ price,
@NotNull Instant observedAt
);
}
src/main/tjava/com/r35157/assetaz/core/service/ticker/PriceSource.tjava
+18
View File
@@ -0,0 +1,18 @@
package com.r35157.assetaz.core.service.ticker;
import com.r35157.libs.valuetypes.basic.TradingPair;
import org.jetbrains.annotations.NotNull;
/**
* Obtains prices for one trading pair and announces them to a {@link PriceSink}.
*/
public interface PriceSource {
@NotNull TradingPair getTradingPair();
@NotNull ΩPriceSourceNameΩ getSourceName();
void start(@NotNull PriceSink priceSink);
void stop();
}
src/main/tjava/com/r35157/assetaz/core/service/ticker/TickerService.tjava
+5
View File
@@ -15,6 +15,11 @@ public interface TickerService {
*/
void start();
/**
* Stops every source started by this service.
*/
void stop();
/**
* Returns the latest available price observation for a trading pair.
*
src/main/tjava/com/r35157/assetaz/core/service/ticker/impl/ref/HardcodedPriceSource.tjava
+114
View File
@@ -0,0 +1,114 @@
package com.r35157.assetaz.core.service.ticker.impl.ref;
import com.r35157.assetaz.core.service.ticker.PriceSink;
import com.r35157.assetaz.core.service.ticker.PriceSource;
import com.r35157.libs.valuetypes.basic.TradingPair;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.math.BigDecimal;
import java.time.Clock;
import java.time.temporal.ChronoUnit;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static com.r35157.libs.valuetypes.basic.WellKnownTradingPairs.EVE_USDC;
/**
* Temporary reference source that produces a constant EVE/USDC price.
*/
public final class HardcodedPriceSource implements PriceSource {
public HardcodedPriceSource() {
this(Clock.systemUTC(), OBSERVATION_DELAY_MINUTES, TimeUnit.MINUTES);
}
HardcodedPriceSource(
@NotNull Clock clock,
long observationDelay,
@NotNull TimeUnit observationDelayUnit
) {
this.clock = Objects.requireNonNull(clock, "clock");
if (observationDelay <= 0) {
throw new IllegalArgumentException("observationDelay must be positive");
}
this.observationDelay = observationDelay;
this.observationDelayUnit = Objects.requireNonNull(
observationDelayUnit,
"observationDelayUnit"
);
}
@Override
public @NotNull TradingPair getTradingPair() {
return EVE_USDC.getTradingPair();
}
@Override
public @NotNull ΩPriceSourceNameΩ getSourceName() {
return SOURCE_NAME;
}
@Override
public synchronized void start(@NotNull PriceSink priceSink) {
if (scheduler != null) {
throw new IllegalStateException("Hardcoded price source is already started");
}
this.priceSink = Objects.requireNonNull(priceSink, "priceSink");
announcePrice();
scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "assetaz-hardcoded-price-source");
thread.setDaemon(true);
return thread;
});
scheduler.scheduleWithFixedDelay(
this::announcePriceSafely,
observationDelay,
observationDelay,
observationDelayUnit
);
}
@Override
public synchronized void stop() {
if (scheduler == null) {
return;
}
scheduler.shutdownNow();
scheduler = null;
}
private void announcePrice() {
priceSink.announce(
this,
new ΩPriceΩ(HARDCODED_PRICE),
clock.instant().truncatedTo(ChronoUnit.MILLIS)
);
}
private void announcePriceSafely() {
try {
announcePrice();
} catch (RuntimeException exception) {
log.error("Hardcoded price source failed to announce EVE_USDC price", exception);
}
}
private static final Logger log = LoggerFactory.getLogger(HardcodedPriceSource.class);
private static final ΩPriceSourceNameΩ SOURCE_NAME = "Hardcoded";
private static final String HARDCODED_PRICE = "14.85";
private static final long OBSERVATION_DELAY_MINUTES = 1;
private PriceSink priceSink;
private final Clock clock;
private final long observationDelay;
private final TimeUnit observationDelayUnit;
private ScheduledExecutorService scheduler;
}
src/main/tjava/com/r35157/assetaz/core/service/ticker/impl/ref/TickerServiceImpl.tjava
+445 -128
View File
@@ -1,6 +1,8 @@
package com.r35157.assetaz.core.service.ticker.impl.ref;
import com.r35157.assetaz.core.service.ticker.PriceObservation;
import com.r35157.assetaz.core.service.ticker.PriceSink;
import com.r35157.assetaz.core.service.ticker.PriceSource;
import com.r35157.assetaz.core.service.ticker.TickerService;
import com.r35157.libs.valuetypes.basic.AssetPrice;
import com.r35157.libs.valuetypes.basic.TradingPair;
@@ -17,60 +19,98 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.time.format.ResolverStyle;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static com.r35157.libs.valuetypes.basic.WellKnownTradingPairs.EVE_USDC;
public final class TickerServiceImpl implements TickerService, PriceSink {
public TickerServiceImpl(PriceSource... priceSources) {
this(DATA_ROOT);
public final class TickerServiceImpl implements TickerService {
public TickerServiceImpl() {
this(PRICE_HISTORY_PATH, Clock.systemUTC());
for (PriceSource priceSource : priceSources) {
addPriceSource(priceSource);
}
}
TickerServiceImpl(@NotNull Path priceHistoryPath, Clock clock) {
this.priceHistoryPath = Objects.requireNonNull(priceHistoryPath, "priceHistoryPath");
this.clock = Objects.requireNonNull(clock, "clock");
TickerServiceImpl(@NotNull Path dataRoot) {
this.dataRoot = Objects.requireNonNull(dataRoot, "dataRoot");
}
@Override
public void start() {
public synchronized void start() {
if (state != LifecycleState.NEW && state != LifecycleState.STOPPED) {
throw new IllegalStateException("Ticker service is already starting or started");
}
state = LifecycleState.STARTING;
latestByPair.clear();
registrationsByKey.values().forEach(SourceRegistration::resetLifecycle);
try {
startInternal();
} catch (IOException exception) {
throw new IllegalStateException(
"Could not start Ticker service using history file: " + priceHistoryPath,
exception
);
activateAndLoadHistories();
state = LifecycleState.STARTED;
startActiveSources();
} catch (IOException | RuntimeException exception) {
rollbackFailedStart(exception);
}
}
@Override
public @NotNull PriceObservation getLatestPrice(@NotNull TradingPair tradingPair) {
public synchronized void stop() {
if (state == LifecycleState.NEW || state == LifecycleState.STOPPED) {
return;
}
if (state != LifecycleState.STARTED) {
throw new IllegalStateException("Ticker service cannot stop while in state " + state);
}
state = LifecycleState.STOPPING;
RuntimeException failure = stopStartedSources();
resetStoppedState();
if (failure != null) {
throw failure;
}
}
@Override
public synchronized @NotNull PriceObservation getLatestPrice(
@NotNull TradingPair tradingPair
) {
Objects.requireNonNull(tradingPair, "tradingPair");
if (!SUPPORTED_PAIR.equals(tradingPair)) {
boolean registered = registrationsByKey.keySet().stream()
.anyMatch(key -> key.tradingPair().equals(tradingPair));
if (!registered) {
throw new IllegalArgumentException(
"Ticker price is unavailable for unsupported trading pair: " + tradingPair
);
}
if (!active) {
throw new IllegalStateException(
"Ticker price is unavailable for inactive trading pair: " + tradingPair
"Ticker has no registered price source for trading pair: " + tradingPair
);
}
PriceObservation observation = latest.get();
boolean active = registrationsByKey.values().stream()
.anyMatch(registration -> registration.active
&& registration.key.tradingPair().equals(tradingPair));
if (!active || state != LifecycleState.STARTED) {
throw new IllegalStateException(
"Ticker has no active price source for trading pair: " + tradingPair
);
}
AtomicReference<PriceObservation> latestReference = latestByPair.get(tradingPair);
PriceObservation observation = latestReference == null ? null : latestReference.get();
if (observation == null) {
throw new IllegalStateException(
"No persisted price is available for trading pair: " + tradingPair
@@ -80,53 +120,203 @@ public final class TickerServiceImpl implements TickerService {
return observation;
}
private synchronized void startInternal() throws IOException {
if (started) {
throw new IllegalStateException("Ticker service has already been started");
}
started = true;
if (!Files.exists(priceHistoryPath)) {
log.warn(
"Ticker pair {} is inactive because its history file is missing: {}",
SUPPORTED_PAIR,
priceHistoryPath
@Override
public void announce(
@NotNull PriceSource source,
@NotNull ΩPriceΩ price,
@NotNull Instant observedAt
) {
Objects.requireNonNull(source, "source");
Objects.requireNonNull(price, "price");
Objects.requireNonNull(observedAt, "observedAt");
if (observedAt.getNano() % NANOS_PER_MILLISECOND != 0) {
throw new IllegalArgumentException(
"Ticker observation timestamp must have millisecond precision: " + observedAt
);
return;
}
PriceObservation loadedLatest;
callbackLock.readLock().lock();
try {
loadedLatest = loadLatestObservation();
} catch (IOException | RuntimeException exception) {
started = false;
throw exception;
SourceRegistration registration = registrationsByInstance.get(source);
if (registration == null) {
throw new IllegalArgumentException("Price callback came from an unregistered source");
}
if (state != LifecycleState.STARTED || !registration.active) {
throw new IllegalStateException(
"Price callback came from an inactive source: "
+ registration.key.tradingPair() + " / "
+ registration.key.sourceName()
);
}
PriceObservation observation = new PriceObservation(
new AssetPrice(price, registration.key.tradingPair()),
observedAt,
registration.key.sourceName()
);
persistAndPublish(registration, observation);
} finally {
callbackLock.readLock().unlock();
}
latest.set(loadedLatest);
active = true;
persistGeneratedObservation();
scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "assetaz-ticker-EVE_USDC");
thread.setDaemon(true);
return thread;
});
scheduler.scheduleWithFixedDelay(
this::persistPeriodicObservation,
OBSERVATION_DELAY_MINUTES,
OBSERVATION_DELAY_MINUTES,
TimeUnit.MINUTES
);
}
private PriceObservation loadLatestObservation() throws IOException {
private synchronized void addPriceSource(@NotNull PriceSource priceSource) {
Objects.requireNonNull(priceSource, "priceSource");
if (state != LifecycleState.NEW) {
throw new IllegalStateException(
"Price sources must be registered before the Ticker service is started"
);
}
if (registrationsByInstance.containsKey(priceSource)) {
throw new IllegalArgumentException("Price source instance is already registered");
}
TradingPair tradingPair = validateTradingPair(priceSource.getTradingPair());
ΩPriceSourceNameΩ sourceName = validateSourceName(priceSource.getSourceName());
SourceKey key = new SourceKey(tradingPair, sourceName);
Path historyPath = historyPath(tradingPair, sourceName);
if (registrationsByKey.containsKey(key)) {
throw new IllegalArgumentException(
"Duplicate price source registration for " + tradingPair
+ " source " + sourceName
+ " addresses history " + historyPath
);
}
SourceRegistration existingHistoryRegistration =
registrationsByHistoryPath.get(historyPath);
if (existingHistoryRegistration != null) {
throw new IllegalArgumentException(
"Price source history is already registered: " + historyPath
+ " is used by "
+ existingHistoryRegistration.key.tradingPair()
+ " source "
+ existingHistoryRegistration.key.sourceName()
);
}
SourceRegistration registration = new SourceRegistration(
priceSource,
key,
historyPath
);
registrationsByKey.put(key, registration);
registrationsByInstance.put(priceSource, registration);
registrationsByHistoryPath.put(historyPath, registration);
}
private void activateAndLoadHistories() throws IOException {
for (SourceRegistration registration : registrationsByKey.values()) {
validateRegistrationMetadata(registration);
if (!Files.exists(registration.historyPath)) {
log.warn(
"Ticker source is inactive because its history is missing: pair={}, source={}, path={}",
registration.key.tradingPair(),
registration.key.sourceName(),
registration.historyPath
);
continue;
}
PriceObservation loadedLatest = loadLatestObservation(registration);
registration.active = true;
considerLatest(loadedLatest);
}
}
private void validateRegistrationMetadata(SourceRegistration registration) {
TradingPair currentPair = validateTradingPair(registration.source.getTradingPair());
ΩPriceSourceNameΩ currentName = validateSourceName(registration.source.getSourceName());
if (!registration.key.tradingPair().equals(currentPair)
|| !registration.key.sourceName().equals(currentName)) {
throw new IllegalStateException(
"Registered price source identity changed before startup: expected "
+ registration.key.tradingPair() + " / "
+ registration.key.sourceName() + ", found "
+ currentPair + " / " + currentName
);
}
}
private void startActiveSources() {
for (SourceRegistration registration : registrationsByKey.values()) {
if (!registration.active) {
continue;
}
registration.started = true;
registration.source.start(this);
}
}
private void rollbackFailedStart(Exception startupFailure) {
state = LifecycleState.STOPPING;
RuntimeException stopFailure = stopStartedSources();
resetStoppedState();
IllegalStateException failure = new IllegalStateException(
"Could not start Ticker service: " + startupFailure.getMessage(),
startupFailure
);
if (stopFailure != null) {
failure.addSuppressed(stopFailure);
}
throw failure;
}
private RuntimeException stopStartedSources() {
// State is already STOPPING. Taking and releasing the write lock drains
// callbacks that were accepted while STARTED. Do not hold it while a
// source stops, because the source may wait for one of its own threads.
callbackLock.writeLock().lock();
try {
} finally {
callbackLock.writeLock().unlock();
}
RuntimeException failure = null;
List<SourceRegistration> registrations = new ArrayList<>(
registrationsByKey.values()
);
for (int index = registrations.size() - 1; index >= 0; index--) {
SourceRegistration registration = registrations.get(index);
if (!registration.started) {
continue;
}
try {
registration.source.stop();
} catch (RuntimeException exception) {
if (failure == null) {
failure = new IllegalStateException(
"One or more price sources failed to stop"
);
}
failure.addSuppressed(exception);
} finally {
registration.started = false;
}
}
return failure;
}
private void resetStoppedState() {
registrationsByKey.values().forEach(SourceRegistration::resetLifecycle);
latestByPair.clear();
state = LifecycleState.STOPPED;
}
private PriceObservation loadLatestObservation(
SourceRegistration registration
) throws IOException {
PriceObservation loadedLatest = null;
try (BufferedReader reader = Files.newBufferedReader(
priceHistoryPath,
registration.historyPath,
StandardCharsets.UTF_8
)) {
String rawLine;
@@ -139,7 +329,12 @@ public final class TickerServiceImpl implements TickerService {
continue;
}
PriceObservation observation = parseObservation(data, rawLine, lineNumber);
PriceObservation observation = parseObservation(
registration,
data,
rawLine,
lineNumber
);
if (loadedLatest == null
|| observation.observedAt().isAfter(loadedLatest.observedAt())) {
loadedLatest = observation;
@@ -151,6 +346,7 @@ public final class TickerServiceImpl implements TickerService {
}
private PriceObservation parseObservation(
SourceRegistration registration,
String data,
String rawLine,
int lineNumber
@@ -158,7 +354,7 @@ public final class TickerServiceImpl implements TickerService {
int separator = data.indexOf(':');
if (separator <= 0 || separator != data.lastIndexOf(':')
|| separator == data.length() - 1) {
throw malformedHistory(lineNumber, rawLine, null);
throw malformedHistory(registration.historyPath, lineNumber, rawLine, null);
}
try {
@@ -170,68 +366,72 @@ public final class TickerServiceImpl implements TickerService {
ΩPriceΩ price = new ΩPriceΩ(data.substring(separator + 1));
return new PriceObservation(
new AssetPrice(price, SUPPORTED_PAIR),
observedAt
new AssetPrice(price, registration.key.tradingPair()),
observedAt,
registration.key.sourceName()
);
} catch (RuntimeException exception) {
throw malformedHistory(lineNumber, rawLine, exception);
}
}
private IOException malformedHistory(
int lineNumber,
String rawLine,
RuntimeException cause
) {
return new IOException(
"Malformed ticker history in " + priceHistoryPath
+ " at line " + lineNumber + ": " + rawLine,
cause
);
}
private synchronized void persistGeneratedObservation() {
PriceObservation observation = new PriceObservation(
new AssetPrice(new ΩPriceΩ(HARDCODED_PRICE), SUPPORTED_PAIR),
clock.instant().truncatedTo(ChronoUnit.MILLIS)
);
try {
persist(observation);
latest.updateAndGet(current -> current == null
|| observation.observedAt().isAfter(current.observedAt())
? observation
: current);
} catch (IOException exception) {
log.error(
"Failed to persist ticker observation for {} to {}; retaining prior latest observation",
SUPPORTED_PAIR,
priceHistoryPath,
throw malformedHistory(
registration.historyPath,
lineNumber,
rawLine,
exception
);
}
}
private void persistPeriodicObservation() {
try {
persistGeneratedObservation();
} catch (RuntimeException exception) {
log.error("Unexpected failure generating ticker observation for {}", SUPPORTED_PAIR, exception);
}
private static IOException malformedHistory(
Path historyPath,
int lineNumber,
String rawLine,
RuntimeException cause
) {
return new IOException(
"Malformed ticker history in " + historyPath
+ " at line " + lineNumber + ": " + rawLine,
cause
);
}
private void persist(PriceObservation observation) throws IOException {
private void persistAndPublish(
SourceRegistration registration,
PriceObservation observation
) {
registration.persistenceLock.lock();
try {
persist(registration.historyPath, observation);
} catch (IOException exception) {
log.error(
"Failed to persist ticker observation: pair={}, source={}, path={}; retaining prior latest observation",
registration.key.tradingPair(),
registration.key.sourceName(),
registration.historyPath,
exception
);
return;
} finally {
registration.persistenceLock.unlock();
}
considerLatest(observation);
}
private static void persist(
Path historyPath,
PriceObservation observation
) throws IOException {
String encodedObservation = TIMESTAMP_FORMATTER.format(
LocalDateTime.ofInstant(observation.observedAt(), ZoneOffset.UTC)
) + ":" + observation.price().price().toPlainString() + "\n";
try (FileChannel channel = FileChannel.open(
priceHistoryPath,
historyPath,
StandardOpenOption.READ,
StandardOpenOption.WRITE
)) {
long size = channel.size();
boolean needsLineSeparator = size > 0 && !endsWithLineSeparator(channel, size);
boolean needsLineSeparator = size > 0
&& !endsWithLineSeparator(channel, size, historyPath);
channel.position(size);
if (needsLineSeparator) {
@@ -245,11 +445,91 @@ public final class TickerServiceImpl implements TickerService {
}
}
private boolean endsWithLineSeparator(FileChannel channel, long size) throws IOException {
private void considerLatest(PriceObservation observation) {
if (observation == null) {
return;
}
latestByPair.computeIfAbsent(
observation.price().tradingPair(),
ignored -> new AtomicReference<>()
).updateAndGet(current -> current == null
|| observation.observedAt().isAfter(current.observedAt())
? observation
: current);
}
private Path historyPath(
TradingPair tradingPair,
ΩPriceSourceNameΩ sourceName
) {
String filename = safeSymbol(tradingPair.base().symbol())
+ "_" + safeSymbol(tradingPair.quote().symbol())
+ ".prices";
return dataRoot
.resolve(tradingPair.base().id().toString())
.resolve(tradingPair.quote().id().toString())
.resolve(sourceName)
.resolve(filename);
}
private static TradingPair validateTradingPair(TradingPair tradingPair) {
Objects.requireNonNull(tradingPair, "priceSource.tradingPair");
Objects.requireNonNull(tradingPair.base(), "priceSource.tradingPair.base");
Objects.requireNonNull(tradingPair.quote(), "priceSource.tradingPair.quote");
Objects.requireNonNull(tradingPair.base().id(), "priceSource.tradingPair.base.id");
Objects.requireNonNull(tradingPair.quote().id(), "priceSource.tradingPair.quote.id");
Objects.requireNonNull(tradingPair.base().symbol(), "priceSource.tradingPair.base.symbol");
Objects.requireNonNull(tradingPair.quote().symbol(), "priceSource.tradingPair.quote.symbol");
return tradingPair;
}
private static ΩPriceSourceNameΩ validateSourceName(ΩPriceSourceNameΩ sourceName) {
Objects.requireNonNull(sourceName, "priceSource.sourceName");
if (sourceName.isBlank()
|| sourceName.equals(".")
|| sourceName.contains("/")
|| sourceName.contains("\\")
|| sourceName.contains("..")
|| sourceName.codePoints().anyMatch(Character::isISOControl)) {
throw new IllegalArgumentException(
"Unsafe price source name cannot be used as a history directory: "
+ sourceName
);
}
return sourceName;
}
private static String safeSymbol(String symbol) {
if (symbol.isEmpty()) {
throw new IllegalArgumentException("Currency symbol cannot be empty");
}
StringBuilder safe = new StringBuilder();
symbol.codePoints().forEach(codePoint -> {
if (codePoint >= 'A' && codePoint <= 'Z'
|| codePoint >= 'a' && codePoint <= 'z'
|| codePoint >= '0' && codePoint <= '9'
|| codePoint == '-'
|| codePoint == '_') {
safe.appendCodePoint(codePoint);
} else {
safe.append('_');
}
});
return safe.toString();
}
private static boolean endsWithLineSeparator(
FileChannel channel,
long size,
Path historyPath
) throws IOException {
ByteBuffer finalByte = ByteBuffer.allocate(1);
channel.position(size - 1);
if (channel.read(finalByte) != 1) {
throw new IOException("Could not read final byte of ticker history: " + priceHistoryPath);
throw new IOException("Could not read final byte of ticker history: " + historyPath);
}
return finalByte.array()[0] == '\n' || finalByte.array()[0] == '\r';
@@ -266,25 +546,62 @@ public final class TickerServiceImpl implements TickerService {
return commentStart < 0 ? line : line.substring(0, commentStart);
}
private enum LifecycleState {
NEW,
STARTING,
STARTED,
STOPPING,
STOPPED
}
private record SourceKey(
TradingPair tradingPair,
ΩPriceSourceNameΩ sourceName
) {
}
private static final class SourceRegistration {
private SourceRegistration(
PriceSource source,
SourceKey key,
Path historyPath
) {
this.source = source;
this.key = key;
this.historyPath = historyPath;
}
private void resetLifecycle() {
active = false;
started = false;
}
private final PriceSource source;
private final SourceKey key;
private final Path historyPath;
private final ReentrantLock persistenceLock = new ReentrantLock();
private volatile boolean active;
private boolean started;
}
private static final Logger log = LoggerFactory.getLogger(TickerServiceImpl.class);
private static final TradingPair SUPPORTED_PAIR = EVE_USDC.getTradingPair();
private static final Path PRICE_HISTORY_PATH = Path.of(
"data",
"assetaz",
"ticker",
"EVE_USDC.prices"
);
private static final Path DATA_ROOT = Path.of("data", "assetaz", "ticker");
private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter
.ofPattern("uuuuMMddHHmmssSSS'Z'", Locale.ROOT)
.withResolverStyle(ResolverStyle.STRICT);
private static final String HARDCODED_PRICE = "14.85";
private static final long OBSERVATION_DELAY_MINUTES = 1;
private static final int NANOS_PER_MILLISECOND = 1_000_000;
private final AtomicReference<PriceObservation> latest = new AtomicReference<>();
private final Path priceHistoryPath;
private final Clock clock;
private final Path dataRoot;
private final Map<SourceKey, SourceRegistration> registrationsByKey =
new LinkedHashMap<>();
private final Map<PriceSource, SourceRegistration> registrationsByInstance =
new IdentityHashMap<>();
private final Map<Path, SourceRegistration> registrationsByHistoryPath =
new LinkedHashMap<>();
private final Map<TradingPair, AtomicReference<PriceObservation>> latestByPair =
new ConcurrentHashMap<>();
private final ReentrantReadWriteLock callbackLock = new ReentrantReadWriteLock();
private volatile boolean active;
private boolean started;
private ScheduledExecutorService scheduler;
private volatile LifecycleState state = LifecycleState.NEW;
}
src/main/tjava/com/r35157/nenjim/hubd/impl/ref/NenjimHubImpl.tjava
+10 -2
View File
@@ -2,6 +2,7 @@ package com.r35157.nenjim.hubd.impl.ref;
import com.fanitas.evelyn.core.Evelyn;
import com.fanitas.evelyn.core.impl.ref.EvelynImpl;
import com.r35157.assetaz.core.service.ticker.impl.ref.HardcodedPriceSource;
import com.r35157.assetaz.core.service.ticker.impl.ref.TickerServiceImpl;
import com.r35157.evelyn.emc.EvelynMissionControl;
import com.r35157.evelyn.emc.impl.ref.EvelynMissionControlImpl;
@@ -88,8 +89,15 @@ public class NenjimHubImpl implements NenjimHub {
*/
}
private void startAssetAZTickerService() throws Exception {
new TickerServiceImpl().start();
private void startAssetAZTickerService() {
// Nenjim creates this unstarted PriceSource first...
HardcodedPriceSource priceSource = new HardcodedPriceSource();
// The TickerServiceImpl will ask Nenjim for implementers of the PriceSource interface in this context
// This do not work yet - so we will just inject it in the constructor now. In the future it will
// not be injected in the constructor but TickerServiceImpl will ask Nenjim for them.
TickerServiceImpl tickerService = new TickerServiceImpl(priceSource);
tickerService.start();
}
@Override