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

63: Add Raydium Pool PriceSource with periodic polling

Browse Source
This commit is contained in:
minimons
2026-08-08 16:10:28 +02:00
parent cac555159d
commit d5b733bc6f
12 changed files with 546 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/tjava/com/r35157/assetaz/services/cis/CurrencyTypeIds.tjava
+4
View File
@@ -21,6 +21,10 @@ public final class CurrencyTypeIds {
public static final UUID USDC_ID =
UUID.fromString("019c3f9f-41d1-7a73-b1df-d4c11c7ff302");
/** Stable UUID for Tether USD. */
public static final UUID USDT_ID =
UUID.fromString("c8669973-0045-468e-8b2b-781d06d123b2");
/** Stable UUID for Solana. */
public static final UUID SOLANA_ID =
UUID.fromString("019e0116-fce5-792f-a647-fa6da4dffec5");
src/main/tjava/com/r35157/assetaz/services/cis/impl/hc/HardcodedCurrencyIdentityService.tjava
+3
View File
@@ -18,6 +18,7 @@ import static com.r35157.assetaz.services.cis.CurrencyTypeIds.EVE_ID;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.SOLANA_ID;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.SYRUPUSDC_ID;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.USDC_ID;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.USDT_ID;
public final class HardcodedCurrencyIdentityService implements CurrencyIdentityService {
public HardcodedCurrencyIdentityService() {
@@ -26,6 +27,8 @@ public final class HardcodedCurrencyIdentityService implements CurrencyIdentityS
solanaMint("meveYG2iXYSkgSUn1T1uxcthH1EGMZdRHGgCntXZA3Y", "EVE")),
entry(USDC_ID, "USD Coin", "USDC",
solanaMint("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", "USDC")),
entry(USDT_ID, "Tether USD", "USDT",
solanaMint("Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB", "USDT")),
entry(SOLANA_ID, "Solana", "SOL",
solanaMint("So11111111111111111111111111111111111111112", "SOL")),
entry(SYRUPUSDC_ID, "SyrupUSDC", "SyrupUSDC",
src/main/tjava/com/r35157/assetaz/services/ticker/plugins/pricesource/impl/raydiumpool/RaydiumPoolPriceSource.tjava
+191 -5
View File
@@ -2,27 +2,213 @@ package com.r35157.assetaz.services.ticker.plugins.pricesource.impl.raydiumpool;
import com.r35157.assetaz.services.ticker.plugins.pricesource.PriceSink;
import com.r35157.assetaz.services.ticker.plugins.pricesource.PriceSource;
import com.r35157.libs.raydium.Raydium;
import com.r35157.libs.valuetypes.basic.AssetPrice;
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;
public final class RaydiumPoolPriceSource implements PriceSource {
public RaydiumPoolPriceSource(
@NotNull Raydium raydium,
@NotNull ΩRaydiumLiquidityPoolIdΩ poolId,
@NotNull TradingPair expectedTradingPair
) {
this(
raydium,
poolId,
expectedTradingPair,
Clock.systemUTC(),
POLLING_DELAY_MINUTES,
TimeUnit.MINUTES
);
}
RaydiumPoolPriceSource(
@NotNull Raydium raydium,
@NotNull ΩRaydiumLiquidityPoolIdΩ poolId,
@NotNull TradingPair expectedTradingPair,
@NotNull Clock clock,
long pollingDelay,
@NotNull TimeUnit pollingDelayUnit
) {
this.raydium = Objects.requireNonNull(raydium, "raydium");
this.poolId = Objects.requireNonNull(poolId, "poolId");
if (poolId.isBlank()) {
throw new IllegalArgumentException("poolId must not be blank");
}
this.expectedTradingPair = Objects.requireNonNull(
expectedTradingPair,
"expectedTradingPair"
);
Objects.requireNonNull(expectedTradingPair.base(), "expectedTradingPair.base");
Objects.requireNonNull(expectedTradingPair.quote(), "expectedTradingPair.quote");
this.sourceName = "Raydium-" + poolId;
this.clock = Objects.requireNonNull(clock, "clock");
if (pollingDelay <= 0) {
throw new IllegalArgumentException("pollingDelay must be positive");
}
this.pollingDelay = pollingDelay;
this.pollingDelayUnit = Objects.requireNonNull(
pollingDelayUnit,
"pollingDelayUnit"
);
}
@Override
public @NotNull TradingPair getTradingPair() {
throw new UnsupportedOperationException("Not Implemented");
return expectedTradingPair;
}
@Override
public @NotNull ΩPriceSourceNameΩ getSourceName() {
throw new UnsupportedOperationException("Not Implemented");
return sourceName;
}
@Override
public void start(@NotNull PriceSink priceSink) {
throw new UnsupportedOperationException("Not Implemented");
public synchronized void start(@NotNull PriceSink priceSink) {
if (scheduler != null || stopping) {
throw new IllegalStateException("Raydium pool price source is already started");
}
PriceSink newPriceSink = Objects.requireNonNull(priceSink, "priceSink");
ScheduledExecutorService newScheduler = Executors.newSingleThreadScheduledExecutor(
runnable -> {
Thread thread = new Thread(
runnable,
"assetaz-raydium-pool-price-source-" + poolId
);
thread.setDaemon(true);
return thread;
}
);
this.priceSink = newPriceSink;
this.scheduler = newScheduler;
newScheduler.scheduleWithFixedDelay(
() -> pollSafely(newScheduler, newPriceSink),
0,
pollingDelay,
pollingDelayUnit
);
}
@Override
public void stop() {
throw new UnsupportedOperationException("Not Implemented");
ScheduledExecutorService schedulerToStop;
synchronized (this) {
if (scheduler == null) {
return;
}
if (stopping) {
return;
}
stopping = true;
schedulerToStop = scheduler;
priceSink = null;
}
schedulerToStop.shutdownNow();
try {
if (!schedulerToStop.awaitTermination(
TERMINATION_TIMEOUT_SECONDS,
TimeUnit.SECONDS
)) {
log.error(
"Raydium pool price source scheduler did not terminate: poolId={}, tradingPair={}",
poolId,
expectedTradingPair
);
}
} catch (InterruptedException exception) {
schedulerToStop.shutdownNow();
Thread.currentThread().interrupt();
} finally {
synchronized (this) {
if (schedulerToStop.isTerminated()) {
scheduler = null;
}
stopping = false;
}
}
}
private void pollSafely(
ScheduledExecutorService pollingScheduler,
PriceSink pollingSink
) {
try {
AssetPrice assetPrice = raydium.fetchPoolPrice(poolId);
TradingPair receivedTradingPair = assetPrice.tradingPair();
if (!expectedTradingPair.equals(receivedTradingPair)) {
log.error(
"Raydium pool returned unexpected trading pair: poolId={}, expected={}, received={}",
poolId,
expectedTradingPair,
receivedTradingPair
);
return;
}
if (!isCurrentRun(pollingScheduler, pollingSink)) {
return;
}
pollingSink.announce(
this,
assetPrice.price(),
clock.instant().truncatedTo(ChronoUnit.MILLIS)
);
} catch (InterruptedException exception) {
Thread.currentThread().interrupt();
if (!isCurrentRun(pollingScheduler, pollingSink)) {
return;
}
log.error(
"Raydium pool price polling was interrupted: poolId={}, tradingPair={}",
poolId,
expectedTradingPair,
exception
);
} catch (Exception exception) {
log.error(
"Raydium pool price polling failed: poolId={}, tradingPair={}",
poolId,
expectedTradingPair,
exception
);
}
}
private synchronized boolean isCurrentRun(
ScheduledExecutorService pollingScheduler,
PriceSink pollingSink
) {
return scheduler == pollingScheduler && priceSink == pollingSink && !stopping;
}
private static final Logger log = LoggerFactory.getLogger(RaydiumPoolPriceSource.class);
private static final long POLLING_DELAY_MINUTES = 1;
private static final long TERMINATION_TIMEOUT_SECONDS = 10;
private final Raydium raydium;
private final ΩRaydiumLiquidityPoolIdΩ poolId;
private final TradingPair expectedTradingPair;
private final ΩPriceSourceNameΩ sourceName;
private final Clock clock;
private final long pollingDelay;
private final TimeUnit pollingDelayUnit;
private PriceSink priceSink;
private ScheduledExecutorService scheduler;
private boolean stopping;
}
src/main/tjava/com/r35157/nenjim/hubd/impl/ref/NenjimHubImpl.tjava
+32 -15
View File
@@ -2,7 +2,9 @@ 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.services.ticker.plugins.pricesource.PriceSource;
import com.r35157.assetaz.services.ticker.plugins.pricesource.impl.hardcoded.HardcodedPriceSource;
import com.r35157.assetaz.services.ticker.plugins.pricesource.impl.raydiumpool.RaydiumPoolPriceSource;
import com.r35157.assetaz.services.ticker.impl.ref.TickerServiceImpl;
import com.r35157.assetaz.services.cis.CurrencyIdentityService;
import com.r35157.assetaz.services.cis.impl.hc.HardcodedCurrencyIdentityService;
@@ -16,6 +18,11 @@ import com.r35157.nenjim.npm.NenjimProcessManager;
import com.r35157.nenjim.npm.impl.ref.NenjimProcessManagerImpl;
import com.r35157.nenjim.ntt.NenjimTestTool;
import com.r35157.nenjim.ntt.impl.ref.NenjimTestToolImpl;
import com.r35157.libs.raydium.Raydium;
import com.r35157.libs.raydium.impl.ref.RaydiumImpl;
import com.r35157.libs.solana.SolanaBlockChain;
import com.r35157.libs.solana.impl.ref.SolanaBlockChainImpl;
import com.r35157.libs.valuetypes.basic.TradingPair;
import com.r35157.stm.SodaTaskManager;
import com.r35157.stm.impl.ref.SodaTaskManagerImpl;
import com.r35157.suwimo.hub.client.SuwimoClient;
@@ -27,6 +34,9 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.EVE_ID;
import static com.r35157.assetaz.services.cis.CurrencyTypeIds.USDT_ID;
public class NenjimHubImpl implements NenjimHub {
public NenjimHubImpl() throws Exception {
log.info("Initializing NenjimHub...");
@@ -61,10 +71,17 @@ public class NenjimHubImpl implements NenjimHub {
}
private void startAutoRunProcesses() {
CurrencyIdentityService currencyIdentityService =
new HardcodedCurrencyIdentityService();
startAssetAZTickerService(currencyIdentityService);
startJupiterPerpsAlarm(currencyIdentityService); // TODO: Hardcoded/hacky way to auto-start but good enough for now.
CurrencyIdentityService cis = new HardcodedCurrencyIdentityService();
SolanaBlockChain solanaBlockChain = new SolanaBlockChainImpl(cis);
Raydium raydium = new RaydiumImpl(solanaBlockChain, cis);
PriceSource hardcodedPriceSource = new HardcodedPriceSource(cis);
PriceSource raydiumPoolPriceSource = createEVEUSDTPriceSource(cis, raydium);
startAssetAZTickerService(hardcodedPriceSource, raydiumPoolPriceSource);
startJupiterPerpsAlarm(cis);
Evelyn evelynProd = new EvelynImpl();
Evelyn evelynTest = new EvelynImpl();
startEvelynMissionControl(evelynProd, evelynTest);
@@ -92,18 +109,16 @@ public class NenjimHubImpl implements NenjimHub {
*/
}
private void startAssetAZTickerService(
CurrencyIdentityService currencyIdentityService
) {
// Nenjim creates this unstarted PriceSource first...
HardcodedPriceSource priceSource = new HardcodedPriceSource(
currencyIdentityService
);
private PriceSource createEVEUSDTPriceSource(CurrencyIdentityService cis, Raydium raydium) {
TradingPair eveUsdt = new TradingPair(cis.resolve(EVE_ID), cis.resolve(USDT_ID));
PriceSource priceSource = new RaydiumPoolPriceSource(raydium, EVE_USDT_RAYDIUM_POOL_ID, eveUsdt);
return priceSource;
}
private void startAssetAZTickerService(PriceSource... priceSources) {
TickerServiceImpl tickerService = new TickerServiceImpl(priceSources);
// 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();
}
@@ -257,6 +272,8 @@ public class NenjimHubImpl implements NenjimHub {
}
private static final Logger log = LoggerFactory.getLogger(NenjimHubImpl.class);
private static final ΩRaydiumLiquidityPoolIdΩ EVE_USDT_RAYDIUM_POOL_ID =
"8rN4BTEzbogQosEQYgsEu18XwfKS5Yoxqwit8zEVFwEe";
private final CountDownLatch shutdownLatch = new CountDownLatch(1);
private HashMap<Integer, NenjimProcess> processes;