diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 3f7f76f..7fa6631 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -25,10 +25,10 @@ jobs: os: [ubuntu-24.04, ubuntu-22.04, macos-15, macos-14, windows-2022, windows-2025] steps: - uses: actions/checkout@v4 - - name: Set up JDK 21 + - name: Set up JDK 25 uses: actions/setup-java@v4 with: - java-version: "21" + java-version: "25" distribution: "temurin" cache: maven - name: Build with Maven diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 06634f7..9b31c45 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,10 +29,10 @@ jobs: os: [ubuntu-24.04, ubuntu-22.04, macos-15, macos-14, windows-2022, windows-2025] steps: - uses: actions/checkout@v4 - - name: Set up JDK 21 + - name: Set up JDK 25 uses: actions/setup-java@v4 with: - java-version: '21' + java-version: '25' distribution: 'temurin' cache: maven - name: Test with Maven diff --git a/Dockerfile b/Dockerfile index 986066e..678c901 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ # syntax=docker/dockerfile:1 -FROM maven:3.9.9-eclipse-temurin-21 AS build +FROM maven:3.9.11-eclipse-temurin-25 AS build WORKDIR /workspace COPY pom.xml ./ @@ -9,7 +9,7 @@ COPY src ./src # Build a Spring Boot executable jar. RUN mvn --batch-mode -DskipTests package -FROM eclipse-temurin:21-jre +FROM eclipse-temurin:25-jre WORKDIR /app COPY --from=build /workspace/target/recceiver-*.jar /app/recceiver.jar diff --git a/pom.xml b/pom.xml index f185492..d23d6e6 100644 --- a/pom.xml +++ b/pom.xml @@ -9,11 +9,10 @@ 4.7.2-SNAPSHOT - 21 - 21 + 25 + 25 UTF-8 - 2.7.12 - 2.1.10 + 4.1.0 @@ -41,11 +40,9 @@ com.fasterxml.jackson.core jackson-databind - - javax.xml.bind - jaxb-api - 2.3.1 + jakarta.annotation + jakarta.annotation-api @@ -56,13 +53,6 @@ - - org.springframework.shell - spring-shell-dependencies - ${spring-shell.version} - pom - import - org.springframework.boot @@ -93,9 +83,10 @@ org.apache.maven.plugins maven-compiler-plugin + 3.14.0 - 21 - 21 + ${maven.compiler.source} + ${maven.compiler.target} diff --git a/src/main/java/org/phoebus/recceiver/MessageProcessor.java b/src/main/java/org/phoebus/recceiver/MessageProcessor.java index 223f655..534640d 100644 --- a/src/main/java/org/phoebus/recceiver/MessageProcessor.java +++ b/src/main/java/org/phoebus/recceiver/MessageProcessor.java @@ -16,12 +16,14 @@ import org.phoebus.recceiver.spi.RecSyncProcessorDispatcher; import org.phoebus.recceiver.spi.RecSyncTransaction; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; public class MessageProcessor { private static final Logger log = Logger.getLogger(MessageProcessor.class.getName()); + private static final ThreadFactory COMMIT_THREAD_FACTORY = Thread.ofVirtual().name("recceiver-commit-", 0).factory(); private final SocketChannel channel; private final String connectionSourceAddress; @@ -355,8 +357,8 @@ private void dispatchTransaction() { // pong replies during CF uploads. For large initial batches (100k+ channels) the upload // can take minutes; blocking this thread would prevent any pings from being sent and the // IOC would close the connection with an EOF before the upload finishes. - Thread.ofVirtual().name("recceiver-commit-", 0).start(() -> - processorDispatcher.dispatchCommit(transaction)); + COMMIT_THREAD_FACTORY.newThread(() -> + processorDispatcher.dispatchCommit(transaction)).start(); } private void notifyConnectionClosed(String reason) { diff --git a/src/main/java/org/phoebus/recceiver/RecceiverApplication.java b/src/main/java/org/phoebus/recceiver/RecceiverApplication.java index eed70b2..c0dd4c9 100644 --- a/src/main/java/org/phoebus/recceiver/RecceiverApplication.java +++ b/src/main/java/org/phoebus/recceiver/RecceiverApplication.java @@ -9,7 +9,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.PropertySource; -import javax.annotation.PreDestroy; +import jakarta.annotation.PreDestroy; import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketException; @@ -25,6 +25,7 @@ import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -87,6 +88,9 @@ public class RecceiverApplication implements CommandLineRunner { @Value("${recceiver.cf.password:}") private String cfPassword; + @Value("${recceiver.cf.http-timeout:30}") + private String cfHttpTimeout; + @Value("${recceiver.cf.delete-on-clean:false}") private String cfDeleteOnClean; @@ -107,6 +111,7 @@ public class RecceiverApplication implements CommandLineRunner { private final ScheduledExecutorService udpAnnouncerService = Executors.newScheduledThreadPool(1); + private static final ThreadFactory CLIENT_THREAD_FACTORY = Thread.ofVirtual().name("recceiver-client-", 0).factory(); private static int serverKey; private RecSyncProcessorDispatcher processorDispatcher; private volatile boolean running = true; @@ -180,7 +185,7 @@ private void register(ServerSocketChannel serverSocket) client.configureBlocking(true); // Run each client handler on its own virtual thread. - Thread.ofVirtual().name("recceiver-client-", 0).start(() -> { + CLIENT_THREAD_FACTORY.newThread(() -> { MessageProcessor p = new MessageProcessor(client, Math.max(1L, Math.round(tcptimeout)), serverKey, processorDispatcher, readBufferSize); try { while (client.isOpen() && client.isConnected()) { @@ -202,7 +207,7 @@ private void register(ServerSocketChannel serverSocket) p.shutdown(); } - }); + }).start(); } private static boolean parseDisconnectEvents(IOException e) { @@ -234,6 +239,7 @@ private Map buildProcessorOptions() { options.put("recceiver.cf.service-url", cfServiceUrl); options.put("recceiver.cf.username", cfUsername); options.put("recceiver.cf.password", cfPassword); + options.put("recceiver.cf.http-timeout", cfHttpTimeout); options.put("recceiver.cf.delete-on-clean", cfDeleteOnClean); options.put("recceiver.cf.clean-on-startup", cfCleanOnStartup); options.put("recceiver.cf.delete-on-shutdown", cfDeleteOnShutdown); diff --git a/src/main/java/org/phoebus/recceiver/spi/RecSyncProcessorRegistry.java b/src/main/java/org/phoebus/recceiver/spi/RecSyncProcessorRegistry.java index e4305bc..b0473d2 100644 --- a/src/main/java/org/phoebus/recceiver/spi/RecSyncProcessorRegistry.java +++ b/src/main/java/org/phoebus/recceiver/spi/RecSyncProcessorRegistry.java @@ -19,6 +19,7 @@ public static RecSyncProcessorDispatcher loadDispatcher(Map opti ServiceLoader directLoader = ServiceLoader.load(RecSyncProcessor.class); for (RecSyncProcessor processor : directLoader) { + log.info("Loading RecSyncProcessor via SPI: " + processor.getClass().getName()); try { RecSyncProcessor initializedProcessor = processor.init(processor.name(), initOptions); if (initializedProcessor != null) { diff --git a/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java b/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java index 8c57723..5cb3af5 100644 --- a/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java +++ b/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java @@ -7,6 +7,7 @@ import java.net.InetAddress; import java.net.UnknownHostException; +import java.time.Duration; import java.time.OffsetDateTime; import java.util.ArrayList; import java.util.Collection; @@ -69,6 +70,7 @@ *
  • {@code recceiver.cf.enabled} (false by default): Enable this processor.
    • *
  • {@code recceiver.cf.service-url} (required if enabled): URL to the ChannelFinder service.
    • *
  • {@code recceiver.cf.username} / {@code recceiver.cf.password} (optional): HTTP basic auth.
    • + *
  • {@code recceiver.cf.http-timeout} (30 by default): Base HTTP timeout in seconds for ChannelFinder requests.
    • *
  • {@code recceiver.cf.alias-enabled} (false by default): Publish alias records as separate channels.
    • *
  • {@code recceiver.cf.infotags} (optional): Comma/space-separated list of info tag names to * publish as ChannelFinder properties.
    • @@ -86,6 +88,7 @@ public class CFRecSyncProcessor implements RecSyncProcessor { private static final String DELETE_ON_CLEAN_KEY = "recceiver.cf.delete-on-clean"; private static final String CLEAN_ON_STARTUP_KEY = "recceiver.cf.clean-on-startup"; private static final String DELETE_ON_SHUTDOWN_KEY = "recceiver.cf.delete-on-shutdown"; + private static final String HTTP_TIMEOUT_KEY = "recceiver.cf.http-timeout"; private static final String ALIAS_ENABLED_KEY = "recceiver.cf.alias-enabled"; private static final String INFOTAGS_KEY = "recceiver.cf.infotags"; private static final String ENVIRONMENT_VARS_KEY = "recceiver.cf.environment-vars"; @@ -134,6 +137,7 @@ public class CFRecSyncProcessor implements RecSyncProcessor { ))); private ChannelFinderStore store; + private Duration httpTimeout; private boolean deleteOnClean; private boolean cleanOnStartup; private boolean deleteOnShutdown; @@ -169,6 +173,7 @@ public RecSyncProcessor init(String name, Map options) { "recceiver.cf.username and recceiver.cf.password must both be configured when authentication is used"); } + this.httpTimeout = Duration.ofSeconds(Long.parseLong(options.getOrDefault(HTTP_TIMEOUT_KEY, "30").trim())); this.store = createStore(serviceUrl, username, password); this.deleteOnClean = Boolean.parseBoolean(options.getOrDefault(DELETE_ON_CLEAN_KEY, "false")); this.cleanOnStartup = Boolean.parseBoolean(options.getOrDefault(CLEAN_ON_STARTUP_KEY, "false")); @@ -186,6 +191,9 @@ public RecSyncProcessor init(String name, Map options) { requiredProperties.addAll(environmentVarMappings.values()); this.store.ensureProperties(requiredProperties, propertyOwner); + log.info("CFRecSyncProcessor initialized with serviceUrl=" + serviceUrl +" httpTimeout=" + httpTimeout + " cleanOnStartup=" + cleanOnStartup + " deleteOnClean=" + deleteOnClean + + " deleteOnShutdown=" + deleteOnShutdown + " aliasEnabled=" + aliasEnabled + + " infoTagWhitelist=" + infoTagWhitelist + " environmentVarMappings=" + environmentVarMappings); cleanupExistingActiveChannelsForRecceiver(); return this; @@ -217,6 +225,7 @@ private void cleanupExistingActiveChannelsForRecceiver() { // Startup cleanup mode: cleanOnStartup=true => delete, false => mark inactive. if (cleanOnStartup) { store.deleteChannels(activeChannels.keySet()); + log.info("CFRecSyncProcessor startup cleanup: deleted " + activeChannels.size() + " channels for recceiverID=" + RECCEIVER_ID); return; } @@ -229,12 +238,13 @@ private void cleanupExistingActiveChannelsForRecceiver() { inactiveChannels.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); } store.updateChannels(inactiveChannels); + log.info("CFRecSyncProcessor startup cleanup: marked " + inactiveChannels.size() + " channels as Inactive for recceiverID=" + RECCEIVER_ID); } protected ChannelFinderStore createStore(String serviceUrl, String username, String password) { return username == null - ? new ChannelFinderHttpStore(serviceUrl) - : new ChannelFinderHttpStore(serviceUrl, username, password); + ? new ChannelFinderHttpStore(serviceUrl, httpTimeout) + : new ChannelFinderHttpStore(serviceUrl, username, password, httpTimeout); } /** diff --git a/src/main/java/org/phoebus/recceiver/spi/cf/ChannelFinderHttpStore.java b/src/main/java/org/phoebus/recceiver/spi/cf/ChannelFinderHttpStore.java index 7976fd1..3a222b7 100644 --- a/src/main/java/org/phoebus/recceiver/spi/cf/ChannelFinderHttpStore.java +++ b/src/main/java/org/phoebus/recceiver/spi/cf/ChannelFinderHttpStore.java @@ -37,14 +37,13 @@ *
  • GET /resources/scroll/{scrollId}: Continue a scroll query when response id is present.
    • *
  • POST /resources/channels: Create or update a list of channels. * The request body is a JSON array of channel objects.
    • - *
  • DELETE /resources/channels/{name}: Delete a single channel.
    • + *
  • DELETE /resources/channels: Delete multiple channels in one request body list.
    • *
  • POST /resources/properties/{name}: Create a property (if it does not exist).
    • * *

    */ public class ChannelFinderHttpStore implements ChannelFinderStore { private static final Logger log = Logger.getLogger(ChannelFinderHttpStore.class.getName()); - private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(10); private static final Duration MAX_BULK_UPDATE_TIMEOUT = Duration.ofMinutes(5); private static final int CHANNELS_PER_TIMEOUT_STEP = 250; private static final Duration TIMEOUT_STEP = Duration.ofSeconds(1); @@ -53,20 +52,26 @@ public class ChannelFinderHttpStore implements ChannelFinderStore { private final ObjectMapper objectMapper; private final URI serviceUri; private final String authorizationHeader; + private final Duration timeout; - public ChannelFinderHttpStore(String serviceUrl) { - this(HttpClient.newHttpClient(), defaultObjectMapper(), URI.create(serviceUrl), null); + public ChannelFinderHttpStore(String serviceUrl, Duration timeout) { + this(HttpClient.newHttpClient(), defaultObjectMapper(), URI.create(serviceUrl), null, timeout); } - public ChannelFinderHttpStore(String serviceUrl, String username, String password) { - this(HttpClient.newHttpClient(), defaultObjectMapper(), URI.create(serviceUrl), basicAuth(username, password)); + public ChannelFinderHttpStore(String serviceUrl, String username, String password, Duration timeout) { + this(HttpClient.newHttpClient(), defaultObjectMapper(), URI.create(serviceUrl), basicAuth(username, password), timeout); } - ChannelFinderHttpStore(HttpClient httpClient, ObjectMapper objectMapper, URI serviceUri, String authorizationHeader) { + ChannelFinderHttpStore(HttpClient httpClient, ObjectMapper objectMapper, URI serviceUri, String authorizationHeader, + Duration timeout) { this.httpClient = Objects.requireNonNull(httpClient, "httpClient"); this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper"); this.serviceUri = normalizeBaseUri(Objects.requireNonNull(serviceUri, "serviceUri")); this.authorizationHeader = authorizationHeader; + this.timeout = Objects.requireNonNull(timeout, "timeout"); + if (timeout.isZero() || timeout.isNegative()) { + throw new IllegalArgumentException("timeout must be positive: " + timeout); + } } @Override @@ -188,10 +193,15 @@ public void updateChannels(Collection channels) { @Override public void deleteChannels(Collection channelNames) { - for (String channelName : channelNames) { - HttpRequest request = requestBuilder(channelUri(channelName)).DELETE().build(); - sendExpectingSuccess(request, "delete channel " + channelName); + if (channelNames.isEmpty()) { + return; } + + List names = new ArrayList<>(channelNames); + HttpRequest request = requestBuilder(channelsUri(), bulkUpdateTimeout(names.size())) + .method("DELETE", jsonBody(names)) + .build(); + sendExpectingSuccess(request, "delete channels " + names.size()); } @Override @@ -247,7 +257,7 @@ private HttpResponse send(HttpRequest request, String action) { } private HttpRequest.Builder requestBuilder(URI uri) { - return requestBuilder(uri, DEFAULT_TIMEOUT); + return requestBuilder(uri, timeout); } private HttpRequest.Builder requestBuilder(URI uri, Duration timeout) { @@ -261,15 +271,11 @@ private HttpRequest.Builder requestBuilder(URI uri, Duration timeout) { return builder; } - // Scale POST timeout with batch size: base + 1s per 250 channels, capped to avoid unbounded waits. - private static Duration bulkUpdateTimeout(int channelCount) { - int safeCount = Math.max(0, channelCount); - long extraSeconds = safeCount / CHANNELS_PER_TIMEOUT_STEP; - Duration scaled = DEFAULT_TIMEOUT.plus(TIMEOUT_STEP.multipliedBy(extraSeconds)); - if (scaled.compareTo(MAX_BULK_UPDATE_TIMEOUT) > 0) { - return MAX_BULK_UPDATE_TIMEOUT; - } - return scaled; + // Scale request timeout with batch size: base + 1s per 250 channels, capped to avoid unbounded waits. + private Duration bulkUpdateTimeout(int channelCount) { + long extraSeconds = Math.max(0, channelCount) / CHANNELS_PER_TIMEOUT_STEP; + Duration scaled = timeout.plus(TIMEOUT_STEP.multipliedBy(extraSeconds)); + return scaled.compareTo(MAX_BULK_UPDATE_TIMEOUT) > 0 ? MAX_BULK_UPDATE_TIMEOUT : scaled; } private HttpRequest.BodyPublisher jsonBody(Object value) { diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index b5a42bf..39d99d6 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -66,15 +66,19 @@ recceiver.cf.service-url=http://localhost:9090/ChannelFinder recceiver.cf.username=admin recceiver.cf.password=adminPass +# Base HTTP timeout in seconds for ChannelFinder calls. Bulk update/delete +# requests add 1s per 250 channels, capped at 5 minutes. +recceiver.cf.http-timeout=30 + # On connection close: true => delete IOC channels, false => mark IOC channels Inactive. -recceiver.cf.delete-on-clean=false +recceiver.cf.delete-on-clean=true # On processor init/startup for this recceiver's currently Active channels: # true => delete channels, false => mark channels Inactive. -recceiver.cf.clean-on-startup=false +recceiver.cf.clean-on-startup=true # On application shutdown: true => delete IOC channels (default), false => mark as Inactive. -recceiver.cf.delete-on-shutdown=false +recceiver.cf.delete-on-shutdown=true # Publish alias records as channels with property alias=. recceiver.cf.alias-enabled=true diff --git a/src/test/java/org/phoebus/recceiver/SimpleTest.java b/src/test/java/org/phoebus/recceiver/SimpleTest.java index 25de202..501d39a 100644 --- a/src/test/java/org/phoebus/recceiver/SimpleTest.java +++ b/src/test/java/org/phoebus/recceiver/SimpleTest.java @@ -2,18 +2,18 @@ import org.junit.jupiter.api.Test; -import javax.xml.bind.DatatypeConverter; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.HexFormat; public class SimpleTest { @Test public void encoding() { String hexString = "1234567890"; - byte[] bytes = DatatypeConverter.parseHexBinary(hexString); + byte[] bytes = HexFormat.of().parseHex(hexString); - System.out.println(DatatypeConverter.printHexBinary(bytes)); + System.out.println(HexFormat.of().formatHex(bytes)); ByteBuffer readBuffer = ByteBuffer.wrap(bytes); ByteBuffer dstBuffer = ByteBuffer.allocate(16); @@ -23,17 +23,17 @@ public void encoding() { dstBuffer.put(readBuffer.array(), 0, 2); readBuffer.position(readBuffer.position() + 2); - System.out.println(DatatypeConverter.printHexBinary(dstBuffer.array())); + System.out.println(HexFormat.of().formatHex(dstBuffer.array())); System.out.println("dst remaining: " + dstBuffer.remaining() + " dst position: " + dstBuffer.position()); System.out.println("read remaining: " + readBuffer.remaining() + " read position: " + readBuffer.position()); dstBuffer.put(readBuffer.array(), readBuffer.position(), 1); readBuffer.position(readBuffer.position() + 1); - System.out.println(DatatypeConverter.printHexBinary(dstBuffer.array())); + System.out.println(HexFormat.of().formatHex(dstBuffer.array())); dstBuffer.put(readBuffer.array(), readBuffer.position(), 1); readBuffer.position(readBuffer.position() + 1); - System.out.println(DatatypeConverter.printHexBinary(dstBuffer.array())); + System.out.println(HexFormat.of().formatHex(dstBuffer.array())); System.out.println("dst remaining: " + dstBuffer.remaining() + " dst position: " + dstBuffer.position()); System.out.println("read remaining: " + readBuffer.remaining() + " read position: " + readBuffer.position()); @@ -49,27 +49,27 @@ public void encoding() { public void compact() { String hexString = "1234567890"; - byte[] bytes = DatatypeConverter.parseHexBinary(hexString); + byte[] bytes = HexFormat.of().parseHex(hexString); ByteBuffer readBuffer = ByteBuffer.allocate(5); readBuffer.put(bytes, 0, 5); - System.out.println(DatatypeConverter.printHexBinary(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); + System.out.println(HexFormat.of().formatHex(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); readBuffer.flip(); - System.out.println(DatatypeConverter.printHexBinary(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); + System.out.println(HexFormat.of().formatHex(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); readBuffer.get(); readBuffer.get(); readBuffer.get(); readBuffer.get(); - System.out.println(DatatypeConverter.printHexBinary(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); + System.out.println(HexFormat.of().formatHex(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); readBuffer.compact(); - System.out.println(DatatypeConverter.printHexBinary(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); + System.out.println(HexFormat.of().formatHex(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); readBuffer.put((byte) 0xFF); readBuffer.put((byte) 0xFF); readBuffer.put((byte) 0xFF); readBuffer.put((byte) 0xFF); - System.out.println(DatatypeConverter.printHexBinary(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); + System.out.println(HexFormat.of().formatHex(readBuffer.array()) + " " + readBuffer.position() + " " + readBuffer.limit()); }