diff --git a/CHANGELOG.md b/CHANGELOG.md index b5d9507..db50999 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,39 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) --- -## [1.0.0] — Unreleased +## [1.0.1-RC] + +A performance and hardening release on top of the 1.0.0 rebrand: Reimport and Clone now run in parallel through JMS queues, mixed-operation batches route correctly, the listing query is parameterized, and workflow history is consistent across all three operations. The `/xapi/transfer` request format and the Java packages are unchanged from 1.0.0. + +### Performance + +- **Parallel processing via JMS queues** — Reimport and Clone no longer run in a single sequential loop. Reimport is dispatched per image session and Clone per subject onto in-process (`vm://`) JMS queues; an in-memory `BatchTransferMonitor` fans the results back in and emits a single terminal *Transfer Complete* / *Warning* event once every item — across every operation in the batch — has finished. +- **Admin-tunable consumer concurrency** — new site-admin settings (and `GET` / `POST /xapi/batch_transfer/jms_queues`) set each queue's `min–max` consumer count. Defaults: **Reimport 4–8**, **Clone 1–2**; set `min = max = 1` for a serial kill-switch. +- **Faster Clone file copy** — the archive directory copy no longer opens every file to check whether it is a catalog; only catalog/XML files are read and the rest are hard-linked directly, sharply cutting per-file I/O on large multi-scan sessions. + +> **Concurrency & storage note** — Reimport throughput is bound by prearchive **storage** and the anonymization pipeline, not CPU or plugin locking. On slow or bind-mounted storage, raising consumer concurrency is counter-productive (concurrent writers contend, and one slot can out-throughput many); tune concurrency up only on fast local/SAN prearchive storage. See `docs/ROADMAP.md`. + +### Changed + +- **Operation-aware batch routing** — a batch containing a mix of Share / Clone / Reimport items is now split by operation and each group routed to its correct path (Reimport queue, Clone queue, or the in-process sequential path for Share). Previously every item in a batch was assumed to share a single operation. +- **Unified workflow history** — the completed-action workflow recorded on the **source** item now uses one consistent past-tense label for all three operations: *"Cloned / Shared / Reimported `` to project ``"*. + +### Removed + +- **Redundant "Files Cloned" workflow** — each cloned item previously produced a second, destination-side *"Files Cloned"* workflow on top of the source-side *"Cloned …"* workflow (four workflow DB writes per item). The destination-side record is gone (two writes now); clone provenance is still captured by the source workflow plus the preserved `original-project` field, and rollback-on-failure is unchanged. + +### Fixed + +- **Reimport expand arrows** — a session whose only expandable children are out-of-scope image assessors no longer shows a dead expand arrow in Reimport mode. Share / Clone are unaffected. +- **Stale anonymization status** — the destination project's anonymization status is re-queried on each selection change instead of being read once at page load, so the *About this transfer* banner always reflects the project actually selected. + +### Security + +- **Parameterized listing query** — the subject / experiment listing SQL behind the transfer screen is now fully parameterized (`:userId`, `:project`, `:ids`) via `NamedParameterJdbcTemplate`; request-derived values are no longer concatenated into the SQL string, closing a SQL-injection vector in the project / id filters. + +--- + +## [1.0.0] ### Renamed — Batch Share → Batch Transfer diff --git a/build.gradle b/build.gradle index f35e8fb..80a1ac2 100644 --- a/build.gradle +++ b/build.gradle @@ -15,7 +15,7 @@ plugins { } group "org.nrg.xnatx.plugins" -version "1.0.0" +version "1.0.1-RC" description "XNAT Batch Transfer Plugin" sourceCompatibility = 1.8 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 876bf74..e00b701 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -1,8 +1,8 @@ # Batch Transfer Plugin — Roadmap -**Plugin Version**: 1.0.0-SNAPSHOT (formerly Batch Share Plugin 2.0.0-SNAPSHOT; renamed and versioned reset at 1.0.0) +**Plugin Version**: 1.0.1-RC (rebranded from Batch Share Plugin 2.0.0-SNAPSHOT; version line reset at 1.0.0) **Target XNAT**: 1.9.3.3 -**Last Updated**: 2026-05-12 +**Last Updated**: 2026-06-22 The Batch Transfer Plugin enables bulk data operations across XNAT projects. Users can Share, Clone, or Reimport subjects, sessions, and assessors in batch from a single interface. **Share** adds data into a destination project without copying (XNAT's standard sharing relationship); **Clone** duplicates data into the destination, producing an independent editable copy; **Reimport** re-ingests image sessions through the destination project's anonymization pipeline. @@ -10,6 +10,14 @@ The Batch Transfer Plugin enables bulk data operations across XNAT projects. Use ## Completed +### 1.0.1-RC — Performance & hardening +- **Parallel processing via JMS queues** — Reimport (per session) and Clone (per subject) dispatched onto in-process (`vm://`) JMS queues instead of a single sequential loop; in-memory `BatchTransferMonitor` fan-in emits one terminal event per batch +- **Admin-tunable queue concurrency** — site-admin settings + `GET`/`POST /xapi/batch_transfer/jms_queues`; defaults Reimport 4–8, Clone 1–2 (`min = max = 1` for serial) +- **Operation-aware batch routing** — mixed Share/Clone/Reimport batches split by operation and routed to the correct queue or the sequential path +- **Faster Clone copy** — archive copy reads only catalog/XML files and hard-links the rest (no longer opens every file) +- **Unified workflow history** — consistent past-tense source-item labels ("Cloned/Shared/Reimported `` to project ``"); removed the redundant destination-side "Files Cloned" workflow (4 → 2 workflow writes per cloned item) +- **Parameterized listing query** — subject/experiment listing SQL fully parameterized (`:userId`, `:project`, `:ids`); closes a SQL-injection vector + ### UI Modernization - Two-panel layout replacing legacy jsTree-based form (sidebar + data table) - Hierarchical data table with expand/collapse, search filtering, and bulk select @@ -39,17 +47,17 @@ The Batch Transfer Plugin enables bulk data operations across XNAT projects. Use ## Planned ### Operations & Workflow +- Expand Reimport type support to include scans-based operations - Implement UX cohort building filters for large (>10k+) datasets where the select/deselect item workflow would be tedious - Launch Advanced Search from destination project Action Menu - Include all XNAT experiment data types - Computed size estimate in the operation-detail panel — replace the static duplication warning with a real byte count for the current selection (e.g. "≈ 4.2 GB will be duplicated to *DestProject*"). Requires either a new XAPI endpoint that walks the archive, or per-row `data-size` attributes emitted by `XDATScreen_batch_transfer.java`, plus rollup logic in `batchTransfer.js`. ### Backend & API -- BatchTransferEvent refactor to limit event/DB traffic +- BatchTransferEvent refactor to limit event/DB traffic (partially addressed in 1.0.1-RC: in-memory `BatchTransferMonitor` fan-in collapses per-batch terminal events to one) - Rethink BatchTransferEvent status to reflect import process as reflected in prearch import. - "Transfer Queued (%s items)" and "Transfer Complete" do not yet reflect true Share/Clone/Reimport status nuance - REST API extension to support directional sharing workflow (transfer-to vs. transfer-from) once the inbound flow is built -- Reimport task distribution via external MQ to allow scaling across multiple XNAT nodes - Extend PROJECT_SHARING_FEATURE and PROJECT_COPYING_FEATURE controls to include a separate PROJECT_IMPORTING_FEATURE (feature-flag naming intentionally deferred from the rebrand) - Per-experiment Reimport timeout to bound importer hangs. `BatchTransferServiceImpl.runImporter` invokes `DicomZipImporter.call()` synchronously, with no upper bound — a stuck prearchive write (wedged NFS mount, deadlocked rebuild queue, slow remote disk) parks the batch-loop thread indefinitely and blocks all subsequent requests in the batch. The 5-minute heartbeat logger makes a hang observable, but recovery still requires a Tomcat restart. Possible fix: wrap `importExperiment` in `Future.get(timeoutMs)` on a separate executor; on timeout, cancel the future, emit a Failed event, and let the batch loop continue with the next request. Caveats: `cancel(true)` cannot reliably stop uninterruptible I/O so the underlying thread leaks; the prearchive has no transactional rollback so a cancelled import may leave partial state; choosing a default timeout is environment-specific (large legitimate sessions can run 30+ min). Defer until a customer incident motivates it; if added, expose the timeout as a plugin config and document explicitly that it means "we gave up waiting", not "we cleaned up". diff --git a/src/main/java/org/nrg/xnat/turbine/modules/screens/XDATScreen_batch_transfer.java b/src/main/java/org/nrg/xnat/turbine/modules/screens/XDATScreen_batch_transfer.java index 5a5b780..07f1ed4 100644 --- a/src/main/java/org/nrg/xnat/turbine/modules/screens/XDATScreen_batch_transfer.java +++ b/src/main/java/org/nrg/xnat/turbine/modules/screens/XDATScreen_batch_transfer.java @@ -22,6 +22,7 @@ import org.nrg.xft.XFTTable; import org.nrg.xft.security.UserI; import org.restlet.data.Status; +import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import java.util.*; import java.util.stream.Collectors; @@ -40,7 +41,7 @@ public class XDATScreen_batch_transfer extends SecureScreen { "LEFT JOIN xdat_user u ON map.groups_groupid_xdat_user_xdat_user_id=u.xdat_user_id\n"; private static final String READABLE_EXPERIMENTS = "FROM xdat_user_groupid gid\n" + - "LEFT JOIN xdat_usergroup grp ON gid.groupid=grp.id AND gid.groups_groupid_xdat_user_xdat_user_id={USER_ID}\n" + + "LEFT JOIN xdat_usergroup grp ON gid.groupid=grp.id AND gid.groups_groupid_xdat_user_xdat_user_id=:userId\n" + "LEFT JOIN xdat_element_access xea ON grp.xdat_usergroup_id=xea.xdat_usergroup_xdat_usergroup_id AND xea.element_name NOT IN ('xnat:projectData','xnat:subjectData')\n" + "LEFT JOIN xdat_field_mapping_set fms ON xea.xdat_element_access_id=fms.permissions_allow_set_xdat_elem_xdat_element_access_id\n" + "LEFT JOIN xdat_field_mapping xfm ON fms.xdat_field_mapping_set_id=xfm.xdat_field_mapping_set_xdat_field_mapping_set_id AND xfm.read_element=1\n" + @@ -111,11 +112,27 @@ protected void doBuildTemplate(RunData data, Context context) throws Exception { itemIds = table.convertColumnToArrayList(FIELD_ID); } - boolean byProject = (sourceProjectId != null); - String safeProj = byProject ? sourceProjectId.replaceAll("[^A-Za-z0-9_\\-]", "") : ""; - String ownedScope = byProject ? "WHERE subj.project='" + safeProj + "'\n" : ""; - String sharedScope = byProject ? "WHERE pp.project='" + safeProj + "'\n" : ""; - String finalFilter = byProject ? "" : ("WHERE " + column + " IN ('" + buildWhereClause(itemIds) + "')"); + boolean byProject = (sourceProjectId != null); + + // Parameterized query: values are bound, not concatenated. ":userId" is reused throughout; + // ":project" / ":ids" are bound per entry point. "column" in finalFilter is a SQL identifier + // (subj.id / SADS.id / IAD.id, chosen above from xsiType) and cannot be bound, so it stays + // concatenated — safe because it is code-controlled. + final MapSqlParameterSource params = + new MapSqlParameterSource("userId", ((XDATUser) user).getXdatUserId()); + String ownedScope = "", sharedScope = "", finalFilter = ""; + if (byProject) { + params.addValue("project", sourceProjectId); + ownedScope = "WHERE subj.project = :project\n"; + sharedScope = "WHERE pp.project = :project\n"; + } else { + if (itemIds == null || itemIds.isEmpty()) { + context.put("sharingMsg", "None of the requested data is available."); + return; + } + params.addValue("ids", itemIds); + finalFilter = "WHERE " + column + " IN (:ids)"; + } // Build the query to retrieve information about the items we want listed. String query = "SELECT DISTINCT ON (SUBJ.ID, SADS.ID, IAD.ID) secondary_id AS project_label,subj.id AS subject_id, subj.label AS subject_label, subj.project AS subject_project, SADS.id AS session_id, SADS.label AS session_label, SADS.project AS session_project, SADS.date AS session_expt_date, SADS.element_name as session_element, IAD.id AS assessor_id, IAD.label AS assessor_label, IAD.project AS assessor_project, IAD.date AS assess_expt_date, IAD.element_name AS assessor_element\n" + @@ -124,7 +141,7 @@ protected void doBuildTemplate(RunData data, Context context) throws Exception { "FROM xnat_subjectData subj\n" + "JOIN (\n" + READABLE_SUBJECTS + - "WHERE xdat_user_id={USER_ID} AND read_element=1 AND element_name='xnat:subjectData' AND field='xnat:subjectData/project') OWNED ON subj.project=OWNED.tag\n" + + "WHERE xdat_user_id=:userId AND read_element=1 AND element_name='xnat:subjectData' AND field='xnat:subjectData/project') OWNED ON subj.project=OWNED.tag\n" + ownedScope + "UNION\n" + "SELECT subj.id, pp.label, subj.project\n" + @@ -132,7 +149,7 @@ protected void doBuildTemplate(RunData data, Context context) throws Exception { " JOIN xnat_projectParticipant pp ON subj.id=pp.subject_id\n" + "JOIN (\n" + READABLE_SUBJECTS + - "WHERE xdat_user_id={USER_ID} AND read_element=1 AND element_name='xnat:subjectData' AND field='xnat:subjectData/sharing/share/project' ) SHARED ON pp.project=SHARED.tag\n" + + "WHERE xdat_user_id=:userId AND read_element=1 AND element_name='xnat:subjectData' AND field='xnat:subjectData/sharing/share/project' ) SHARED ON pp.project=SHARED.tag\n" + sharedScope + ") SUBJ\n" + "LEFT JOIN (\n" + @@ -169,14 +186,12 @@ protected void doBuildTemplate(RunData data, Context context) throws Exception { finalFilter + ";"; - query = query.replace("{USER_ID}", ((XDATUser) user).getXdatUserId().toString()); - - // Execute the query to get a list of experiments - XFTTable experiments = XFTTable.Execute(query, user.getDBName(), user.getUsername()); + // Execute the parameterized query to get a list of experiments. + List> rows = XDAT.getNamedParameterJdbcTemplate().queryForList(query, params); // Build ONE item set (all readable items), then compute per-project eligibility once. // Share/Clone are feature-gated; Reimport is always allowed, so the set is unfiltered. - Map items = buildItems(experiments, search); + Map items = buildItems(rows); if (items.isEmpty()) { context.put("sharingMsg", "None of the requested data is available."); return; @@ -231,13 +246,13 @@ private void addShareIdDisplayField(SchemaElement rootElement, DisplaySearch sea * @param search - DisplaySearch * @return Hashtable */ - private Map buildItems(XFTTable t, DisplaySearch search) { + private Map buildItems(List> rows) { Map items = new HashMap<>(); // For each experiment row, build the project→subject→session→assessor tree. The set is // unfiltered — per-project Share/Clone eligibility is computed once in doBuildTemplate, // and Reimport is always allowed. - for(Hashtable exp : t.rowHashs()){ + for (Map exp : rows) { String project = (String) exp.get("subject_project"); String subject_id = (String) exp.get("subject_id"); @@ -287,15 +302,6 @@ private Map buildItems(XFTTable t, DisplaySearch search) return sortItemContainersByLabel(items); } - /** - * Function builds a comma delimited string of id's to be - * inserted in the where clause of a sql query. - * @param ids - a list of ids - * @return a comma delimited string of id's - */ - private String buildWhereClause(List ids){ - return StringUtils.join(ids, "','"); - } private static Map sortItemContainersByLabel(Map items) { items.forEach((key, value) -> value.sortChildren()); diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/api/BatchTransferQueueSettingsApi.java b/src/main/java/org/nrg/xnatx/plugins/transfer/api/BatchTransferQueueSettingsApi.java new file mode 100644 index 0000000..2f05d1f --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/api/BatchTransferQueueSettingsApi.java @@ -0,0 +1,85 @@ +package org.nrg.xnatx.plugins.transfer.api; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import lombok.extern.slf4j.Slf4j; +import org.nrg.action.ClientException; +import org.nrg.action.ServerException; +import org.nrg.framework.annotations.XapiRestController; +import org.nrg.prefs.exceptions.InvalidPreferenceName; +import org.nrg.xapi.rest.AbstractXapiRestController; +import org.nrg.xapi.rest.XapiRequestMapping; +import org.nrg.xdat.security.services.RoleHolder; +import org.nrg.xdat.security.services.UserManagementServiceI; +import org.nrg.xnatx.plugins.transfer.jms.preferences.BatchTransferQueuePrefsBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.MediaType; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.ResponseBody; + +import java.util.Map; + +import static org.nrg.xdat.security.helpers.AccessLevel.Admin; + +/** + * Admin REST API backing the "Site-wide JMS Queue Settings" panel (defined in the batchTransfer + * spawner {@code site-settings.yaml}). GET returns the current Batch Transfer queue concurrency + * preferences as a map; POST updates them via {@link BatchTransferQueuePrefsBean#setBatch}. Modeled + * on container-service's {@code QueueSettingsRestApi}, but restricted to site administrators. + * + *

The preference bean extends a {@code HashMap}, so it serializes directly as the + * GET response; the spawner form's field {@code name}s match the preference keys + * ({@code reimportConcurrencyMin}/{@code reimportConcurrencyMax}). + */ +@Api("Batch Transfer JMS Queue Settings API") +@XapiRestController +@RequestMapping(value = "/batch_transfer/jms_queues") +@Slf4j +public class BatchTransferQueueSettingsApi extends AbstractXapiRestController { + + private final BatchTransferQueuePrefsBean queuePrefs; + + @Autowired + public BatchTransferQueueSettingsApi(final BatchTransferQueuePrefsBean queuePrefs, + final UserManagementServiceI userManagementService, + final RoleHolder roleHolder) { + super(userManagementService, roleHolder); + this.queuePrefs = queuePrefs; + } + + @ApiOperation(value = "Returns the Batch Transfer JMS queue concurrency settings.", response = Map.class, responseContainer = "Map") + @ApiResponses({@ApiResponse(code = 200, message = "Queue settings successfully retrieved."), + @ApiResponse(code = 401, message = "Must be authenticated to access the XNAT REST API."), + @ApiResponse(code = 403, message = "Administrator access required."), + @ApiResponse(code = 500, message = "Unexpected error")}) + @XapiRequestMapping(produces = {MediaType.APPLICATION_JSON_VALUE}, method = RequestMethod.GET, restrictTo = Admin) + @ResponseBody + public Map getQueueSettings() { + return queuePrefs; + } + + @ApiOperation(value = "Sets the Batch Transfer JMS queue concurrency settings.") + @ApiResponses({@ApiResponse(code = 200, message = "Queue settings successfully set."), + @ApiResponse(code = 400, message = "Invalid input."), + @ApiResponse(code = 401, message = "Must be authenticated to access the XNAT REST API."), + @ApiResponse(code = 403, message = "Administrator access required."), + @ApiResponse(code = 500, message = "Unexpected error")}) + @XapiRequestMapping(consumes = {MediaType.APPLICATION_FORM_URLENCODED_VALUE, MediaType.APPLICATION_JSON_VALUE}, + method = RequestMethod.POST, restrictTo = Admin) + @ResponseBody + public void setQueueSettings(@ApiParam(value = "The queue settings properties to set.", required = true) + @RequestBody final Map properties) throws ClientException, ServerException { + try { + queuePrefs.setBatch(properties); + } catch (InvalidPreferenceName e) { + throw new ClientException(e.getMessage()); + } catch (Exception e) { + throw new ServerException(e.getMessage()); + } + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/config/BatchTransferJmsConfig.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/config/BatchTransferJmsConfig.java new file mode 100644 index 0000000..06aa4da --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/config/BatchTransferJmsConfig.java @@ -0,0 +1,83 @@ +package org.nrg.xnatx.plugins.transfer.jms.config; + +import org.apache.activemq.command.ActiveMQQueue; +import org.nrg.xnatx.plugins.transfer.jms.errors.BatchTransferJmsErrorHandler; +import org.nrg.xnatx.plugins.transfer.jms.preferences.BatchTransferQueuePrefsBean; +import org.nrg.xnatx.plugins.transfer.jms.requests.CloneSubjectRequest; +import org.nrg.xnatx.plugins.transfer.jms.requests.TransferItemRequest; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jms.annotation.EnableJms; +import org.springframework.jms.config.DefaultJmsListenerContainerFactory; +import org.springframework.jms.listener.DefaultMessageListenerContainer; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +/** + * JMS wiring for parallel batch transfers, modeled on container-service's {@code ContainersConfig}. + * Reuses xnat-web's embedded broker / connection factory / {@code JmsTemplate} (defined in + * {@code MqConfig}) — this only adds the queue and the listener-container factory. + * + *

    + *
  • {@code @EnableJms} turns on {@code @JmsListener} processing for the plugin.
    • + *
  • Consumer concurrency comes from {@link BatchTransferQueuePrefsBean} (admin-tunable).
    • + *
+ */ +@Configuration +@EnableJms +public class BatchTransferJmsConfig { + + public static final String REIMPORT_LISTENER_FACTORY = "batchTransferReimportListenerFactory"; + public static final String REIMPORT_LISTENER_ID = "batchTransferReimportListener"; + public static final String CLONE_LISTENER_FACTORY = "batchTransferCloneListenerFactory"; + public static final String CLONE_LISTENER_ID = "batchTransferCloneListener"; + + /** Scale consumers down after ~5 minutes idle (300 receives * 1s timeout). */ + private static final int IDLE_RECEIVES_PER_TASK_LIMIT = 300; + private static final long RECEIVE_TIMEOUT_MS = 1000L; + + @Bean(name = TransferItemRequest.DESTINATION) + public Destination transferItemRequest() { + return new ActiveMQQueue(TransferItemRequest.DESTINATION); + } + + @Bean(name = REIMPORT_LISTENER_FACTORY) + public DefaultJmsListenerContainerFactory batchTransferReimportListenerFactory( + final BatchTransferQueuePrefsBean prefs, + @Qualifier("springConnectionFactory") final ConnectionFactory connectionFactory) { + final DefaultJmsListenerContainerFactory factory = defaultFactory(connectionFactory); + factory.setConcurrency(prefs.getReimportConcurrencyMin() + "-" + prefs.getReimportConcurrencyMax()); + return factory; + } + + @Bean(name = CloneSubjectRequest.DESTINATION) + public Destination cloneSubjectRequest() { + return new ActiveMQQueue(CloneSubjectRequest.DESTINATION); + } + + @Bean(name = CLONE_LISTENER_FACTORY) + public DefaultJmsListenerContainerFactory batchTransferCloneListenerFactory( + final BatchTransferQueuePrefsBean prefs, + @Qualifier("springConnectionFactory") final ConnectionFactory connectionFactory) { + final DefaultJmsListenerContainerFactory factory = defaultFactory(connectionFactory); + factory.setConcurrency(prefs.getCloneConcurrencyMin() + "-" + prefs.getCloneConcurrencyMax()); + return factory; + } + + private DefaultJmsListenerContainerFactory defaultFactory(final ConnectionFactory connectionFactory) { + final DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory() { + @Override + protected DefaultMessageListenerContainer createContainerInstance() { + final DefaultMessageListenerContainer container = super.createContainerInstance(); + container.setIdleReceivesPerTaskLimit(IDLE_RECEIVES_PER_TASK_LIMIT); + container.setReceiveTimeout(RECEIVE_TIMEOUT_MS); + return container; + } + }; + factory.setConnectionFactory(connectionFactory); + factory.setErrorHandler(new BatchTransferJmsErrorHandler()); + return factory; + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/errors/BatchTransferJmsErrorHandler.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/errors/BatchTransferJmsErrorHandler.java new file mode 100644 index 0000000..2008c62 --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/errors/BatchTransferJmsErrorHandler.java @@ -0,0 +1,20 @@ +package org.nrg.xnatx.plugins.transfer.jms.errors; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.util.ErrorHandler; + +/** + * Error handler wired into the Batch Transfer JMS listener-container factories. A listener that + * throws would otherwise have its exception swallowed by the container; here it is logged to + * {@code batch-transfer.log}. The listeners themselves catch-and-convert (recording the per-item + * workflow as failed and emitting a fail event) and never rethrow, so this is a backstop for + * unexpected container-level errors rather than the normal per-item failure path. + */ +@Slf4j +public class BatchTransferJmsErrorHandler implements ErrorHandler { + + @Override + public void handleError(final Throwable t) { + log.error("Batch Transfer JMS listener error", t); + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/CloneSubjectListener.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/CloneSubjectListener.java new file mode 100644 index 0000000..b6a9c4f --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/CloneSubjectListener.java @@ -0,0 +1,94 @@ +package org.nrg.xnatx.plugins.transfer.jms.listeners; + +import lombok.extern.slf4j.Slf4j; +import org.nrg.framework.services.NrgEventService; +import org.nrg.xdat.security.services.UserManagementServiceI; +import org.nrg.xft.security.UserI; +import org.nrg.xnatx.plugins.transfer.event.BatchTransferEvent; +import org.nrg.xnatx.plugins.transfer.jms.config.BatchTransferJmsConfig; +import org.nrg.xnatx.plugins.transfer.jms.requests.CloneSubjectRequest; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; +import org.nrg.xnatx.plugins.transfer.model.EventInfo; +import org.nrg.xnatx.plugins.transfer.model.TransferMode; +import org.nrg.xnatx.plugins.transfer.model.TransferRequest; +import org.nrg.xnatx.plugins.transfer.service.BatchTransferService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jms.annotation.JmsListener; +import org.springframework.stereotype.Component; + +/** + * Consumes one subject's Clone bundle per message and processes its items serially in this one + * consumer — so the destination subject/experiment is created once (by the first item that needs it) + * and reused by the rest, with no cross-thread get-or-create race or shared-source mutation. + * Parallelism is across subjects: the listener-container concurrency (from + * {@code BatchTransferQueuePrefsBean}) is how many subjects clone at once. + * + *

Never rethrows (a thrown exception would trigger JMS redelivery and reprocess the bundle). Each + * item reports its outcome to {@link BatchTransferMonitor} exactly once, so the batch's completion + * count always advances even when some items fail. + */ +@Slf4j +@Component +public class CloneSubjectListener { + + private final BatchTransferService service; + private final UserManagementServiceI userManagementService; + private final NrgEventService eventService; + private final BatchTransferMonitor monitor; + + @Autowired + public CloneSubjectListener(final BatchTransferService service, + final UserManagementServiceI userManagementService, + final NrgEventService eventService, + final BatchTransferMonitor monitor) { + this.service = service; + this.userManagementService = userManagementService; + this.eventService = eventService; + this.monitor = monitor; + } + + @JmsListener(id = BatchTransferJmsConfig.CLONE_LISTENER_ID, + containerFactory = BatchTransferJmsConfig.CLONE_LISTENER_FACTORY, + destination = CloneSubjectRequest.DESTINATION) + public void onRequest(final CloneSubjectRequest request) { + final String trackingId = request.getTrackingId(); + + final UserI user; + try { + user = userManagementService.getUser(request.getUsername()); + } catch (Exception e) { + log.error("Clone subject {}: could not load user {}", request.getSubjectId(), request.getUsername(), e); + // Report every item in the bundle so fan-in still completes. + for (final CloneSubjectRequest.CloneItem item : request.getItems()) { + emitFail(request, item.getItemId() + " failed. Cause: could not load user " + request.getUsername()); + monitor.itemDone(trackingId, true); + } + return; + } + + // One consumer owns this subject; process its items in order so the destination subject/ + // experiment is created once and reused (no get-or-create race). A failed parent fails its + // dependents item-by-item; siblings still proceed. + for (final CloneSubjectRequest.CloneItem item : request.getItems()) { + boolean failed = false; + try { + eventService.triggerEvent(BatchTransferEvent.progress(request.getRequestingUserId(), + monitor.currentPercent(trackingId), trackingId, + "Cloning " + item.getItemId() + " to " + item.getDestinationProject())); + service.processItem(new TransferRequest(item.getDestinationProject(), item.getItemId(), TransferMode.CLONE), + user, new EventInfo(trackingId, monitor.currentPercent(trackingId))); + } catch (Exception e) { + log.error("Clone {} failed", item.getItemId(), e); + failed = true; + emitFail(request, item.getItemId() + " failed. Cause: " + (e.getCause() == null ? e.getMessage() : e.getCause().getMessage())); + } finally { + monitor.itemDone(trackingId, failed); + } + } + } + + private void emitFail(final CloneSubjectRequest request, final String message) { + eventService.triggerEvent(BatchTransferEvent.fail(request.getRequestingUserId(), + monitor.currentPercent(request.getTrackingId()), request.getTrackingId(), message)); + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/TransferReimportListener.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/TransferReimportListener.java new file mode 100644 index 0000000..dc9255e --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/listeners/TransferReimportListener.java @@ -0,0 +1,87 @@ +package org.nrg.xnatx.plugins.transfer.jms.listeners; + +import lombok.extern.slf4j.Slf4j; +import org.nrg.framework.services.NrgEventService; +import org.nrg.xdat.security.services.UserManagementServiceI; +import org.nrg.xft.security.UserI; +import org.nrg.xnatx.plugins.transfer.event.BatchTransferEvent; +import org.nrg.xnatx.plugins.transfer.jms.config.BatchTransferJmsConfig; +import org.nrg.xnatx.plugins.transfer.jms.requests.TransferItemRequest; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; +import org.nrg.xnatx.plugins.transfer.model.EventInfo; +import org.nrg.xnatx.plugins.transfer.model.TransferMode; +import org.nrg.xnatx.plugins.transfer.model.TransferRequest; +import org.nrg.xnatx.plugins.transfer.service.BatchTransferService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jms.annotation.JmsListener; +import org.springframework.stereotype.Component; + +/** + * Consumes one Reimport item per message; the listener-container's concurrency (from + * {@code BatchTransferQueuePrefsBean}) is the parallelism. Reloads the requesting user and runs the + * shared per-item transfer logic via {@link BatchTransferService#processItem} — which, for Reimport, + * drives the session's own persistent workflow (created inside {@code importExperiment}) to its + * terminal state. The listener keeps no workflow of its own; it just reports each item's outcome to + * {@link BatchTransferMonitor}, whose counter fires the batch terminal event once every item is in. + * + *

Never rethrows — a thrown exception would trigger JMS redelivery and reprocess the item. + * Every path reports exactly one outcome to the monitor (in {@code finally}), so the batch's + * completion count always advances and fan-in can't stall on a handled error. + */ +@Slf4j +@Component +public class TransferReimportListener { + + private final BatchTransferService service; + private final UserManagementServiceI userManagementService; + private final NrgEventService eventService; + private final BatchTransferMonitor monitor; + + @Autowired + public TransferReimportListener(final BatchTransferService service, + final UserManagementServiceI userManagementService, + final NrgEventService eventService, + final BatchTransferMonitor monitor) { + this.service = service; + this.userManagementService = userManagementService; + this.eventService = eventService; + this.monitor = monitor; + } + + @JmsListener(id = BatchTransferJmsConfig.REIMPORT_LISTENER_ID, + containerFactory = BatchTransferJmsConfig.REIMPORT_LISTENER_FACTORY, + destination = TransferItemRequest.DESTINATION) + public void onRequest(final TransferItemRequest request) { + final String trackingId = request.getTrackingId(); + final String itemId = request.getItemId(); + + final UserI user; + try { + user = userManagementService.getUser(request.getUsername()); + } catch (Exception e) { + log.error("Reimport {}: could not load user {}", itemId, request.getUsername(), e); + emitFail(request, "Could not load user " + request.getUsername()); + monitor.itemDone(trackingId, true); + return; + } + + boolean failed = false; + try { + eventService.triggerEvent(BatchTransferEvent.progress(request.getRequestingUserId(), + monitor.currentPercent(trackingId), trackingId, "Reimporting " + itemId + " to " + request.getDestinationProject())); + service.processItem(new TransferRequest(request.getDestinationProject(), itemId, TransferMode.REIMPORT), + user, new EventInfo(trackingId, monitor.currentPercent(trackingId))); + } catch (Exception e) { + log.error("Reimport {} failed", itemId, e); + failed = true; + emitFail(request, itemId + " failed. Cause: " + (e.getCause() == null ? e.getMessage() : e.getCause().getMessage())); + } finally { + monitor.itemDone(trackingId, failed); + } + } + + private void emitFail(final TransferItemRequest request, final String message) { + eventService.triggerEvent(BatchTransferEvent.fail(request.getRequestingUserId(), + monitor.currentPercent(request.getTrackingId()), request.getTrackingId(), message)); + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/preferences/BatchTransferQueuePrefsBean.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/preferences/BatchTransferQueuePrefsBean.java new file mode 100644 index 0000000..8082a26 --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/preferences/BatchTransferQueuePrefsBean.java @@ -0,0 +1,84 @@ +package org.nrg.xnatx.plugins.transfer.jms.preferences; + +import lombok.extern.slf4j.Slf4j; +import org.nrg.framework.configuration.ConfigPaths; +import org.nrg.framework.services.NrgEventServiceI; +import org.nrg.framework.utilities.OrderedProperties; +import org.nrg.prefs.annotations.NrgPreference; +import org.nrg.prefs.annotations.NrgPreferenceBean; +import org.nrg.prefs.exceptions.InvalidPreferenceName; +import org.nrg.prefs.services.NrgPreferenceService; +import org.nrg.xdat.preferences.EventTriggeringAbstractPreferenceBean; +import org.springframework.beans.factory.annotation.Autowired; + +/** + * Admin-tunable concurrency for the Batch Transfer JMS queues. Surfaces in the site-admin + * preferences UI (the {@link NrgPreferenceBean} annotation is meta-{@code @Component}, so the + * plugin's component scan registers it). The JMS listener-container factories in + * {@code BatchTransferJmsConfig} read these to set each queue's {@code min-max} consumer concurrency. + * + *

Defaults are deliberately conservative: each Reimport consumer spawns a streaming-zip producer + * thread plus a {@code DicomZipImporter}, and the prearchive Rebuild each one queues is itself + * parallel (up to {@code prearchiveOperationJmsListenerMaxConcurrency}, default 40/node) — so plugin + * concurrency stacks on the downstream rebuild stage. Set min == max == 1 for a serial kill-switch. + */ +@Slf4j +@NrgPreferenceBean(toolId = "batch-transfer-jms-queue", + toolName = "Batch Transfer JMS Queue Preferences", + description = "Consumer concurrency for the Batch Transfer Reimport and Clone JMS queues") +public class BatchTransferQueuePrefsBean extends EventTriggeringAbstractPreferenceBean { + + public static final String REIMPORT_CONCURRENCY_MIN_DFLT = "4"; + public static final String REIMPORT_CONCURRENCY_MAX_DFLT = "8"; + public static final String CLONE_CONCURRENCY_MIN_DFLT = "1"; + public static final String CLONE_CONCURRENCY_MAX_DFLT = "2"; + + private static final String REIMPORT_MIN = "reimportConcurrencyMin"; + private static final String REIMPORT_MAX = "reimportConcurrencyMax"; + private static final String CLONE_MIN = "cloneConcurrencyMin"; + private static final String CLONE_MAX = "cloneConcurrencyMax"; + + @Autowired + public BatchTransferQueuePrefsBean(final NrgPreferenceService preferenceService, + final NrgEventServiceI eventService, + final ConfigPaths configPaths, + final OrderedProperties initPrefs) { + super(preferenceService, eventService, configPaths, initPrefs); + } + + @NrgPreference(defaultValue = REIMPORT_CONCURRENCY_MIN_DFLT) + public Integer getReimportConcurrencyMin() { + return getIntegerValue(REIMPORT_MIN); + } + + public void setReimportConcurrencyMin(final Integer value) throws InvalidPreferenceName { + setIntegerValue(value, REIMPORT_MIN); + } + + @NrgPreference(defaultValue = REIMPORT_CONCURRENCY_MAX_DFLT) + public Integer getReimportConcurrencyMax() { + return getIntegerValue(REIMPORT_MAX); + } + + public void setReimportConcurrencyMax(final Integer value) throws InvalidPreferenceName { + setIntegerValue(value, REIMPORT_MAX); + } + + @NrgPreference(defaultValue = CLONE_CONCURRENCY_MIN_DFLT) + public Integer getCloneConcurrencyMin() { + return getIntegerValue(CLONE_MIN); + } + + public void setCloneConcurrencyMin(final Integer value) throws InvalidPreferenceName { + setIntegerValue(value, CLONE_MIN); + } + + @NrgPreference(defaultValue = CLONE_CONCURRENCY_MAX_DFLT) + public Integer getCloneConcurrencyMax() { + return getIntegerValue(CLONE_MAX); + } + + public void setCloneConcurrencyMax(final Integer value) throws InvalidPreferenceName { + setIntegerValue(value, CLONE_MAX); + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/CloneSubjectRequest.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/CloneSubjectRequest.java new file mode 100644 index 0000000..5ad68d4 --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/CloneSubjectRequest.java @@ -0,0 +1,40 @@ +package org.nrg.xnatx.plugins.transfer.jms.requests; + +import lombok.Data; + +import java.io.Serializable; +import java.util.List; + +/** + * One subject's worth of Clone work. The producer ({@code BatchTransferServiceImpl.enqueueClone}) + * groups a Clone batch's selected items by subject and sends one of these per subject to the + * {@value #DESTINATION} queue. {@code CloneSubjectListener} processes a subject's items serially in a + * single consumer, so the destination subject/experiment is created once and reused — avoiding the + * get-or-create race and shared-source mutation that make concurrent Clone of items under one subject + * unsafe. Parallelism is across subjects (different subjects → different bundles → different consumers). + * + *

Plain {@link Serializable} POJO (travels over JMS): no {@code UserI} (reloaded from {@link #username}). + * Bean name {@value #DESTINATION} equals {@code uncapitalize("CloneSubjectRequest")} for + * {@code XDAT.sendJmsRequest} routing. + */ +@Data +public class CloneSubjectRequest implements Serializable { + + public static final String DESTINATION = "cloneSubjectRequest"; + + private static final long serialVersionUID = 1L; + + private final String trackingId; + private final String subjectId; + private final List items; + private final String username; + private final Integer requestingUserId; + + /** One selected item under the subject, paired with its destination project. */ + @Data + public static class CloneItem implements Serializable { + private static final long serialVersionUID = 1L; + private final String itemId; + private final String destinationProject; + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/TransferItemRequest.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/TransferItemRequest.java new file mode 100644 index 0000000..da21981 --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/requests/TransferItemRequest.java @@ -0,0 +1,32 @@ +package org.nrg.xnatx.plugins.transfer.jms.requests; + +import lombok.Data; + +import java.io.Serializable; + +/** + * One queued Reimport item. The producer ({@code BatchTransferServiceImpl.enqueueReimport}) sends one + * of these per selected image session to the {@value #DESTINATION} queue; {@code TransferReimportListener} + * consumes them concurrently (the listener-container concurrency is the parallelism). + * + *

Plain {@link Serializable} POJO — it travels over JMS, so it carries no {@code UserI} (the + * consumer reloads the user from {@link #username}) and no workflow handle: Reimport's session + * workflow is created and completed inside {@code importExperiment}, and batch completion is tracked + * by {@code BatchTransferMonitor}'s in-memory counter rather than per-message workflow ids. + * + *

The bean name {@value #DESTINATION} equals {@code uncapitalize("TransferItemRequest")}, so + * {@code XDAT.sendJmsRequest(request)} resolves this queue by the request class's simple name. + */ +@Data +public class TransferItemRequest implements Serializable { + + public static final String DESTINATION = "transferItemRequest"; + + private static final long serialVersionUID = 1L; + + private final String trackingId; + private final String itemId; + private final String destinationProject; + private final String username; + private final Integer requestingUserId; +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/jms/tasks/BatchTransferMonitor.java b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/tasks/BatchTransferMonitor.java new file mode 100644 index 0000000..5e9e1e1 --- /dev/null +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/jms/tasks/BatchTransferMonitor.java @@ -0,0 +1,98 @@ +package org.nrg.xnatx.plugins.transfer.jms.tasks; + +import lombok.extern.slf4j.Slf4j; +import org.nrg.framework.services.NrgEventService; +import org.nrg.xnatx.plugins.transfer.event.BatchTransferEvent; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * In-memory fan-in for the JMS-parallelized Reimport batches. The producer {@link #register}s a batch + * with its item count; each consumer reports its item's outcome via {@link #itemDone}; the consumer + * that reports the final item emits the single terminal {@code Completed}/{@code Warning} event (the + * one the old sequential loop emitted inline). + * + *

An in-memory counter is used rather than counting per-item workflows because Reimport's session + * workflow is created mid-processing (inside {@code importExperiment}), so an item that fails before + * that point would have no workflow to count — whereas {@code itemDone}, called from the consumer's + * {@code finally}, always advances. Per-item workflows still exist (created by {@code importExperiment}) + * for audit; they just aren't the completion signal. + * + *

State is per-JVM. With XNAT's in-process ({@code vm://}) broker, a batch's producer, consumers, + * and this registry all live in one JVM, so the count is coherent. A mid-batch server restart loses + * the registry, so that batch's terminal event won't auto-fire (its session workflows still complete) — + * the same durability the previous {@code ExecutorService} flow had. + */ +@Slf4j +@Component +public class BatchTransferMonitor { + + private final NrgEventService eventService; + + private final Map active = new ConcurrentHashMap<>(); + + @Autowired + public BatchTransferMonitor(final NrgEventService eventService) { + this.eventService = eventService; + } + + /** Registered by the producer before enqueuing, with the number of items it will send. */ + public void register(final String trackingId, final Integer userId, final int total) { + active.put(trackingId, new BatchInfo(userId, total)); + } + + /** Best-effort progress percent for per-item InProgress events; clamped 0..99 (100 is terminal). */ + public int currentPercent(final String trackingId) { + final BatchInfo info = active.get(trackingId); + if (info == null || info.total <= 0) { + return 0; + } + final int pct = (int) Math.floor((info.completed.get() * 100.0) / info.total); + return Math.max(0, Math.min(99, pct)); + } + + /** + * Reports one item's outcome. The consumer that brings the completed count up to the batch total + * emits the terminal event exactly once and clears the batch. Must be called once per item on + * every path (success or handled failure), or fan-in stalls. + */ + public void itemDone(final String trackingId, final boolean failed) { + final BatchInfo info = active.get(trackingId); + if (info == null) { + return; + } + if (failed) { + info.failed.incrementAndGet(); + } + if (info.completed.incrementAndGet() != info.total) { + return; // not the last item + } + // Last item in — emit this batch's single terminal event. incrementAndGet() == total is seen + // by exactly one thread, and every failed-increment happens-before it, so the count is exact. + final int failures = info.failed.get(); + if (failures > 0) { + eventService.triggerEvent(BatchTransferEvent.warn(info.userId, 100, trackingId, + String.format("Transfer Complete with %d %s. Please review this log carefully.", + failures, failures == 1 ? "warning" : "warnings"))); + } else { + eventService.triggerEvent(BatchTransferEvent.complete(info.userId, trackingId, "Transfer Complete")); + } + active.remove(trackingId); + } + + private static final class BatchInfo { + private final Integer userId; + private final int total; + private final AtomicInteger completed = new AtomicInteger(0); + private final AtomicInteger failed = new AtomicInteger(0); + + private BatchInfo(final Integer userId, final int total) { + this.userId = userId; + this.total = total; + } + } +} diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/model/TransferMode.java b/src/main/java/org/nrg/xnatx/plugins/transfer/model/TransferMode.java index 02cc29b..23cb6d1 100644 --- a/src/main/java/org/nrg/xnatx/plugins/transfer/model/TransferMode.java +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/model/TransferMode.java @@ -3,16 +3,18 @@ import com.fasterxml.jackson.annotation.JsonValue; public enum TransferMode { - CLONE("Clone", "Cloning"), - SHARE("Share", "Sharing"), - REIMPORT("Reimport", "Reimporting"); + CLONE("Clone", "Cloning", "Cloned"), + SHARE("Share", "Sharing", "Shared"), + REIMPORT("Reimport", "Reimporting", "Reimported"); private final String value; private final String action; + private final String pastAction; - TransferMode(String value, String action) { + TransferMode(String value, String action, String pastAction) { this.value = value; this.action = action; + this.pastAction = pastAction; } @JsonValue @@ -24,6 +26,11 @@ public String getAction() { return action; } + /** Past-tense form ("Cloned"/"Shared"/"Reimported") for labeling completed-action workflows. */ + public String getPastAction() { + return pastAction; + } + public String toString() { return getValue(); } diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/plugin/BatchTransferPlugin.java b/src/main/java/org/nrg/xnatx/plugins/transfer/plugin/BatchTransferPlugin.java index da32f82..bb2d0ee 100644 --- a/src/main/java/org/nrg/xnatx/plugins/transfer/plugin/BatchTransferPlugin.java +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/plugin/BatchTransferPlugin.java @@ -6,6 +6,7 @@ "org.nrg.xnatx.plugins.transfer.api", "org.nrg.xnatx.plugins.transfer.service", "org.nrg.xnatx.plugins.transfer.event", + "org.nrg.xnatx.plugins.transfer.jms", }) @XnatPlugin(value = "batchTransferPlugin", name = "Batch Transfer Plugin", diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/service/BatchTransferService.java b/src/main/java/org/nrg/xnatx/plugins/transfer/service/BatchTransferService.java index a5c7c6a..68ff76c 100644 --- a/src/main/java/org/nrg/xnatx/plugins/transfer/service/BatchTransferService.java +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/service/BatchTransferService.java @@ -1,8 +1,17 @@ package org.nrg.xnatx.plugins.transfer.service; import org.nrg.xnatx.plugins.transfer.model.BatchTransfer; +import org.nrg.xnatx.plugins.transfer.model.EventInfo; +import org.nrg.xnatx.plugins.transfer.model.TransferRequest; import org.nrg.xft.security.UserI; public interface BatchTransferService { void submitTransferRequest(BatchTransfer batchTransferRequest, UserI user); + + /** + * Validates, loads, permission-checks, and performs a single transfer item by mode. Shared by + * the sequential Share/Clone path and the JMS consumers (the parallel Reimport path), so it is + * the single source of truth for per-item transfer logic. Throws on any failure. + */ + void processItem(TransferRequest request, UserI user, EventInfo eventInfo) throws Exception; } diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImpl.java b/src/main/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImpl.java index 020c492..fb27b72 100644 --- a/src/main/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImpl.java +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImpl.java @@ -36,6 +36,10 @@ import org.nrg.xnatx.plugins.transfer.model.EventInfo; import org.nrg.xnatx.plugins.transfer.model.Fields; import org.nrg.xnatx.plugins.transfer.model.TransferRequest; +import org.nrg.xnatx.plugins.transfer.jms.requests.TransferItemRequest; +import org.nrg.xnatx.plugins.transfer.jms.requests.CloneSubjectRequest; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; +import org.nrg.xdat.XDAT; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -47,15 +51,17 @@ import java.io.PipedInputStream; import java.io.PipedOutputStream; import java.io.UncheckedIOException; +import java.nio.file.FileVisitOption; +import java.nio.file.FileVisitResult; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Stream; -import java.util.stream.StreamSupport; import java.util.zip.Deflater; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; @@ -67,16 +73,19 @@ public class BatchTransferServiceImpl implements BatchTransferService { private final ExecutorService executorService; private final AnonUtils anonUtils; private final DicomObjectIdentifier identifier; + private final BatchTransferMonitor monitor; @Autowired public BatchTransferServiceImpl(final NrgEventService eventService, final ExecutorService executorService, final AnonUtils anonUtils, - final ContextService contextService) { + final ContextService contextService, + final BatchTransferMonitor monitor) { this.eventService = eventService; this.executorService = executorService; this.anonUtils = anonUtils; this.identifier = contextService.getBean("dicomObjectIdentifier", DicomObjectIdentifier.class); + this.monitor = monitor; } public void submitTransferRequest(BatchTransfer batchTransferRequest, UserI user) { @@ -94,93 +103,220 @@ public void submitTransferRequest(BatchTransfer batchTransferRequest, UserI user } private void batchTransfer(BatchTransfer batchTransferRequest, UserI user) { - final List failures = new ArrayList<>(); + final List requests = batchTransferRequest.getRequests(); + if (requests == null || requests.isEmpty()) { + return; + } final String trackingId = batchTransferRequest.getTrackingId(); - final int numRequests = batchTransferRequest.getRequests().size(); - int count = 0; - for (TransferRequest request : batchTransferRequest.getRequests()) { - count++; - String modeAction = request.getMode() != null ? request.getMode().getAction() : "UNKNOWN MODE"; - final int progress = Math.round(((float) count / numRequests) * 100) - 1; + // A batch usually carries a single operation, but the UI can mix them. Split by operation in + // one pass so each item takes the right path: Reimport fans out to a JMS queue per item; Clone + // fans out to a JMS queue per subject (one consumer owns each subject); Share (and any + // null/unknown mode) is processed in-process on the sequential path. + final List reimportItems = new ArrayList<>(); + final List cloneItems = new ArrayList<>(); + final List sequentialItems = new ArrayList<>(); + for (final TransferRequest request : requests) { + final TransferMode mode = request.getMode(); + if (mode == TransferMode.REIMPORT) { + reimportItems.add(request); + } else if (mode == TransferMode.CLONE) { + cloneItems.add(request); + } else { + sequentialItems.add(request); + } + } + + // Register the whole batch once; every item — whether queued (Reimport/Clone) or processed + // in-process (Share) — reports to the monitor, which emits the single terminal event after the + // last one finishes. This keeps one completion signal even when operations are mixed. + monitor.register(trackingId, user.getID(), requests.size()); + if (!reimportItems.isEmpty()) { + enqueueReimport(trackingId, reimportItems, user); + } + if (!cloneItems.isEmpty()) { + enqueueClone(trackingId, cloneItems, user); + } + if (!sequentialItems.isEmpty()) { + runSequential(trackingId, sequentialItems, user); + } + } + + /** + * Producer for the parallel Reimport path. Enqueues one {@link TransferItemRequest} per item; + * {@code TransferReimportListener} consumes them concurrently and each reports completion to the + * {@link BatchTransferMonitor}. The batch is registered with the monitor once by the caller + * ({@link #batchTransfer}), so this only enqueues; it runs on the executor thread, off the HTTP + * path. Each session's own workflow is created (and completed/failed) inside {@code importExperiment}. + */ + private void enqueueReimport(final String trackingId, final List requests, final UserI user) { + for (final TransferRequest request : requests) { try { - final String message = String.format("%s %s into project: %s", modeAction, request.getId(), request.getDestinationProject()); - eventService.triggerEvent(BatchTransferEvent.progress(user.getID(), progress, batchTransferRequest.getTrackingId(), message)); - - validateRequest(request); - final EventInfo eventInfo = new EventInfo(trackingId, progress); - final XnatProjectdata destinationProjectData = XnatUtils.getProject(request.getDestinationProject(), user); - final ArchivableItem item = XnatUtils.getArchivableItem(request.getId(), null); - - // We need to retrieve the item outside of the user's context so we can access archive path, etc. - // This means we need to carefully check our permissions on it. - if (!Permissions.canRead(user, item)) { - throw new Exception("You do not have permission to read " + item.getId()); - } + XDAT.sendJmsRequest(new TransferItemRequest(trackingId, request.getId(), + request.getDestinationProject(), user.getUsername(), user.getID())); + } catch (Exception e) { + // Couldn't enqueue this item — report it as a failed completion so the batch still finishes. + log.error("Failed to queue reimport for {}", request.getId(), e); + eventService.triggerEvent(BatchTransferEvent.fail(user.getID(), 0, trackingId, + request.getId() + " could not be queued. Cause: " + e.getMessage())); + monitor.itemDone(trackingId, true); + } + } + } - final String sourceProject = item.getProject(); - if (sourceProject.equals(request.getDestinationProject())) { - throw new Exception("Destination project cannot be the same as the source project."); - } + /** + * Producer for the parallel Clone path. Groups the items by their subject and enqueues one + * {@link CloneSubjectRequest} per subject, so {@code CloneSubjectListener} processes each subject's + * items serially in a single consumer (creating the destination subject/experiment once, with no + * get-or-create race). Parallelism is across subjects. The batch is registered with the monitor once + * by the caller ({@link #batchTransfer}); each item reports completion via its consumer. + */ + private void enqueueClone(final String trackingId, final List requests, final UserI user) { + final Map> bySubject = new LinkedHashMap<>(); + for (final TransferRequest request : requests) { + try { + final String subjectId = resolveSubjectId(request.getId()); + bySubject.computeIfAbsent(subjectId, k -> new ArrayList<>()) + .add(new CloneSubjectRequest.CloneItem(request.getId(), request.getDestinationProject())); + } catch (Exception e) { + // Couldn't determine the subject — report this item failed so the batch still finishes. + log.error("Failed to group clone item {}", request.getId(), e); + eventService.triggerEvent(BatchTransferEvent.fail(user.getID(), 0, trackingId, + request.getId() + " could not be queued. Cause: " + e.getMessage())); + monitor.itemDone(trackingId, true); + } + } - if (request.getMode() == TransferMode.SHARE) { - if (!Features.checkRestrictedFeature(user, item.getProject(), Features.PROJECT_SHARING_FEATURE)) { - throw new Exception("You do not have permission to share this data."); - } - } else { - if (!Features.checkRestrictedFeature(user, item.getProject(), Fields.PROJECT_COPYING_FEATURE)) { - throw new Exception("You do not have permission to clone this data."); - } + for (final Map.Entry> entry : bySubject.entrySet()) { + try { + XDAT.sendJmsRequest(new CloneSubjectRequest(trackingId, entry.getKey(), entry.getValue(), + user.getUsername(), user.getID())); + } catch (Exception e) { + // Couldn't enqueue this subject's bundle — report all of its items failed. + log.error("Failed to queue clone bundle for subject {}", entry.getKey(), e); + for (final CloneSubjectRequest.CloneItem item : entry.getValue()) { + eventService.triggerEvent(BatchTransferEvent.fail(user.getID(), 0, trackingId, + item.getItemId() + " could not be queued. Cause: " + e.getMessage())); + monitor.itemDone(trackingId, true); } + } + } + } - if (!Permissions.canCreate(user, item.getXSIType() + "/project", destinationProjectData.getId())) { - throw new Exception("You do not have permission to create " + item.getXSIType() + " in " + - destinationProjectData.getId()); - } + /** Resolves an item's subject id for Clone grouping: subject → itself; experiment → its subject; assessor → its session's subject. */ + private String resolveSubjectId(final String itemId) throws Exception { + final ArchivableItem item = XnatUtils.getArchivableItem(itemId, null); + final String subjectId; + if (item instanceof XnatSubjectdata) { + subjectId = item.getId(); + } else if (item instanceof XnatImageassessordata) { + subjectId = (String) ((XnatImageassessordata) item).getImageSessionData().getProperty("subject_ID"); + } else if (item instanceof XnatExperimentdata) { + subjectId = (String) ((XnatExperimentdata) item).getProperty("subject_ID"); + } else { + subjectId = null; + } + if (StringUtils.isBlank(subjectId)) { + throw new Exception("Could not determine the subject for " + itemId); + } + return subjectId; + } - if (item instanceof XnatSubjectdata) { - final XnatSubjectdata sourceSubject = (XnatSubjectdata) item; - final XnatSubjectdata existingSubject = XnatSubjectdata.GetSubjectByProjectIdentifier(destinationProjectData.getId(), - sourceSubject.getLabel(), null, false); - - if (request.getMode().equals(TransferMode.SHARE)) { - shareSubject(sourceSubject, existingSubject, destinationProjectData, user, eventInfo); - } else if (request.getMode().equals(TransferMode.CLONE)) { - getOrCopySubject(sourceSubject, existingSubject, destinationProjectData, user, eventInfo); - } else if (request.getMode().equals(TransferMode.REIMPORT)) { - log.warn("Reimport operation is not supported for subjects."); - } else { - throw new Exception(String.format("Unsupported mode %s", request.getMode())); - } - } else if (item instanceof XnatExperimentdata) { - final XnatExperimentdata sourceExperiment = (XnatExperimentdata) item; - final XnatExperimentdata existingExperiment = XnatExperimentdata.GetExptByProjectIdentifier(destinationProjectData.getId(), - sourceExperiment.getLabel(), null, false); - - if (request.getMode().equals(TransferMode.SHARE)) { - shareExperiment(sourceExperiment, existingExperiment, destinationProjectData, user, eventInfo); - } else if (request.getMode().equals(TransferMode.CLONE)) { - getOrCopyExperimentOrAssessor(sourceExperiment, existingExperiment, destinationProjectData, user, eventInfo); - } else if (request.getMode().equals(TransferMode.REIMPORT)) { - importExperiment(sourceExperiment, destinationProjectData, user, eventInfo); - } else { - throw new Exception(String.format("Unsupported mode %s", request.getMode())); - } - } else { - throw new Exception(String.format("Unsupported xsiType %s", item.getXSIType())); - } + /** + * In-process path for Share (and any null/unknown-mode) items. Calls {@link #processItem} per item + * and reports each one to the {@link BatchTransferMonitor}; the monitor (not this method) emits the + * batch's single terminal event once every item across all operations has finished — so this path + * composes with the parallel Reimport/Clone queues in a mixed batch. + */ + private void runSequential(final String trackingId, final List requests, final UserI user) { + for (final TransferRequest request : requests) { + final String modeAction = request.getMode() != null ? request.getMode().getAction() : "UNKNOWN MODE"; + final int progress = monitor.currentPercent(trackingId); + boolean failed = false; + try { + eventService.triggerEvent(BatchTransferEvent.progress(user.getID(), progress, trackingId, + String.format("%s %s into project: %s", modeAction, request.getId(), request.getDestinationProject()))); + processItem(request, user, new EventInfo(trackingId, progress)); } catch (Exception e) { + failed = true; log.debug(e.getMessage(), e); final String err = String.format("%s %s failed. Cause: %s", request.getId(), request.getMode(), e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); - failures.add(err); eventService.triggerEvent(BatchTransferEvent.fail(user.getID(), progress, trackingId, err)); } + // Report completion; the monitor emits the batch's terminal event when the last item finishes. + monitor.itemDone(trackingId, failed); + } + } + + /** + * Validates, loads, permission-checks, and dispatches a single transfer item by mode. Shared by + * the sequential path ({@link #runSequential}) and the JMS consumers, so it is the single source + * of truth for per-item transfer logic. Throws on any failure; the caller records it (a fail + * event in the sequential path, a failed workflow + fail event in the consumer). + */ + @Override + public void processItem(final TransferRequest request, final UserI user, final EventInfo eventInfo) throws Exception { + validateRequest(request); + final XnatProjectdata destinationProjectData = XnatUtils.getProject(request.getDestinationProject(), user); + final ArchivableItem item = XnatUtils.getArchivableItem(request.getId(), null); + + // We retrieve the item outside of the user's context so we can access archive path, etc., + // so we carefully check our permissions on it here. + if (!Permissions.canRead(user, item)) { + throw new Exception("You do not have permission to read " + item.getId()); + } + + final String sourceProject = item.getProject(); + if (sourceProject.equals(request.getDestinationProject())) { + throw new Exception("Destination project cannot be the same as the source project."); + } + + if (request.getMode() == TransferMode.SHARE) { + if (!Features.checkRestrictedFeature(user, item.getProject(), Features.PROJECT_SHARING_FEATURE)) { + throw new Exception("You do not have permission to share this data."); + } + } else { + if (!Features.checkRestrictedFeature(user, item.getProject(), Fields.PROJECT_COPYING_FEATURE)) { + throw new Exception("You do not have permission to clone this data."); + } } - if (!failures.isEmpty()) { - eventService.triggerEvent(BatchTransferEvent.warn(user.getID(), 100, batchTransferRequest.getTrackingId(), String.format("Transfer Complete with %s %s. Please review this log carefully.", failures.size(), failures.size() == 1 ? "warning" : "warnings"))); + if (!Permissions.canCreate(user, item.getXSIType() + "/project", destinationProjectData.getId())) { + throw new Exception("You do not have permission to create " + item.getXSIType() + " in " + + destinationProjectData.getId()); + } + + if (item instanceof XnatSubjectdata) { + final XnatSubjectdata sourceSubject = (XnatSubjectdata) item; + final XnatSubjectdata existingSubject = XnatSubjectdata.GetSubjectByProjectIdentifier(destinationProjectData.getId(), + sourceSubject.getLabel(), null, false); + + if (request.getMode().equals(TransferMode.SHARE)) { + shareSubject(sourceSubject, existingSubject, destinationProjectData, user, eventInfo); + } else if (request.getMode().equals(TransferMode.CLONE)) { + getOrCopySubject(sourceSubject, existingSubject, destinationProjectData, user, eventInfo); + } else if (request.getMode().equals(TransferMode.REIMPORT)) { + log.warn("Reimport operation is not supported for subjects."); + } else { + throw new Exception(String.format("Unsupported mode %s", request.getMode())); + } + } else if (item instanceof XnatExperimentdata) { + final XnatExperimentdata sourceExperiment = (XnatExperimentdata) item; + // Reimport does not use the existing-experiment lookup; skip that DB query for it. + final XnatExperimentdata existingExperiment = (request.getMode() == TransferMode.REIMPORT) ? null + : XnatExperimentdata.GetExptByProjectIdentifier(destinationProjectData.getId(), sourceExperiment.getLabel(), null, false); + + if (request.getMode().equals(TransferMode.SHARE)) { + shareExperiment(sourceExperiment, existingExperiment, destinationProjectData, user, eventInfo); + } else if (request.getMode().equals(TransferMode.CLONE)) { + getOrCopyExperimentOrAssessor(sourceExperiment, existingExperiment, destinationProjectData, user, eventInfo); + } else if (request.getMode().equals(TransferMode.REIMPORT)) { + importExperiment(sourceExperiment, destinationProjectData, user, eventInfo); + } else { + throw new Exception(String.format("Unsupported mode %s", request.getMode())); + } } else { - eventService.triggerEvent(BatchTransferEvent.complete(user.getID(), batchTransferRequest.getTrackingId(), "Transfer Complete")); + throw new Exception(String.format("Unsupported xsiType %s", item.getXSIType())); } } @@ -212,7 +348,8 @@ private void importExperiment(XnatExperimentdata sourceExperiment, XnatProjectda final StreamingZipFileWriter wrapper = new StreamingZipFileWriter(sourcePath, sourceExperiment.getId() + ".zip"); final AtomicReference> uriRef = new AtomicReference<>(); - XnatUtils.doActionWithWorkflow(user, sourceExperiment, "Streaming experiment for import.", () -> { + XnatUtils.doActionWithWorkflow(user, sourceExperiment, + sourceWorkflowLabel(TransferMode.REIMPORT, sourceExperiment.getLabel(), destinationProjectData.getId()), () -> { final List result = runImporter(user, wrapper, params); if (result == null || result.isEmpty()) { throw new Exception("No DICOM files found in source experiment " + sourceExperiment.getId() + "; nothing to reimport."); @@ -360,16 +497,35 @@ public InputStream getInputStream() throws IOException { private void produce() { final ZipOutputStream zos = new ZipOutputStream(new BufferedOutputStream(pipedOut, 65536)); - zos.setLevel(Deflater.BEST_SPEED); + zos.setLevel(Deflater.NO_COMPRESSION); final byte[] buffer = new byte[65536]; - try (Stream paths = Files.walk(sourceDir)) { - // writeEntry returns false once the consumer closes the pipe; allMatch then stops - // streaming. Real source-read failures throw and are caught (and recorded) below. - paths.filter(Files::isRegularFile) - .filter(p -> StreamSupport.stream(p.spliterator(), false) - .anyMatch(part -> part.toString().equals("DICOM"))) - .filter(p -> !p.getFileName().toString().toLowerCase().endsWith("catalog.xml")) - .allMatch(p -> writeEntry(zos, p, buffer)); + try { + // Prune everything except the SCANS directory directly under the source root (RESOURCES, + // the session XML, etc.) and follow symlinks. The walk supplies each file's attributes, + // so there is no second stat per file. + Files.walkFileTree(sourceDir, EnumSet.of(FileVisitOption.FOLLOW_LINKS), Integer.MAX_VALUE, + new SimpleFileVisitor() { + @Override + public FileVisitResult preVisitDirectory(final Path dir, final BasicFileAttributes attrs) { + // Only descend into the SCANS directory immediately under the source root. + if (sourceDir.equals(dir.getParent()) + && !dir.getFileName().toString().equalsIgnoreCase("SCANS")) { + return FileVisitResult.SKIP_SUBTREE; + } + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs) { + if (attrs.isRegularFile() + && hasPathComponent(file, "DICOM") + && !file.getFileName().toString().toLowerCase().endsWith("catalog.xml")) { + // writeEntry returns false once the consumer closes the pipe → stop walking. + return writeEntry(zos, file, buffer) ? FileVisitResult.CONTINUE : FileVisitResult.TERMINATE; + } + return FileVisitResult.CONTINUE; + } + }); } catch (UncheckedIOException e) { recordError(e.getCause()); // a source DICOM file could not be read } catch (IOException e) { @@ -379,6 +535,16 @@ private void produce() { } } + /** True if any component of {@code p} matches {@code component}, ignoring case (archive folders are conventionally upper-case). */ + private static boolean hasPathComponent(final Path p, final String component) { + for (final Path part : p) { + if (part.toString().equalsIgnoreCase(component)) { + return true; + } + } + return false; + } + private void recordError(final Throwable cause) { error.set(cause); log.warn("Streaming zip producer for {} failed", displayName, cause); @@ -556,7 +722,7 @@ private void copyAssessor(XnatImageassessordata sourceAssess, XnatProjectdata de fixPaths(res, filepath, newFilepath); } - XnatUtils.doActionWithWorkflow(user, sourceAssess, "Cloned into project: " + newAssessor.getProject(), () -> { + XnatUtils.doActionWithWorkflow(user, sourceAssess, sourceWorkflowLabel(TransferMode.CLONE, sourceAssess.getLabel(), destinationProjectData.getId()), () -> { saveItemAndCopyFiles(user, sourceAssess, newAssessor, filepath, newFilepath); return true; }); @@ -634,7 +800,7 @@ private XnatExperimentdata copyExperiment(XnatExperimentdata sourceExpt, XnatPro fixPaths(res, filepath, newFilepath); } - XnatUtils.doActionWithWorkflow(user, sourceExpt, "Cloned into project: " + newExperiment.getProject(), () -> { + XnatUtils.doActionWithWorkflow(user, sourceExpt, sourceWorkflowLabel(TransferMode.CLONE, sourceExpt.getLabel(), destinationProjectData.getId()), () -> { saveItemAndCopyFiles(user, sourceExpt, newExperiment, filepath, newFilepath); return true; }); @@ -683,13 +849,21 @@ private XnatSubjectdata copySubject(XnatSubjectdata sourceSubject, XnatProjectda fixPaths(res, filepath, newFilepath); } - XnatUtils.doActionWithWorkflow(user, sourceSubject, "Cloned into project: " + newSubject.getProject(), () -> { + XnatUtils.doActionWithWorkflow(user, sourceSubject, sourceWorkflowLabel(TransferMode.CLONE, sourceSubject.getLabel(), destinationProjectData.getId()), () -> { saveItemAndCopyFiles(user, sourceSubject, newSubject, filepath, newFilepath); return true; }); return newSubject; } + /** + * Builds the source-item workflow label for Clone/Reimport, e.g. "Cloned sess1 to project P2". + * Centralized so the wording stays consistent across operations. (Share keeps its own label.) + */ + private static String sourceWorkflowLabel(final TransferMode mode, final String sourceLabel, final String destProjectId) { + return mode.getPastAction() + " " + sourceLabel + " to project " + destProjectId; + } + /** * Saves the new item, links the source item's files into the new item's archive directory, and will run the destination * project's anonymization script if the item is an image session. @@ -701,7 +875,8 @@ private XnatSubjectdata copySubject(XnatSubjectdata sourceSubject, XnatProjectda * @param dest - The destination filepath * @throws Exception */ - private void saveItemAndCopyFiles(final UserI user, final ArchivableItem sourceItem, final ArchivableItem newItem, final String source, final String dest) throws Exception { + // Package-private (not private) so BatchTransferServiceImplTest can drive the save+link path directly. + void saveItemAndCopyFiles(final UserI user, final ArchivableItem sourceItem, final ArchivableItem newItem, final String source, final String dest) throws Exception { try { EventDetails details = EventUtils.newEventInstance(EventUtils.CATEGORY.DATA, EventUtils.TYPE.WEB_SERVICE, "Item Saved"); SaveItemHelper.authorizedSave(newItem, user, false, false, details); @@ -709,11 +884,12 @@ private void saveItemAndCopyFiles(final UserI user, final ArchivableItem sourceI throw new Exception("Failed to save item", e); } + // Link files directly (no separate workflow): the enclosing source-item workflow + // (e.g. "Cloned sess1 to project P2") already wraps this whole method, so a link failure + // still fails that workflow and triggers the rollback below. The destination copy + // intentionally carries no plugin workflow of its own. try { - XnatUtils.doActionWithWorkflow(user, newItem, "Files Cloned", () -> { - FileUtil.linkFiles(source, dest); - return true; - }); + FileUtil.linkFiles(source, dest); } catch (Exception e) { XnatUtils.deleteItemWithoutSecurity(newItem); final Path destDir = Paths.get(dest); diff --git a/src/main/java/org/nrg/xnatx/plugins/transfer/util/file/BatchTransferDirectoryCopy.java b/src/main/java/org/nrg/xnatx/plugins/transfer/util/file/BatchTransferDirectoryCopy.java index 6a5efd0..9397366 100644 --- a/src/main/java/org/nrg/xnatx/plugins/transfer/util/file/BatchTransferDirectoryCopy.java +++ b/src/main/java/org/nrg/xnatx/plugins/transfer/util/file/BatchTransferDirectoryCopy.java @@ -76,7 +76,7 @@ public FileVisitResult visitFile(Path file, BasicFileAttributes attr) throws IOE } String destFileName = FileUtil.getDestFileName(file, sourcePath, destinationPath); - if (!CatalogUtils.isCatalogFile(file.toFile()) && useHardLink) { + if (!isCatalogFile(file) && useHardLink) { log.debug("Creating a link for {} to {}", file, destFileName); FileUtil.createLinkFile(file.toString(), destFileName, recreateIfExisted); } else { @@ -93,6 +93,13 @@ public FileVisitResult visitFile(Path file, BasicFileAttributes attr) throws IOE return FileVisitResult.CONTINUE; } + private static boolean isCatalogFile(Path file) { + final String name = file.getFileName().toString(); + if (name.endsWith("_catalog.xml")) return true; // standard catalogs + if (!name.endsWith(".xml")) return false; // DICOM/data: never read them + return CatalogUtils.isCatalogFile(file.toFile()); // rare odd-named .xml: still sniff + } + @Override public FileVisitResult preVisitDirectory(Path file, BasicFileAttributes attr) throws IOException { final String destFileName = FileUtil.getDestFileName(file, sourcePath, destinationPath); diff --git a/src/main/resources/META-INF/resources/scripts/xnat/plugin/batchTransfer/batchTransfer.js b/src/main/resources/META-INF/resources/scripts/xnat/plugin/batchTransfer/batchTransfer.js index f010539..736a97b 100644 --- a/src/main/resources/META-INF/resources/scripts/xnat/plugin/batchTransfer/batchTransfer.js +++ b/src/main/resources/META-INF/resources/scripts/xnat/plugin/batchTransfer/batchTransfer.js @@ -108,29 +108,20 @@ var XNAT = getObject(XNAT || {}); } renderProjects(document.getElementById('bt-project-select'), projects); $('#bt-project-select').prop('disabled', false); - fetchAnonForProjects(projects); }); window.projectLoader.init(); } - // Cache each project's anonymization state up-front so the operation-detail - // panel banner can render synchronously when the user picks a destination. - function fetchAnonForProjects(projects) { - projects.forEach(function(p) { - XNAT.plugin.batchtransfer.checkProjectAnon(p.id, function(anonEnabled) { - projectAnonCache[p.id] = anonEnabled; - }); - }); - } - // ── Anonymization Check ── - XNAT.plugin.batchtransfer.checkProjectAnon = function(projectId, callback) { + XNAT.plugin.batchtransfer.checkProjectAnon = function(projectId, callback, forceRefresh) { if (!projectId) { callback(null, 'No project selected'); return; } - if (projectAnonCache.hasOwnProperty(projectId)) { + // forceRefresh re-queries the server even on a cache hit, so the destination's anon + // status is current at selection time (it may have changed since the page loaded). + if (!forceRefresh && projectAnonCache.hasOwnProperty(projectId)) { callback(projectAnonCache[projectId], null); return; } @@ -248,6 +239,11 @@ var XNAT = getObject(XNAT || {}); // A non-reimportable subject is hidden, but its (reimportable) sessions still render: // showRows() treats a bt-import-hidden parent as logically visible. function applyOperationEligibility(op) { + // In Reimport, a session's only children are image assessors, which are not re-importable. + // Their expand arrows would open to nothing. Flag the body so CSS suppresses the arrows here. + var body = document.getElementById('bt-data-body'); + if (body) body.classList.toggle('bt-reimport-mode', op === 'Reimport'); + var attr = (op === 'Clone') ? 'data-cloneable' : (op === 'Reimport') ? 'data-reimportable' : 'data-shareable'; @@ -313,14 +309,13 @@ var XNAT = getObject(XNAT || {}); function bindProjectSelect() { $(document).on('change', '#bt-project-select', function() { selectedProject = $(this).val() || null; - if (selectedProject && projectAnonCache.hasOwnProperty(selectedProject)) { - selectedAnon = projectAnonCache[selectedProject]; - updateAnonBanner(); - } else if (selectedProject) { + if (selectedProject) { + // Re-query the destination's anon status on every selection so the banner is + // fresh, rather than reflecting whatever was cached when the page loaded. XNAT.plugin.batchtransfer.checkProjectAnon(selectedProject, function(anonEnabled) { selectedAnon = anonEnabled; updateAnonBanner(); - }); + }, true); } else { selectedAnon = null; updateAnonBanner(); @@ -603,7 +598,7 @@ var XNAT = getObject(XNAT || {}); function updateSummary() { $('#bt-sum-op').text(selectedOp); - $('#bt-sum-dest').text(selectedProject || '—'); + $('#bt-sum-dest').text(selectedProject || '-'); var subjects = 0, sessions = 0, assessors = 0, selectedCount = 0, eligibleCount = 0; var rows = document.querySelectorAll('#bt-data-body tr'); diff --git a/src/main/resources/META-INF/resources/style/xnat/plugin/batchTransfer/batchTransfer.css b/src/main/resources/META-INF/resources/style/xnat/plugin/batchTransfer/batchTransfer.css index 966acc6..dcc4bf3 100644 --- a/src/main/resources/META-INF/resources/style/xnat/plugin/batchTransfer/batchTransfer.css +++ b/src/main/resources/META-INF/resources/style/xnat/plugin/batchTransfer/batchTransfer.css @@ -925,6 +925,11 @@ display: none !important; } +/* Reimport mode: sessions are the only eligible rows, so suppress the otherwise-empty expand arrows. */ +#bt-data-body.bt-reimport-mode .bt-expand { + display: none; +} + /* Import disabled rows (subjects & assessors not importable) */ .bt-import-disabled { opacity: 0.4; diff --git a/src/main/resources/META-INF/xnat/spawner/batchTransfer/site-settings.yaml b/src/main/resources/META-INF/xnat/spawner/batchTransfer/site-settings.yaml new file mode 100644 index 0000000..5928b39 --- /dev/null +++ b/src/main/resources/META-INF/xnat/spawner/batchTransfer/site-settings.yaml @@ -0,0 +1,87 @@ +#### +#### Batch Transfer — Site-wide JMS Queue Settings +#### Adds a "Batch Transfer" group to Site Administration with a "JMS Queue" tab. The form is bound +#### to BatchTransferQueueSettingsApi (/xapi/batch_transfer/jms_queues), which reads/writes +#### BatchTransferQueuePrefsBean. Field names match the preference keys. +#### + +batchTransferQueueSettingsForm: + label: "Site-wide JMS Queue Settings" + kind: panel.form + name: batchTransferQueueSettings + id: batch-transfer-queue-settings-form + contentType: json + method: POST + action: "/xapi/batch_transfer/jms_queues" + contents: + restartNote: + tag: div.warning + contents: + "Changing JMS concurrency settings requires a Tomcat restart to take effect." + queueDescription: + tag: div.message + contents: + "Consumer concurrency for the Batch Transfer JMS queues: the Reimport queue parallelizes per image session; the Clone queue parallelizes per subject (one consumer handles a whole subject). Each reimport also feeds XNAT's prearchive rebuild queue (tuned separately under Site Administration), so keep these values modest. Set a queue's min and max both to 1 to disable its parallelism." + reimportQueueMin: + kind: panel.input.text + name: reimportConcurrencyMin + label: Reimport queue min concurrency + validation: "onblur integer greaterThan:0" + element: + title: Set min concurrency for Reimport queue consumers + description: > + Minimum number of concurrent consumers for the Batch Transfer Reimport queue. + Default is 4, must be greater than 0. + reimportQueueMax: + kind: panel.input.text + name: reimportConcurrencyMax + label: Reimport queue max concurrency + validation: "onblur integer greaterThan:0" + element: + title: Set max concurrency for Reimport queue consumers + description: > + Maximum number of concurrent consumers for the Batch Transfer Reimport queue. + Default is 8, must be greater than or equal to the minimum concurrency. + cloneQueueMin: + kind: panel.input.text + name: cloneConcurrencyMin + label: Clone queue min concurrency + validation: "onblur integer greaterThan:0" + element: + title: Set min concurrency for Clone queue consumers + description: > + Minimum number of concurrent consumers for the Batch Transfer Clone queue (each consumer + handles all of one subject's items). Default is 1, must be greater than 0. + cloneQueueMax: + kind: panel.input.text + name: cloneConcurrencyMax + label: Clone queue max concurrency + validation: "onblur integer greaterThan:0" + element: + title: Set max concurrency for Clone queue consumers + description: > + Maximum number of concurrent consumers for the Batch Transfer Clone queue. + Default is 2, must be greater than or equal to the minimum concurrency. + + +################################################# +#### Root Site Admin Spawner Config Object #### +################################################# + +siteSettings: + kind: tabs + name: batchTransferAdminPage + label: Administer Batch Transfer + meta: + tabGroups: + batchTransferTabGroup: Batch Transfer + contains: tabs + tabs: + batchTransferQueueTab: + kind: tab + name: batchTransferQueueTab + label: JMS Queue + group: batchTransferTabGroup + active: true + contents: + ${batchTransferQueueSettingsForm} diff --git a/src/test/java/org/nrg/xnatx/plugins/transfer/config/BatchTransferServiceConfig.java b/src/test/java/org/nrg/xnatx/plugins/transfer/config/BatchTransferServiceConfig.java index d320be2..de6a767 100644 --- a/src/test/java/org/nrg/xnatx/plugins/transfer/config/BatchTransferServiceConfig.java +++ b/src/test/java/org/nrg/xnatx/plugins/transfer/config/BatchTransferServiceConfig.java @@ -5,6 +5,7 @@ import org.nrg.framework.services.ContextService; import org.nrg.framework.services.NrgEventService; import org.nrg.xnat.helpers.merge.AnonUtils; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -24,7 +25,8 @@ public class BatchTransferServiceConfig { public BatchTransferServiceImpl batchTransferService(final NrgEventService nrgEventService, final ExecutorService executorService, final AnonUtils anonUtils, - final ContextService contextService) { - return Mockito.spy(new BatchTransferServiceImpl(nrgEventService, executorService, anonUtils, contextService)); + final ContextService contextService, + final BatchTransferMonitor batchTransferMonitor) { + return Mockito.spy(new BatchTransferServiceImpl(nrgEventService, executorService, anonUtils, contextService, batchTransferMonitor)); } } diff --git a/src/test/java/org/nrg/xnatx/plugins/transfer/config/MockConfig.java b/src/test/java/org/nrg/xnatx/plugins/transfer/config/MockConfig.java index 2e612ab..7f4de77 100644 --- a/src/test/java/org/nrg/xnatx/plugins/transfer/config/MockConfig.java +++ b/src/test/java/org/nrg/xnatx/plugins/transfer/config/MockConfig.java @@ -9,6 +9,7 @@ import org.nrg.xft.security.UserI; import org.nrg.xnat.DicomObjectIdentifier; import org.nrg.xnat.helpers.merge.AnonUtils; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -82,4 +83,9 @@ public XnatImageassessordata imageAssessor() { public XnatSubjectdata subject() { return mock(XnatSubjectdata.class); } + + @Bean + public BatchTransferMonitor batchTransferMonitor() { + return mock(BatchTransferMonitor.class); + } } diff --git a/src/test/java/org/nrg/xnatx/plugins/transfer/model/TransferModeTest.java b/src/test/java/org/nrg/xnatx/plugins/transfer/model/TransferModeTest.java new file mode 100644 index 0000000..d5095f2 --- /dev/null +++ b/src/test/java/org/nrg/xnatx/plugins/transfer/model/TransferModeTest.java @@ -0,0 +1,27 @@ +package org.nrg.xnatx.plugins.transfer.model; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +/** + * Verifies the label forms used to build workflow messages. {@code getPastAction()} backs the unified + * source-item workflow labels ("Cloned/Reimported <label> to project <dest>"). + */ +public class TransferModeTest { + + @Test + public void getPastAction_returnsPastTenseLabels() { + assertEquals("Cloned", TransferMode.CLONE.getPastAction()); + assertEquals("Shared", TransferMode.SHARE.getPastAction()); + assertEquals("Reimported", TransferMode.REIMPORT.getPastAction()); + } + + @Test + public void getActionAndValue_unchanged() { + assertEquals("Cloning", TransferMode.CLONE.getAction()); + assertEquals("Clone", TransferMode.CLONE.getValue()); + assertEquals("Reimporting", TransferMode.REIMPORT.getAction()); + assertEquals("Reimport", TransferMode.REIMPORT.getValue()); + } +} diff --git a/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImplTest.java b/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImplTest.java index dce6c9a..5cde05e 100644 --- a/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImplTest.java +++ b/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/BatchTransferServiceImplTest.java @@ -3,9 +3,14 @@ import org.nrg.xnatx.plugins.transfer.config.BatchTransferServiceConfig; import org.nrg.xnatx.plugins.transfer.event.BatchTransferEvent; import org.nrg.xnatx.plugins.transfer.model.BatchTransfer; +import org.nrg.xnatx.plugins.transfer.model.EventInfo; import org.nrg.xnatx.plugins.transfer.model.TransferMode; import org.nrg.xnatx.plugins.transfer.model.TransferRequest; +import org.nrg.xnatx.plugins.transfer.util.FileUtil; import org.nrg.xnatx.plugins.transfer.util.XnatUtils; +import org.nrg.xnatx.plugins.transfer.jms.requests.CloneSubjectRequest; +import org.nrg.xnatx.plugins.transfer.jms.requests.TransferItemRequest; +import org.nrg.xnatx.plugins.transfer.jms.tasks.BatchTransferMonitor; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -16,16 +21,19 @@ import org.mockito.MockedStatic; import org.mockito.Mockito; import org.nrg.framework.services.NrgEventService; +import org.nrg.xdat.XDAT; import org.nrg.xdat.om.XnatImageassessordata; import org.nrg.xdat.om.XnatImagesessiondata; import org.nrg.xdat.om.XnatProjectdata; import org.nrg.xdat.om.XnatSubjectdata; -import org.nrg.xdat.om.base.BaseXnatExperimentdata; import org.nrg.xdat.om.base.BaseXnatSubjectdata; import org.nrg.xdat.security.helpers.Features; import org.nrg.xdat.security.helpers.Permissions; +import org.nrg.xft.event.EventUtils; import org.nrg.xft.security.UserI; +import org.nrg.xft.utils.SaveItemHelper; import org.nrg.xnat.helpers.prearchive.PrearcSession; +import org.nrg.xnat.turbine.utils.ArchivableItem; import org.nrg.xnat.helpers.prearchive.PrearcUtils; import org.nrg.xnat.helpers.uri.URIManager; import org.nrg.xnat.restlet.util.FileWriterWrapperI; @@ -43,29 +51,33 @@ import java.util.concurrent.ExecutorService; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mockConstruction; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** - * Unit tests for the IMPORT operation of {@link BatchTransferServiceImpl}. + * Unit tests for the per-item REIMPORT logic of {@link BatchTransferServiceImpl}. * - *

The service bean is a Mockito spy (see BatchTransferServiceConfig) so tests - * can stub the protected {@code runImporter} seam without loading - * {@code DicomZipImporter} — whose supertype static initialisers require - * XDAT / Spring wiring that isn't present in a unit-test JVM. + *

Since the Reimport/Clone batch paths now fan out to JMS (a consumer calls {@code processItem} + * per item), these tests exercise {@link BatchTransferServiceImpl#processItem} directly — the shared + * per-item engine — rather than driving a batch through {@code submitTransferRequest}. {@code processItem} + * does the validate/load/permission/dispatch work and throws on failure; it does not emit + * activity-monitor events (the orchestrators do), so the assertions are on what it throws/returns and + * on the {@code runImporter}/prearchive side effects. + * + *

The service bean is a Mockito spy (see {@code BatchTransferServiceConfig}) so tests can stub the + * protected {@code runImporter} seam without loading {@code DicomZipImporter}. */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(classes = BatchTransferServiceConfig.class) @@ -81,6 +93,7 @@ public class BatchTransferServiceImplTest { @Autowired private XnatImagesessiondata imageSession; @Autowired private XnatImageassessordata imageAssessor; @Autowired private XnatSubjectdata subject; + @Autowired private BatchTransferMonitor monitor; private static final String DEST_PROJECT = "destProj"; private static final String SRC_PROJECT = "srcProj"; @@ -91,12 +104,11 @@ public class BatchTransferServiceImplTest { @Before public void setUp() throws Exception { Mockito.reset(service, nrgEventService, executorService, user, - destinationProjectData, imageSession, imageAssessor, subject); + destinationProjectData, imageSession, imageAssessor, subject, monitor); when(user.getID()).thenReturn(42); - // Run submitted Runnables on the test thread so per-thread MockedStatic - // scopes opened in a test apply to the batch loop. + // Run submitted Runnables on the test thread (used by the empty/null submit test). doAnswer(inv -> { ((Runnable) inv.getArgument(0)).run(); return null; @@ -111,17 +123,22 @@ public void setUp() throws Exception { // Helpers // ----------------------------------------------------------------- - private BatchTransfer importRequest(String id) { - return new BatchTransfer( - Collections.singletonList( - new TransferRequest(DEST_PROJECT, id, TransferMode.REIMPORT)), - TRACKING_ID); + private TransferRequest reimport(final String id) { + return new TransferRequest(DEST_PROJECT, id, TransferMode.REIMPORT); + } + + private EventInfo eventInfo() { + return new EventInfo(TRACKING_ID, 0); } - private List captureEvents() { - ArgumentCaptor cap = ArgumentCaptor.forClass(BatchTransferEvent.class); - verify(nrgEventService, atLeastOnce()).triggerEvent(cap.capture()); - return cap.getAllValues(); + /** Calls processItem and returns the thrown exception, or null if it returned normally. */ + private Exception processItemCatching(final TransferRequest request) { + try { + service.processItem(request, user, eventInfo()); + return null; + } catch (Exception e) { + return e; + } } private Map uriProps() { @@ -146,7 +163,7 @@ private void stubImageSessionAsSource() throws Exception { // ----------------------------------------------------------------- /** - * Null or empty request list: never submit a task and never emit events. + * Null or empty request list: submitTransferRequest never submits a task and never emits events. */ @Test public void submitTransferRequest_emptyOrNullRequests_isNoop() { @@ -158,12 +175,11 @@ public void submitTransferRequest_emptyOrNullRequests_isNoop() { } /** - * IMPORT on an XnatImageassessordata throws at the top of importExperiment - * (BatchTransferServiceImpl.java:191-193). The failure is caught by batchTransfer - * and results in a Failed event plus a terminal Warning event (line 168). + * REIMPORT of an XnatImageassessordata throws at the top of importExperiment + * ("Reimport operation is not supported for assessors."). */ @Test - public void importAssessor_throwsAndFailsBatch() { + public void importAssessor_throws() { when(imageAssessor.getXSIType()).thenReturn("xnat:imageAssessorData"); when(imageAssessor.getProject()).thenReturn(SRC_PROJECT); when(imageAssessor.getId()).thenReturn(EXP_ID); @@ -173,105 +189,75 @@ public void importAssessor_throwsAndFailsBatch() { MockedStatic perms = mockStatic(Permissions.class); MockedStatic feats = mockStatic(Features.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(imageAssessor); - + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(imageAssessor); perms.when(() -> Permissions.canRead(user, imageAssessor)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); - service.submitTransferRequest(importRequest(EXP_ID), user); + final Exception thrown = processItemCatching(reimport(EXP_ID)); + assertNotNull("expected processItem to throw for an assessor reimport", thrown); + assertTrue("message should mention assessor rejection: " + thrown.getMessage(), + thrown.getMessage() != null && thrown.getMessage().contains("not supported for assessors")); } - - List events = captureEvents(); - BatchTransferEvent failed = events.stream() - .filter(e -> e.getStatus() == BatchTransferEvent.Status.Failed) - .findFirst().orElse(null); - assertNotNull("expected a Failed event", failed); - assertTrue("failure message should mention assessor rejection: " + failed.getMessage(), - failed.getMessage() != null - && failed.getMessage().contains("not supported for assessors")); - assertEquals(BatchTransferEvent.Status.Warning, events.get(events.size() - 1).getStatus()); } /** - * IMPORT on an XnatSubjectdata hits the subject branch at - * BatchTransferServiceImpl.java:137-138, which just logs a warning and does not - * throw. The batch completes normally with no Failed event. + * REIMPORT of an XnatSubjectdata is a no-op (logs a warning); processItem returns without + * throwing and never invokes the importer seam. */ @Test - public void importSubject_logsAndContinues() throws Exception { + public void importSubject_isNoop() throws Exception { when(subject.getXSIType()).thenReturn("xnat:subjectData"); when(subject.getProject()).thenReturn(SRC_PROJECT); when(subject.getId()).thenReturn(EXP_ID); when(subject.getLabel()).thenReturn(LABEL); - try (MockedStatic utils = mockStatic(XnatUtils.class); - MockedStatic perms = mockStatic(Permissions.class); - MockedStatic feats = mockStatic(Features.class); + try (MockedStatic utils = mockStatic(XnatUtils.class); + MockedStatic perms = mockStatic(Permissions.class); + MockedStatic feats = mockStatic(Features.class); MockedStatic subMock = mockStatic(BaseXnatSubjectdata.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(subject); - + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(subject); perms.when(() -> Permissions.canRead(user, subject)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); subMock.when(() -> BaseXnatSubjectdata.GetSubjectByProjectIdentifier( eq(DEST_PROJECT), eq(LABEL), any(UserI.class), eq(false))).thenReturn(null); - service.submitTransferRequest(importRequest(EXP_ID), user); + service.processItem(reimport(EXP_ID), user, eventInfo()); - // Subject IMPORT must never invoke the importer seam. verify(service, never()).runImporter( any(UserI.class), any(FileWriterWrapperI.class), any(Map.class)); } - - List events = captureEvents(); - assertTrue("subject IMPORT should not emit a Failed event", - events.stream().noneMatch(e -> e.getStatus() == BatchTransferEvent.Status.Failed)); - assertEquals(BatchTransferEvent.Status.Completed, events.get(events.size() - 1).getStatus()); } /** - * validateRequest (BatchTransferServiceImpl.java:574-586): blank id, blank - * destination project, and null operation each produce a Failed event - * before XnatUtils.getProject is invoked. + * validateRequest: blank id, blank destination project, and null operation each make processItem + * throw before XnatUtils.getProject is invoked. */ @Test - public void validateRequest_missingFields_fails() { - List bad = new ArrayList<>(); + public void validateRequest_missingFields_throwsBeforeLoad() { + final List bad = new ArrayList<>(); bad.add(new TransferRequest(DEST_PROJECT, "", TransferMode.REIMPORT)); // blank id bad.add(new TransferRequest("", EXP_ID, TransferMode.REIMPORT)); // blank destination bad.add(new TransferRequest(DEST_PROJECT, EXP_ID, null)); // null operation - for (TransferRequest req : bad) { - Mockito.reset(nrgEventService); + for (final TransferRequest req : bad) { try (MockedStatic utils = mockStatic(XnatUtils.class)) { - service.submitTransferRequest( - new BatchTransfer(Collections.singletonList(req), TRACKING_ID), user); + final Exception thrown = processItemCatching(req); + assertNotNull("expected processItem to throw for " + req, thrown); utils.verify(() -> XnatUtils.getProject(anyString(), any(UserI.class)), never()); } - List events = captureEvents(); - assertTrue("expected Failed event for " + req, - events.stream().anyMatch(e -> e.getStatus() == BatchTransferEvent.Status.Failed)); } } /** - * When source project equals destination project, batchTransfer emits a Failed - * event (BatchTransferServiceImpl.java:109-111) before importExperiment runs. + * When source project equals destination project, processItem throws before importExperiment runs. */ @Test - public void importSameSourceAndDest_fails() throws Exception { + public void importSameSourceAndDest_throws() { when(imageSession.getXSIType()).thenReturn("xnat:mrSessionData"); when(imageSession.getProject()).thenReturn(DEST_PROJECT); // source == dest when(imageSession.getId()).thenReturn(EXP_ID); @@ -281,34 +267,22 @@ public void importSameSourceAndDest_fails() throws Exception { MockedStatic perms = mockStatic(Permissions.class); MockedStatic feats = mockStatic(Features.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(imageSession); + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(imageSession); perms.when(() -> Permissions.canRead(user, imageSession)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); - service.submitTransferRequest(importRequest(EXP_ID), user); + final Exception thrown = processItemCatching(reimport(EXP_ID)); + assertNotNull("expected processItem to throw when source == dest", thrown); + assertTrue("message should mention the source/destination conflict: " + thrown.getMessage(), + thrown.getMessage() != null && thrown.getMessage().contains("same as the source project")); } - - List events = captureEvents(); - BatchTransferEvent failed = events.stream() - .filter(e -> e.getStatus() == BatchTransferEvent.Status.Failed) - .findFirst().orElse(null); - assertNotNull("expected a Failed event", failed); - assertTrue("Failed message should mention the source/destination conflict: " - + failed.getMessage(), - failed.getMessage() != null - && failed.getMessage().contains("same as the source project")); } /** - * Happy path: IMPORT on an XnatImagesessiondata invokes the (spy-stubbed) - * runImporter with a streaming wrapper and commits URIs to the prearchive. - * Terminal event is Completed. + * Happy path: REIMPORT of an XnatImagesessiondata invokes the (spy-stubbed) runImporter with a + * streaming wrapper and commits the returned URIs to the prearchive. processItem returns normally. */ @Test public void importImagesession_happyPath() throws Exception { @@ -319,40 +293,29 @@ public void importImagesession_happyPath() throws Exception { doReturn(importerUris).when(service).runImporter( any(UserI.class), any(FileWriterWrapperI.class), any(Map.class)); - try (MockedStatic utils = mockStatic(XnatUtils.class); - MockedStatic perms = mockStatic(Permissions.class); - MockedStatic feats = mockStatic(Features.class); - MockedStatic expt = mockStatic(BaseXnatExperimentdata.class); - MockedStatic prearc = mockStatic(PrearcUtils.class); - // Bypass PrearcSession / PrearchiveOperationRequest real constructors — - // they depend on XDAT/prearchive state not available in unit tests. + try (MockedStatic utils = mockStatic(XnatUtils.class); + MockedStatic perms = mockStatic(Permissions.class); + MockedStatic feats = mockStatic(Features.class); + MockedStatic prearc = mockStatic(PrearcUtils.class); + // Bypass PrearcSession / PrearchiveOperationRequest real constructors. MockedConstruction sess = mockConstruction(PrearcSession.class); MockedConstruction req = mockConstruction(PrearchiveOperationRequest.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(imageSession); + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(imageSession); utils.when(() -> XnatUtils.doActionWithWorkflow( any(UserI.class), any(), anyString(), any(Callable.class))) .thenAnswer(inv -> { ((Callable) inv.getArgument(3)).call(); return true; }); - perms.when(() -> Permissions.canRead(user, imageSession)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); - expt.when(() -> BaseXnatExperimentdata.GetExptByProjectIdentifier( - eq(DEST_PROJECT), eq(LABEL), any(UserI.class), eq(false))) - .thenReturn(null); - + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); prearc.when(() -> PrearcUtils.parseURI(anyString())).thenReturn(uriProps()); - service.submitTransferRequest(importRequest(EXP_ID), user); + service.processItem(reimport(EXP_ID), user, eventInfo()); verify(service).runImporter(eq(user), any(FileWriterWrapperI.class), any(Map.class)); prearc.verify(() -> PrearcUtils.parseURI(importerUris.get(0))); @@ -360,35 +323,25 @@ public void importImagesession_happyPath() throws Exception { assertEquals(1, sess.constructed().size()); assertEquals(1, req.constructed().size()); } - - List events = captureEvents(); - assertEquals(BatchTransferEvent.Status.Waiting, events.get(0).getStatus()); - assertEquals(BatchTransferEvent.Status.Completed, events.get(events.size() - 1).getStatus()); - assertTrue(events.stream().anyMatch(e -> e.getStatus() == BatchTransferEvent.Status.InProgress)); - assertFalse(events.stream().anyMatch(e -> e.getStatus() == BatchTransferEvent.Status.Failed)); } /** - * If runImporter returns an empty URI list (source experiment had no - * DICOM files), importExperiment throws inside the workflow callable so - * the batch emits a Failed event instead of silently completing. + * If runImporter returns an empty URI list, importExperiment throws inside the workflow callable + * ("No DICOM files ... nothing to reimport"), so processItem throws. */ @Test - public void importerReturnsEmpty_failsBatch() throws Exception { + public void importerReturnsEmpty_throws() throws Exception { stubImageSessionAsSource(); doReturn(Collections.emptyList()).when(service).runImporter( any(UserI.class), any(FileWriterWrapperI.class), any(Map.class)); - try (MockedStatic utils = mockStatic(XnatUtils.class); - MockedStatic perms = mockStatic(Permissions.class); - MockedStatic feats = mockStatic(Features.class); - MockedStatic expt = mockStatic(BaseXnatExperimentdata.class)) { + try (MockedStatic utils = mockStatic(XnatUtils.class); + MockedStatic perms = mockStatic(Permissions.class); + MockedStatic feats = mockStatic(Features.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(imageSession); + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(imageSession); utils.when(() -> XnatUtils.doActionWithWorkflow( any(UserI.class), any(), anyString(), any(Callable.class))) .thenAnswer(inv -> { @@ -396,47 +349,33 @@ public void importerReturnsEmpty_failsBatch() throws Exception { return true; }); perms.when(() -> Permissions.canRead(user, imageSession)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); - expt.when(() -> BaseXnatExperimentdata.GetExptByProjectIdentifier( - eq(DEST_PROJECT), eq(LABEL), any(UserI.class), eq(false))) - .thenReturn(null); - - service.submitTransferRequest(importRequest(EXP_ID), user); - } + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); - List events = captureEvents(); - BatchTransferEvent failed = events.stream() - .filter(e -> e.getStatus() == BatchTransferEvent.Status.Failed) - .findFirst().orElse(null); - assertNotNull("expected a Failed event when importer returns empty URI list", failed); - assertTrue("Failed message should mention no DICOM files: " + failed.getMessage(), - failed.getMessage() != null && failed.getMessage().contains("No DICOM files")); + final Exception thrown = processItemCatching(reimport(EXP_ID)); + assertNotNull("expected processItem to throw when the importer returns an empty URI list", thrown); + assertTrue("message should mention no DICOM files: " + thrown.getMessage(), + thrown.getMessage() != null && thrown.getMessage().contains("No DICOM files")); + } } /** - * If runImporter throws, batchTransfer's per-request try/catch catches it - * and emits a Failed event. (The workflow wrap in importExperiment - * rewraps the cause as " Failed".) + * If runImporter throws, the failure propagates out of processItem (the real workflow wrap would + * fail the workflow and rethrow; here doActionWithWorkflow is stubbed to invoke the callable). */ @Test - public void importerThrows_emitsFailedEvent() throws Exception { + public void importerThrows_propagates() throws Exception { stubImageSessionAsSource(); doThrow(new RuntimeException("boom")).when(service).runImporter( any(UserI.class), any(FileWriterWrapperI.class), any(Map.class)); - try (MockedStatic utils = mockStatic(XnatUtils.class); - MockedStatic perms = mockStatic(Permissions.class); - MockedStatic feats = mockStatic(Features.class); - MockedStatic expt = mockStatic(BaseXnatExperimentdata.class)) { + try (MockedStatic utils = mockStatic(XnatUtils.class); + MockedStatic perms = mockStatic(Permissions.class); + MockedStatic feats = mockStatic(Features.class)) { - utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)) - .thenReturn(destinationProjectData); - utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)) - .thenReturn(imageSession); + utils.when(() -> XnatUtils.getProject(DEST_PROJECT, user)).thenReturn(destinationProjectData); + utils.when(() -> XnatUtils.getArchivableItem(EXP_ID, null)).thenReturn(imageSession); utils.when(() -> XnatUtils.doActionWithWorkflow( any(UserI.class), any(), anyString(), any(Callable.class))) .thenAnswer(inv -> { @@ -444,19 +383,110 @@ public void importerThrows_emitsFailedEvent() throws Exception { return true; }); perms.when(() -> Permissions.canRead(user, imageSession)).thenReturn(true); - perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))) - .thenReturn(true); - feats.when(() -> Features.checkRestrictedFeature( - eq(user), anyString(), anyString())).thenReturn(true); - expt.when(() -> BaseXnatExperimentdata.GetExptByProjectIdentifier( - eq(DEST_PROJECT), eq(LABEL), any(UserI.class), eq(false))) - .thenReturn(null); - - service.submitTransferRequest(importRequest(EXP_ID), user); + perms.when(() -> Permissions.canCreate(eq(user), anyString(), eq(DEST_PROJECT))).thenReturn(true); + feats.when(() -> Features.checkRestrictedFeature(eq(user), anyString(), anyString())).thenReturn(true); + + final Exception thrown = processItemCatching(reimport(EXP_ID)); + assertNotNull("expected processItem to propagate a runImporter failure", thrown); + } + } + + // ----------------------------------------------------------------- + // Clone — saveItemAndCopyFiles after the destination-workflow collapse + // ----------------------------------------------------------------- + + /** + * File-linking now runs without its own "Files Cloned" workflow (the destination workflow was + * collapsed into the source-item workflow created by the copy* methods). A link failure must still + * roll the saved item back (deleteItemWithoutSecurity) and propagate, via saveItemAndCopyFiles's + * own catch — and no workflow is created inside this method. + */ + @Test + public void saveItemAndCopyFiles_linkFailure_rollsBackAndCreatesNoWorkflow() throws Exception { + final ArchivableItem newItem = Mockito.mock(ArchivableItem.class); + final String source = tmp.newFolder("clone-src").getAbsolutePath(); + final String dest = tmp.getRoot().getAbsolutePath() + "/clone-dst/leaf"; + + try (MockedStatic events = mockStatic(EventUtils.class); + MockedStatic save = mockStatic(SaveItemHelper.class); + MockedStatic files = mockStatic(FileUtil.class); + MockedStatic utils = mockStatic(XnatUtils.class)) { + + files.when(() -> FileUtil.linkFiles(anyString(), anyString())) + .thenThrow(new RuntimeException("link boom")); + + Exception thrown = null; + try { + service.saveItemAndCopyFiles(user, newItem, newItem, source, dest); + } catch (Exception e) { + thrown = e; + } + + assertNotNull("expected a link failure to propagate out of saveItemAndCopyFiles", thrown); + utils.verify(() -> XnatUtils.deleteItemWithoutSecurity(newItem)); + utils.verify(() -> XnatUtils.doActionWithWorkflow( + any(UserI.class), any(), anyString(), any(Callable.class)), never()); } + } + + // ----------------------------------------------------------------- + // batchTransfer — split a mixed batch by operation + // ----------------------------------------------------------------- - List events = captureEvents(); - assertTrue("expected a Failed event when runImporter throws", - events.stream().anyMatch(e -> e.getStatus() == BatchTransferEvent.Status.Failed)); + /** + * A batch can mix operations. batchTransfer must split by operation and route each to its path — + * Reimport → reimport queue, Clone → clone queue (grouped by subject), Share → in-process — while + * registering the batch with the monitor exactly once (one terminal event for the whole batch). + */ + @Test + public void batchTransfer_mixedBatch_splitsByOperation() throws Exception { + final TransferRequest reimportReq = new TransferRequest(DEST_PROJECT, "exp_reimport", TransferMode.REIMPORT); + final TransferRequest cloneReq1 = new TransferRequest(DEST_PROJECT, "exp_clone1", TransferMode.CLONE); + final TransferRequest cloneReq2 = new TransferRequest(DEST_PROJECT, "exp_clone2", TransferMode.CLONE); + final TransferRequest shareReq = new TransferRequest(DEST_PROJECT, "exp_share", TransferMode.SHARE); + final List all = new ArrayList<>(); + all.add(reimportReq); all.add(cloneReq1); all.add(cloneReq2); all.add(shareReq); + final BatchTransfer batch = new BatchTransfer(all, TRACKING_ID); + + // Share items run in-process via processItem; stub it (the service is a spy) so this router test + // doesn't execute the real share logic. Reimport/Clone items are enqueued, not processed here. + Mockito.doNothing().when(service).processItem(any(TransferRequest.class), any(UserI.class), any(EventInfo.class)); + + try (MockedStatic xdat = mockStatic(XDAT.class); + MockedStatic utils = mockStatic(XnatUtils.class)) { + + // Clone grouping resolves each clone item's subject (both → SUBJ1, so one bundle of two). + utils.when(() -> XnatUtils.getArchivableItem("exp_clone1", null)).thenReturn(imageSession); + utils.when(() -> XnatUtils.getArchivableItem("exp_clone2", null)).thenReturn(imageSession); + when(imageSession.getProperty("subject_ID")).thenReturn("SUBJ1"); + + service.submitTransferRequest(batch, user); // mock executor runs batchTransfer on this thread + + // Registered once for the whole batch (all four items) → a single terminal event. + verify(monitor).register(TRACKING_ID, 42, 4); + + // Two JMS sends: one reimport item, one clone bundle. + final ArgumentCaptor sent = ArgumentCaptor.forClass(Object.class); + xdat.verify(() -> XDAT.sendJmsRequest(sent.capture()), times(2)); + + final TransferItemRequest reimportSent = sent.getAllValues().stream() + .filter(r -> r instanceof TransferItemRequest).map(r -> (TransferItemRequest) r) + .findFirst().orElse(null); + assertNotNull("reimport item should be queued to the reimport queue", reimportSent); + assertEquals("exp_reimport", reimportSent.getItemId()); + + final CloneSubjectRequest cloneSent = sent.getAllValues().stream() + .filter(r -> r instanceof CloneSubjectRequest).map(r -> (CloneSubjectRequest) r) + .findFirst().orElse(null); + assertNotNull("clone items should be queued to the clone queue", cloneSent); + assertEquals("SUBJ1", cloneSent.getSubjectId()); + assertEquals(2, cloneSent.getItems().size()); + + // Share item is processed in-process; reimport/clone items were enqueued, not processed. + final ArgumentCaptor processed = ArgumentCaptor.forClass(TransferRequest.class); + verify(service).processItem(processed.capture(), any(UserI.class), any(EventInfo.class)); + assertEquals("exp_share", processed.getValue().getId()); + assertEquals(TransferMode.SHARE, processed.getValue().getMode()); + } } } diff --git a/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/StreamingZipFileWriterTest.java b/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/StreamingZipFileWriterTest.java index bc48bf6..66c6ab7 100644 --- a/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/StreamingZipFileWriterTest.java +++ b/src/test/java/org/nrg/xnatx/plugins/transfer/service/impl/StreamingZipFileWriterTest.java @@ -30,16 +30,19 @@ public class StreamingZipFileWriterTest { @Rule public TemporaryFolder tmp = new TemporaryFolder(); - /** Verifies the DICOM filter, the catalog.xml exclusion, and entry contents. */ + /** Verifies the SCANS+DICOM filter (case-insensitive), the catalog.xml exclusion, and entry contents. */ @Test public void streamsDicomFilesAndExcludesCatalog() throws Exception { final Path source = tmp.newFolder("source").toPath(); - final Path dicomDir = Files.createDirectories(source.resolve("scans/1/DICOM")); + final Path dicomDir = Files.createDirectories(source.resolve("SCANS/1/DICOM")); Files.write(dicomDir.resolve("img1.dcm"), "img1-bytes".getBytes(StandardCharsets.UTF_8)); Files.write(dicomDir.resolve("img2.dcm"), "img2-bytes".getBytes(StandardCharsets.UTF_8)); Files.write(dicomDir.resolve("catalog.xml"), "".getBytes(StandardCharsets.UTF_8)); // Non-DICOM file outside the DICOM dir: must be filtered out. - Files.write(source.resolve("scans/1/NOTE.txt"), "nope".getBytes(StandardCharsets.UTF_8)); + Files.write(source.resolve("SCANS/1/NOTE.txt"), "nope".getBytes(StandardCharsets.UTF_8)); + // DICOM file with no SCANS ancestor: must be filtered out by the (intentional) SCANS narrowing. + Files.write(Files.createDirectories(source.resolve("DICOM")).resolve("orphan.dcm"), + "orphan".getBytes(StandardCharsets.UTF_8)); try (BatchTransferServiceImpl.StreamingZipFileWriter w = new BatchTransferServiceImpl.StreamingZipFileWriter(source, "exp.zip")) { @@ -50,12 +53,14 @@ public void streamsDicomFilesAndExcludesCatalog() throws Exception { assertEquals("expected exactly the two DICOM files, got " + entries.keySet(), 2, entries.size()); - assertArrayContains(entries, "scans/1/DICOM/img1.dcm", "img1-bytes"); - assertArrayContains(entries, "scans/1/DICOM/img2.dcm", "img2-bytes"); + assertArrayContains(entries, "SCANS/1/DICOM/img1.dcm", "img1-bytes"); + assertArrayContains(entries, "SCANS/1/DICOM/img2.dcm", "img2-bytes"); assertNull("catalog.xml should be excluded", - entries.get("scans/1/DICOM/catalog.xml")); + entries.get("SCANS/1/DICOM/catalog.xml")); assertNull("NOTE.txt should be excluded (not in DICOM dir)", - entries.get("scans/1/NOTE.txt")); + entries.get("SCANS/1/NOTE.txt")); + assertNull("DICOM file with no SCANS ancestor should be excluded", + entries.get("DICOM/orphan.dcm")); } } @@ -84,7 +89,7 @@ public void getInputStreamTwice_throws() throws Exception { @Test public void awaitProducerTimeout_throwsIOException() throws Exception { final Path source = tmp.newFolder("source").toPath(); - final Path dicomDir = Files.createDirectories(source.resolve("DICOM")); + final Path dicomDir = Files.createDirectories(source.resolve("SCANS/1/DICOM")); // Incompressible random bytes so Deflater.BEST_SPEED can't shrink the // payload below the pipe buffer — without this, the producer would // finish instantly and the timeout would never fire.