Make processor enabled runtime

src/main/java/org/phoebus/channelfinder/processors/ChannelProcessor.java

@@ -8,7 +8,9 @@ public interface ChannelProcessor {
 
   boolean enabled();
 
-  String processorInfo();
+  void setEnabled(boolean enabled);
+
+  ChannelProcessorInfo processorInfo();
 
   long process(List<Channel> channels) throws JsonProcessingException;
 }

src/main/java/org/phoebus/channelfinder/processors/ChannelProcessorInfo.java

@@ -0,0 +1,5 @@
+package org.phoebus.channelfinder.processors;
+
+import java.util.Map;
+
+public record ChannelProcessorInfo(String name, boolean enabled, Map<String, String> properties) {}

src/main/java/org/phoebus/channelfinder/processors/ChannelProcessorManager.java

@@ -25,6 +25,7 @@
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PutMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -75,10 +76,13 @@ public long processorCount() {
             responseCode = "200",
             description = "List of processor-info",
             content =
-                @Content(array = @ArraySchema(schema = @Schema(implementation = String.class))))
+                @Content(
+                    array =
+                        @ArraySchema(
+                            schema = @Schema(implementation = ChannelProcessorInfo.class))))
       })
-  @GetMapping("/info")
-  public List<String> processorInfo() {
+  @GetMapping("/processors")
+  public List<ChannelProcessorInfo> processorInfo() {
     return channelProcessorService.getProcessorsInfo();
   }
 
@@ -151,4 +155,17 @@ public long processChannels(
   public void processChannels(List<Channel> channels) {
     channelProcessorService.sendToProcessors(channels);
   }
+
+  @Operation(summary = "Set if the processor is enabled or not")
+  @PutMapping(
+      value = "/processor/{processorName}/enabled",
+      produces = {"application/json"},
+      consumes = {"application/json"})
+  public void setProcessorEnabled(
+      @PathVariable("processorName") String processorName,
+      @Parameter(description = "Value of enabled to set, default value: true")
+          @RequestParam(required = false, name = "enabled", defaultValue = "true")
+          Boolean enabled) {
+    channelProcessorService.setProcessorEnabled(processorName, enabled);
+  }
 }

src/main/java/org/phoebus/channelfinder/processors/ChannelProcessorService.java

@@ -2,6 +2,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.Spliterator;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -17,21 +19,35 @@ public class ChannelProcessorService {
 
   private static final Logger logger = Logger.getLogger(ChannelProcessorService.class.getName());
 
-  @Autowired private List<ChannelProcessor> channelProcessors;
+  private final List<ChannelProcessor> channelProcessors;
 
-  @Autowired private TaskExecutor taskExecutor;
+  private final TaskExecutor taskExecutor;
 
-  @Value("${processors.chunking.size:10000}")
-  private int chunkSize;
+  private final int chunkSize;
+
+  public ChannelProcessorService(
+      @Autowired List<ChannelProcessor> channelProcessors,
+      @Autowired TaskExecutor taskExecutor,
+      @Value("${processors.chunking.size:10000}") int chunkSize) {
+    this.channelProcessors = channelProcessors;
+    this.taskExecutor = taskExecutor;
+    this.chunkSize = chunkSize;
+  }
 
   long getProcessorCount() {
     return channelProcessors.size();
   }
 
-  List<String> getProcessorsInfo() {
-    return channelProcessors.stream()
-        .map(ChannelProcessor::processorInfo)
-        .collect(Collectors.toList());
+  List<ChannelProcessorInfo> getProcessorsInfo() {
+    return channelProcessors.stream().map(ChannelProcessor::processorInfo).toList();
+  }
+
+  void setProcessorEnabled(String name, boolean enabled) {
+    Optional<ChannelProcessor> processor =
+        channelProcessors.stream()
+            .filter(p -> Objects.equals(p.processorInfo().name(), name))
+            .findFirst();
+    processor.ifPresent(channelProcessor -> channelProcessor.setEnabled(enabled));
   }
 
   /**
@@ -60,7 +76,6 @@ public void sendToProcessors(List<Channel> channels) {
                         while (true) {
                           List<Channel> chunk = new ArrayList<>(chunkSize);
                           for (int i = 0; i < chunkSize && split.tryAdvance(chunk::add); i++) {}
-                          ;
                           if (chunk.isEmpty()) break;
                           channelProcessor.process(chunk);
                         }

src/main/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessor.java

@@ -15,6 +15,7 @@
 import org.phoebus.channelfinder.entity.Channel;
 import org.phoebus.channelfinder.entity.Property;
 import org.phoebus.channelfinder.processors.ChannelProcessor;
+import org.phoebus.channelfinder.processors.ChannelProcessorInfo;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
@@ -64,8 +65,15 @@ public boolean enabled() {
   }
 
   @Override
-  public String processorInfo() {
-    Map<String, String> processorProperties =
+  public void setEnabled(boolean enabled) {
+    this.aaEnabled = enabled;
+  }
+
+  @Override
+  public ChannelProcessorInfo processorInfo() {
+    return new ChannelProcessorInfo(
+        "AAChannelProcessor",
+        aaEnabled,
         Map.of(
             "archiveProperty",
             archivePropertyName,
@@ -74,8 +82,7 @@ public String processorInfo() {
             "Archivers",
             aaURLs.keySet().toString(),
             "AutoPauseOn",
-            autoPauseOptions.toString());
-    return "AAChannelProcessor: ProcessProperties " + processorProperties;
+            autoPauseOptions.toString()));
   }
 
   /**
@@ -199,7 +206,7 @@ private void addChannelChange(
       aaArchivePVS.get(archiverAlias).add(newArchiverPV);
     }
   }
-  
+
   private ArchiveAction pickArchiveAction(String archiveStatus, String pvStatus) {
     if (archiveStatus.equals("Being archived") && (pvStatus.equals(PV_STATUS_INACTIVE))) {
       return ArchiveAction.PAUSE;
@@ -209,52 +216,57 @@ private ArchiveAction pickArchiveAction(String archiveStatus, String pvStatus) {
         && !archiveStatus.equals("Paused")
         && pvStatus.equals(PV_STATUS_ACTIVE)) { // If archive status anything else
       return ArchiveAction.ARCHIVE;
-
     }
 
     return ArchiveAction.NONE;
   }
 
   private Map<ArchiveAction, List<ArchivePVOptions>> getArchiveActions(
-    Map<String, ArchivePVOptions> archivePVS, ArchiverInfo archiverInfo) {
+      Map<String, ArchivePVOptions> archivePVS, ArchiverInfo archiverInfo) {
     if (archiverInfo == null) {
       return Map.of();
     }
 
     logger.log(Level.INFO, () -> String.format("Get archiver status in archiver %s", archiverInfo));
 
     Map<ArchiveAction, List<ArchivePVOptions>> result = new EnumMap<>(ArchiveAction.class);
-    Arrays.stream(ArchiveAction.values()).forEach(archiveAction -> result.put(archiveAction, new ArrayList<>()));
+    Arrays.stream(ArchiveAction.values())
+        .forEach(archiveAction -> result.put(archiveAction, new ArrayList<>()));
     // Don't request to archive an empty list.
     if (archivePVS.isEmpty()) {
       return result;
     }
-    List<Map<String, String>> statuses = archiverClient.getStatuses(archivePVS, archiverInfo.url(), archiverInfo.alias());
+    List<Map<String, String>> statuses =
+        archiverClient.getStatuses(archivePVS, archiverInfo.url(), archiverInfo.alias());
     logger.log(Level.FINER, "Statuses {0}", statuses);
-    statuses.forEach(archivePVStatusJsonMap -> {
-      String archiveStatus = archivePVStatusJsonMap.get("status");
-      String pvName = archivePVStatusJsonMap.get("pvName");
+    statuses.forEach(
+        archivePVStatusJsonMap -> {
+          String archiveStatus = archivePVStatusJsonMap.get("status");
+          String pvName = archivePVStatusJsonMap.get("pvName");
 
-      if (archiveStatus == null || pvName == null) {
-        logger.log(Level.WARNING, "Missing status or pvName in archivePVStatusJsonMap: {0}", archivePVStatusJsonMap);
-        return;
-      }
+          if (archiveStatus == null || pvName == null) {
+            logger.log(
+                Level.WARNING,
+                "Missing status or pvName in archivePVStatusJsonMap: {0}",
+                archivePVStatusJsonMap);
+            return;
+          }
 
-      ArchivePVOptions archivePVOptions = archivePVS.get(pvName);
-      if (archivePVOptions == null) {
-        logger.log(Level.WARNING, "archivePVS does not contain pvName: {0}", pvName);
-        return;
-      }
+          ArchivePVOptions archivePVOptions = archivePVS.get(pvName);
+          if (archivePVOptions == null) {
+            logger.log(Level.WARNING, "archivePVS does not contain pvName: {0}", pvName);
+            return;
+          }
 
-      String pvStatus = archivePVOptions.getPvStatus();
-      ArchiveAction action = pickArchiveAction(archiveStatus, pvStatus);
+          String pvStatus = archivePVOptions.getPvStatus();
+          ArchiveAction action = pickArchiveAction(archiveStatus, pvStatus);
 
-      List<ArchivePVOptions> archivePVOptionsList = result.get(action);
-      archivePVOptionsList.add(archivePVOptions);
-    });
+          List<ArchivePVOptions> archivePVOptionsList = result.get(action);
+          archivePVOptionsList.add(archivePVOptions);
+        });
     return result;
   }
-  
+
   private ArchivePVOptions createArchivePV(
       List<String> policyList, Channel channel, String archiveProperty, String pvStaus) {
     ArchivePVOptions newArchiverPV = new ArchivePVOptions();

src/test/java/org/phoebus/channelfinder/processors/ChannelProcessorServiceTest.java

@@ -0,0 +1,112 @@
+package org.phoebus.channelfinder.processors;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.phoebus.channelfinder.entity.Channel;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest()
+@TestPropertySource(value = "classpath:application_test.properties")
+class ChannelProcessorServiceTest {
+
+  private ChannelProcessorService channelProcessorService;
+
+  @Autowired private DummyProcessor dummyProcessor;
+
+  @Configuration
+  static class TestConfig {
+    @Bean
+    public DummyProcessor dummyProcessor() {
+      return new DummyProcessor();
+    }
+  }
+
+  static class DummyProcessor implements ChannelProcessor {
+    private final AtomicBoolean enabled = new AtomicBoolean(false);
+    private final AtomicBoolean processed = new AtomicBoolean(false);
+
+    @Override
+    public boolean enabled() {
+      return enabled.get();
+    }
+
+    @Override
+    public void setEnabled(boolean enabled) {
+      this.enabled.set(enabled);
+    }
+
+    @Override
+    public ChannelProcessorInfo processorInfo() {
+      return new ChannelProcessorInfo("DummyProcessor", enabled.get(), Map.of());
+    }
+
+    @Override
+    public long process(List<Channel> channels) throws JsonProcessingException {
+      processed.set(true);
+      return channels.size();
+    }
+
+    public boolean hasBeenProcessed() {
+      return processed.get();
+    }
+
+    public void reset() {
+      processed.set(false);
+    }
+  }
+
+  @BeforeEach
+  void setUp() {
+    channelProcessorService =
+        new ChannelProcessorService(List.of(dummyProcessor), Runnable::run, 10);
+  }
+
+  @Test
+  void testEnableAndDisableDummyProcessor() {
+    // Initially, the dummy processor should be disabled
+    Assertions.assertFalse(
+        dummyProcessor.enabled(), "Dummy processor should be disabled initially");
+
+    // Enable the dummy processor
+    channelProcessorService.setProcessorEnabled("DummyProcessor", true);
+    Assertions.assertTrue(dummyProcessor.enabled(), "Dummy processor should be enabled");
+
+    // Disable the dummy processor
+    channelProcessorService.setProcessorEnabled("DummyProcessor", false);
+    Assertions.assertFalse(dummyProcessor.enabled(), "Dummy processor should be disabled");
+  }
+
+  @Test
+  void testDummyProcessorProcessing() {
+    // Disable the dummy processor
+    dummyProcessor.reset();
+    channelProcessorService.setProcessorEnabled("DummyProcessor", false);
+    Assertions.assertFalse(dummyProcessor.enabled(), "Dummy processor should be disabled");
+
+    // Process a channel - dummy processor should not be called
+    channelProcessorService.sendToProcessors(
+        Collections.singletonList(new Channel("test-channel")));
+    Assertions.assertFalse(
+        dummyProcessor.hasBeenProcessed(), "Dummy processor should not have been called");
+
+    // Enable the dummy processor
+    channelProcessorService.setProcessorEnabled("DummyProcessor", true);
+    Assertions.assertTrue(dummyProcessor.enabled(), "Dummy processor should be enabled");
+
+    // Process a channel - dummy processor should be called
+    channelProcessorService.sendToProcessors(
+        Collections.singletonList(new Channel("test-channel")));
+    Assertions.assertTrue(
+        dummyProcessor.hasBeenProcessed(), "Dummy processor should have been called");
+  }
+}

src/test/java/org/phoebus/channelfinder/processors/AAChannelProcessorIT.java → src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorIT.java

@@ -1,4 +1,4 @@
-package org.phoebus.channelfinder.processors;
+package org.phoebus.channelfinder.processors.aa;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
@@ -21,7 +21,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.phoebus.channelfinder.entity.Channel;
 import org.phoebus.channelfinder.entity.Property;
-import org.phoebus.channelfinder.processors.aa.AAChannelProcessor;
+import org.phoebus.channelfinder.processors.ChannelProcessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
 import org.springframework.test.context.TestPropertySource;

src/test/java/org/phoebus/channelfinder/processors/AAChannelProcessorMultiArchiverIT.java → src/test/java/org/phoebus/channelfinder/processors/aa/AAChannelProcessorMultiArchiverIT.java

@@ -1,10 +1,10 @@
-package org.phoebus.channelfinder.processors;
+package org.phoebus.channelfinder.processors.aa;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.phoebus.channelfinder.processors.AAChannelProcessorIT.activeProperty;
-import static org.phoebus.channelfinder.processors.AAChannelProcessorIT.archiveProperty;
-import static org.phoebus.channelfinder.processors.AAChannelProcessorIT.inactiveProperty;
+import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.activeProperty;
+import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.archiveProperty;
+import static org.phoebus.channelfinder.processors.aa.AAChannelProcessorIT.inactiveProperty;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -25,8 +25,6 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.phoebus.channelfinder.entity.Channel;
-import org.phoebus.channelfinder.processors.aa.AAChannelProcessor;
-import org.phoebus.channelfinder.processors.aa.ArchiveAction;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
 import org.springframework.test.context.TestPropertySource;