You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by ha...@apache.org on 2026/07/31 08:34:14 UTC
(hop) branch main updated: add progress monitor to marketplace, fixes #7693 (#7696)
This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 2e94d0fbc6 add progress monitor to marketplace, fixes #7693 (#7696)
2e94d0fbc6 is described below
commit 2e94d0fbc6683c303679009b485ffe7b7a8e6d4d
Author: Hans Van Akelyen <ha...@gmail.com>
AuthorDate: Fri Jul 31 10:34:09 2026 +0200
add progress monitor to marketplace, fixes #7693 (#7696)
* add progress monitor to marketplace, fixes #7693
* fix final iteration of cli installer
---
.../command/ConsoleInstallListener.java | 267 ++++++++++++++++
.../marketplace/command/MarketplaceCommand.java | 21 +-
.../hop/marketplace/env/EnvironmentApplier.java | 33 +-
.../hop/marketplace/gui/HopEnvironmentDialog.java | 65 +++-
.../hop/marketplace/gui/MarketplaceDialog.java | 326 +++++++++++++++-----
.../gui/ProgressMonitorInstallListener.java | 306 +++++++++++++++++++
.../hop/marketplace/install/IInstallListener.java | 71 +++++
.../hop/marketplace/install/PluginInstaller.java | 30 +-
.../hop/marketplace/install/TransferFormat.java | 105 +++++++
.../hop/marketplace/resolve/ITransferListener.java | 64 ++++
.../marketplace/resolve/MavenRepositoryClient.java | 105 ++++++-
.../gui/messages/messages_en_US.properties | 22 ++
.../command/ConsoleInstallListenerTest.java | 252 ++++++++++++++++
.../gui/ProgressMonitorInstallListenerTest.java | 334 +++++++++++++++++++++
.../marketplace/install/PluginInstallerTest.java | 142 +++++++++
.../marketplace/install/TransferFormatTest.java | 92 ++++++
.../resolve/MavenRepositoryClientDownloadTest.java | 242 +++++++++++++++
17 files changed, 2388 insertions(+), 89 deletions(-)
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/ConsoleInstallListener.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/ConsoleInstallListener.java
new file mode 100644
index 0000000000..8ccda4fcfc
--- /dev/null
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/ConsoleInstallListener.java
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.command;
+
+import java.io.PrintStream;
+import org.apache.hop.marketplace.install.IInstallListener;
+import org.apache.hop.marketplace.install.TransferFormat;
+
+/**
+ * Reports install progress on a terminal with a live progress bar, the way package managers do:
+ *
+ * <pre>
+ * Downloading: Data Vault (353MB)
+ * [████████████░░░░░░░░░░░░] 52% 1.8MB/s - 184MB of 353MB, 1 min left
+ * </pre>
+ *
+ * <p>The bar line is rewritten in place with a carriage return. That only works on a terminal, so
+ * when output is redirected to a file or a pipe — a CI log, {@code hop marketplace install >
+ * out.txt} — the per-chunk updates are dropped and only the milestone lines are printed. Without
+ * that split, a redirected install would write thousands of bar frames into the log.
+ *
+ * <p>Callbacks arrive on the download thread, but a CLI install has no other thread writing to
+ * stdout, so no synchronisation is needed.
+ */
+public class ConsoleInstallListener implements IInstallListener {
+
+ /** Minimum gap between redraws. Fast enough to look live, slow enough not to flicker. */
+ static final long UPDATE_INTERVAL_MS = 200L;
+
+ private static final int BAR_WIDTH = 24;
+
+ /** Speed is averaged over this window so the number does not flicker on a bursty link. */
+ private static final long SPEED_WINDOW_MS = 2000L;
+
+ /** Block glyphs, as most modern CLI tools draw a bar. */
+ private static final char[] BLOCK_GLYPHS = {'█', '░'};
+
+ /**
+ * Fallback for consoles whose encoding cannot represent the blocks (legacy Windows code pages).
+ */
+ private static final char[] ASCII_GLYPHS = {'#', '-'};
+
+ private final PrintStream out;
+ private final boolean interactive;
+ private final char[] glyphs;
+ private final IClock clock;
+
+ /** Characters written to the current in-place line, so the next redraw can erase it fully. */
+ private int openLineWidth;
+
+ private long lastUpdateMs;
+
+ /** Set once the full-bar frame has been drawn, so a repeated final callback is not redrawn. */
+ private boolean transferFinished;
+
+ private long transferStartMs;
+ private long windowStartMs;
+ private long windowStartBytes;
+ private long speedBytesPerSec;
+
+ /** Indirection over the clock so throttling and speed can be unit-tested. */
+ public interface IClock {
+ long millis();
+ }
+
+ /**
+ * A listener writing to stdout, drawing a live bar only when stdout is a terminal.
+ *
+ * <p>{@code System.console()} returns null when output is redirected to a file or pipe — the same
+ * check {@code ConsoleLoggingEventListener} uses to decide on colour.
+ */
+ public static ConsoleInstallListener forStdOut() {
+ return new ConsoleInstallListener(
+ System.out, System.console() != null, glyphsFor(System.out), System::currentTimeMillis);
+ }
+
+ ConsoleInstallListener(PrintStream out, boolean interactive, char[] glyphs, IClock clock) {
+ this.out = out;
+ this.interactive = interactive;
+ this.glyphs = glyphs;
+ this.clock = clock;
+ }
+
+ /**
+ * Block glyphs when the stream's encoding can carry them, ASCII otherwise. A cp1252 Windows
+ * console would otherwise render the bar as question marks.
+ */
+ static char[] glyphsFor(PrintStream stream) {
+ try {
+ return stream.charset().newEncoder().canEncode(BLOCK_GLYPHS[0]) ? BLOCK_GLYPHS : ASCII_GLYPHS;
+ } catch (UnsupportedOperationException e) {
+ // A charset that cannot produce an encoder; ASCII is always safe.
+ return ASCII_GLYPHS;
+ }
+ }
+
+ @Override
+ public void item(String label, int index, int total) {
+ if (total > 1) {
+ milestone(String.format("[%d/%d] %s", index + 1, total, label));
+ }
+ }
+
+ @Override
+ public void phase(Phase phase, String detail) {
+ // RESOLVE is instant and DOWNLOAD is announced by started() with the size, so only the local
+ // work worth waiting on gets a line.
+ switch (phase) {
+ case UNZIP -> milestone(" Unpacking...");
+ case ACTIVATE -> milestone(" Installing files...");
+ default -> {
+ // nothing to announce
+ }
+ }
+ }
+
+ @Override
+ public void started(String label, long totalBytes) {
+ long now = clock.millis();
+ transferStartMs = now;
+ windowStartMs = now;
+ windowStartBytes = 0L;
+ speedBytesPerSec = 0L;
+ lastUpdateMs = 0L;
+ transferFinished = false;
+ milestone(
+ totalBytes < 0
+ ? "Downloading: " + label
+ : "Downloading: " + label + " (" + TransferFormat.bytes(totalBytes) + ")");
+ }
+
+ @Override
+ public void transferred(long bytesSoFar, long totalBytes) {
+ if (!interactive || transferFinished) {
+ return;
+ }
+ long now = clock.millis();
+ // The last chunk must never be throttled away: it is what turns the bar into a full 100%, and
+ // it closes the line so the next log statement does not land on top of it. Everything else is
+ // subject to the redraw floor.
+ boolean finished = totalBytes > 0 && bytesSoFar >= totalBytes;
+ if (!finished && now - lastUpdateMs < UPDATE_INTERVAL_MS) {
+ return;
+ }
+ lastUpdateMs = now;
+ updateSpeed(now, bytesSoFar);
+ redraw(progressLine(bytesSoFar, totalBytes));
+ if (finished) {
+ transferFinished = true;
+ endLine();
+ }
+ }
+
+ /** Call when the work has finished so the in-place bar line is closed off. */
+ public void complete() {
+ endLine();
+ }
+
+ /** The line rewritten in place: bar, percentage, then the same read-out the GUI shows. */
+ String progressLine(long bytesSoFar, long totalBytes) {
+ StringBuilder sb = new StringBuilder(" ");
+ if (totalBytes > 0) {
+ // Floor, not round: "100%" alongside "1 sec left" reads as a lie, so only a finished
+ // transfer is allowed to show 100.
+ int percent = (int) Math.min(100L, bytesSoFar * 100L / totalBytes);
+ int filled = percent * BAR_WIDTH / 100;
+ sb.append('[');
+ sb.append(String.valueOf(glyphs[0]).repeat(filled));
+ sb.append(String.valueOf(glyphs[1]).repeat(BAR_WIDTH - filled));
+ sb.append(']').append(String.format("%4d%% ", percent));
+ }
+ if (speedBytesPerSec > 0L) {
+ sb.append(TransferFormat.speed(speedBytesPerSec));
+ if (totalBytes > 0) {
+ sb.append(" - ")
+ .append(TransferFormat.bytes(bytesSoFar))
+ .append(" of ")
+ .append(TransferFormat.bytes(totalBytes));
+ long remaining = totalBytes - bytesSoFar;
+ if (remaining > 0) {
+ sb.append(", ").append(eta(remaining / speedBytesPerSec));
+ }
+ } else {
+ sb.append(" - ").append(TransferFormat.bytes(bytesSoFar)).append(" downloaded");
+ }
+ } else if (totalBytes > 0) {
+ sb.append(TransferFormat.bytes(bytesSoFar))
+ .append(" of ")
+ .append(TransferFormat.bytes(totalBytes));
+ } else {
+ sb.append("starting...");
+ }
+ return sb.toString();
+ }
+
+ private static String eta(long seconds) {
+ TransferFormat.Eta remaining = TransferFormat.eta(seconds);
+ String unit =
+ switch (remaining.unit()) {
+ case SECONDS -> remaining.isSingular() ? "sec" : "secs";
+ case MINUTES -> remaining.isSingular() ? "min" : "mins";
+ case HOURS -> remaining.isSingular() ? "hr" : "hrs";
+ };
+ return remaining.amount() + " " + unit + " left";
+ }
+
+ /**
+ * Bytes per second over the trailing window, falling back to the average over the whole transfer
+ * until the first window closes.
+ */
+ private void updateSpeed(long now, long bytesSoFar) {
+ long windowMs = now - windowStartMs;
+ if (windowMs >= SPEED_WINDOW_MS) {
+ speedBytesPerSec = (bytesSoFar - windowStartBytes) * 1000L / windowMs;
+ windowStartMs = now;
+ windowStartBytes = bytesSoFar;
+ } else if (speedBytesPerSec == 0L) {
+ long elapsed = now - transferStartMs;
+ if (elapsed > 0) {
+ speedBytesPerSec = bytesSoFar * 1000L / elapsed;
+ }
+ }
+ }
+
+ /** A line that always gets printed, on its own row. */
+ private void milestone(String text) {
+ endLine();
+ out.println(text);
+ }
+
+ /**
+ * Overwrite the current line, padding to erase whatever the previous, longer line left behind.
+ */
+ private void redraw(String text) {
+ out.print('\r');
+ out.print(text);
+ if (openLineWidth > text.length()) {
+ out.print(" ".repeat(openLineWidth - text.length()));
+ }
+ out.flush();
+ openLineWidth = text.length();
+ }
+
+ /** Finish any in-place line so following output starts on a fresh row. */
+ private void endLine() {
+ if (openLineWidth > 0) {
+ out.println();
+ out.flush();
+ openLineWidth = 0;
+ }
+ }
+}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/MarketplaceCommand.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/MarketplaceCommand.java
index 0eaeca2ee2..88b09986b6 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/MarketplaceCommand.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/command/MarketplaceCommand.java
@@ -181,9 +181,16 @@ public class MarketplaceCommand implements Runnable, IHopCommand, IHasHopMetadat
} else if (!coordinate.trim().equals(gav.artifactId()) || !coordinate.contains(":")) {
System.out.println("Resolved " + coordinate + " → " + gav.gav());
}
- InstallReceipt receipt =
- new PluginInstaller(log, hopHome, config)
- .install(gav, true, repoId, target.preferredRepoId());
+ ConsoleInstallListener progress = ConsoleInstallListener.forStdOut();
+ InstallReceipt receipt;
+ try {
+ receipt =
+ new PluginInstaller(log, hopHome, config)
+ .install(gav, true, repoId, target.preferredRepoId(), progress);
+ } finally {
+ // Close off the in-place bar line so what follows starts on a fresh row.
+ progress.complete();
+ }
System.out.println(
"Plugin "
+ gav.gav()
@@ -413,7 +420,13 @@ public class MarketplaceCommand implements Runnable, IHopCommand, IHasHopMetadat
Path hopHome = HopHome.resolve();
Path envPath = Path.of(file).toAbsolutePath().normalize();
HopEnvironmentSpec env = HopEnvironmentLoader.load(envPath);
- new EnvironmentApplier(log, hopHome, config).apply(env, prune);
+ ConsoleInstallListener progress = ConsoleInstallListener.forStdOut();
+ try {
+ new EnvironmentApplier(log, hopHome, config).apply(env, prune, progress);
+ } finally {
+ // Close off the in-place bar line so what follows starts on a fresh row.
+ progress.complete();
+ }
System.out.println(
"Environment applied from "
+ envPath
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/env/EnvironmentApplier.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/env/EnvironmentApplier.java
index c48640e37a..1fd94db3b3 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/env/EnvironmentApplier.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/env/EnvironmentApplier.java
@@ -32,6 +32,7 @@ import org.apache.hop.marketplace.catalog.OptionalPluginInfo;
import org.apache.hop.marketplace.command.MarketplaceCommand;
import org.apache.hop.marketplace.config.MarketplaceConfig;
import org.apache.hop.marketplace.config.MarketplaceRepository;
+import org.apache.hop.marketplace.install.IInstallListener;
import org.apache.hop.marketplace.install.InstallReceipt;
import org.apache.hop.marketplace.install.PluginInstaller;
import org.apache.hop.marketplace.install.PluginUninstaller;
@@ -101,16 +102,37 @@ public class EnvironmentApplier {
* Install missing plugins/deps; optionally prune marketplace plugins not listed in the env file.
*/
public void apply(HopEnvironmentSpec env, boolean prune) throws HopException {
+ apply(env, prune, IInstallListener.NONE);
+ }
+
+ /**
+ * @param listener receives per-artifact and byte-level progress across the whole batch, and can
+ * cancel between chunks. Pass {@link IInstallListener#NONE} for headless callers.
+ */
+ public void apply(HopEnvironmentSpec env, boolean prune, IInstallListener listener)
+ throws HopException {
+ IInstallListener progress = listener == null ? IInstallListener.NONE : listener;
MarketplaceConfig config = configFromEnv(env);
String defaultVersion = resolveEnvVersion(env);
PluginInstaller installer = new PluginInstaller(log, hopHome, config);
installer.activateAllPending();
+ // Every reference is one batch item, satisfied ones included: they complete instantly and keep
+ // the bar monotonic, which is friendlier than a total that shrinks as we discover what is
+ // already present.
+ int totalItems = nullSafe(env.getPlugins()).size() + nullSafe(env.getDependencies()).size();
+ int itemIndex = 0;
+
Set<String> desiredArtifacts = new HashSet<>();
for (HopEnvironmentSpec.PluginRef ref : nullSafe(env.getPlugins())) {
if (StringUtils.isBlank(ref.getArtifactId())) {
+ itemIndex++;
continue;
}
+ if (progress.isCancelled()) {
+ throw new HopException("Applying the environment was cancelled");
+ }
+ progress.item(ref.getArtifactId(), itemIndex++, totalItems);
desiredArtifacts.add(ref.getArtifactId());
String groupId =
StringUtils.isNotBlank(ref.getGroupId()) ? ref.getGroupId() : config.getGroupId();
@@ -126,7 +148,7 @@ public class EnvironmentApplier {
if (needsInstall) {
MavenCoordinates coords = new MavenCoordinates(groupId, ref.getArtifactId(), version);
log.logBasic("Applying environment: installing " + coords.gav());
- installer.install(coords, true);
+ installer.install(coords, true, null, null, progress);
} else {
log.logBasic("Applying environment: " + ref.getArtifactId() + " already satisfied");
}
@@ -134,8 +156,13 @@ public class EnvironmentApplier {
for (HopEnvironmentSpec.DependencyRef dep : nullSafe(env.getDependencies())) {
if (StringUtils.isAnyBlank(dep.getGroupId(), dep.getArtifactId(), dep.getVersion())) {
+ itemIndex++;
continue;
}
+ if (progress.isCancelled()) {
+ throw new HopException("Applying the environment was cancelled");
+ }
+ progress.item(dep.getArtifactId(), itemIndex++, totalItems);
String target = StringUtils.defaultIfBlank(dep.getTarget(), "lib/jdbc");
Path dir = hopHome.resolve(target);
Path jar = dir.resolve(dep.getArtifactId() + "-" + dep.getVersion() + ".jar");
@@ -157,10 +184,12 @@ public class EnvironmentApplier {
+ coords.version()
+ ".jar";
log.logBasic("Downloading dependency " + coords.gav() + " → " + target);
+ progress.phase(IInstallListener.Phase.DOWNLOAD, target);
try {
Files.createDirectories(dir);
new MavenRepositoryClient(log)
- .downloadArtifact(config.primaryRepository(), relativePath, coords.gav(), jar);
+ .downloadArtifact(
+ config.primaryRepository(), relativePath, coords.gav(), jar, progress);
} catch (IOException e) {
throw new HopException("Failed to install dependency " + coords.gav(), e);
}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/HopEnvironmentDialog.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/HopEnvironmentDialog.java
index 08c9df8c61..3036c7085c 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/HopEnvironmentDialog.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/HopEnvironmentDialog.java
@@ -17,6 +17,7 @@
package org.apache.hop.marketplace.gui;
+import java.lang.reflect.InvocationTargetException;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -24,11 +25,13 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
+import org.apache.hop.core.IRunnableWithProgress;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.core.variables.Variables;
@@ -48,6 +51,7 @@ import org.apache.hop.ui.core.PropsUi;
import org.apache.hop.ui.core.dialog.BaseDialog;
import org.apache.hop.ui.core.dialog.ErrorDialog;
import org.apache.hop.ui.core.dialog.MessageBox;
+import org.apache.hop.ui.core.dialog.ProgressMonitorDialog;
import org.apache.hop.ui.core.gui.GuiResource;
import org.apache.hop.ui.core.gui.HopNamespace;
import org.apache.hop.ui.core.gui.WindowProperty;
@@ -1049,22 +1053,71 @@ public class HopEnvironmentDialog extends Dialog {
return;
}
HopEnvironmentSpec env = HopEnvironmentLoader.load(currentPath);
- new EnvironmentApplier(log, hopHome, live).apply(env, prune);
+ if (!runApply(env, live, prune)) {
+ notifyStatus(BaseMessages.getString(PKG, "MarketplaceDialog.Status.Cancelled"));
+ return;
+ }
MessageBox box = new MessageBox(shell, SWT.OK | SWT.ICON_INFORMATION);
box.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Apply.Done.Header"));
box.setMessage(
BaseMessages.getString(
PKG, "MarketplaceDialog.Apply.Done.Message", currentPath.toString()));
box.open();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ showApplyError(e);
} catch (Exception e) {
- new ErrorDialog(
- shell,
- BaseMessages.getString(PKG, "HopEnvironmentDialog.Error.Header"),
- BaseMessages.getString(PKG, "MarketplaceDialog.Apply.Error"),
- e);
+ showApplyError(e);
}
}
+ private void showApplyError(Exception e) {
+ // ProgressMonitorDialog wraps the real failure; unwrap so the dialog shows the cause.
+ Throwable cause =
+ e instanceof InvocationTargetException ite && ite.getCause() != null ? ite.getCause() : e;
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG, "HopEnvironmentDialog.Error.Header"),
+ BaseMessages.getString(PKG, "MarketplaceDialog.Apply.Error"),
+ cause);
+ }
+
+ /**
+ * Apply the environment on a worker thread behind a cancellable progress dialog. This path can
+ * install many plugins and download many jars, so running it inline froze the GUI for minutes.
+ *
+ * @return true when the environment was applied, false when the user cancelled
+ */
+ private boolean runApply(HopEnvironmentSpec env, MarketplaceConfig live, boolean prune)
+ throws InvocationTargetException, InterruptedException {
+ EnvironmentApplier applier = new EnvironmentApplier(log, hopHome, live);
+ AtomicBoolean cancelled = new AtomicBoolean(false);
+
+ IRunnableWithProgress operation =
+ monitor -> {
+ ProgressMonitorInstallListener listener = new ProgressMonitorInstallListener(monitor);
+ // No item name up front: apply() names each artifact via IInstallListener.item().
+ listener.begin(BaseMessages.getString(PKG, "MarketplaceDialog.Progress.Applying"), null);
+ try {
+ applier.apply(env, prune, listener);
+ } catch (Exception e) {
+ if (!monitor.isCanceled()) {
+ // Throw WITHOUT calling monitor.done(): done() disposes the progress shell and ends
+ // ProgressMonitorDialog's pump loop before it can observe the exception, which would
+ // report a failed apply as a success. Catching Exception rather than HopException
+ // keeps a RuntimeException from killing the worker and hanging the dialog.
+ throw new InvocationTargetException(e, e.getMessage());
+ }
+ cancelled.set(true);
+ }
+ listener.complete();
+ monitor.done();
+ };
+
+ new ProgressMonitorDialog(shell).run(true, operation);
+ return !cancelled.get();
+ }
+
private void populateExtraPlugins(HopEnvironmentSpec env, EnvironmentDrift drift)
throws Exception {
Set<String> desired = new HashSet<>();
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/MarketplaceDialog.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/MarketplaceDialog.java
index 2da895b970..fdfa581a7b 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/MarketplaceDialog.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/MarketplaceDialog.java
@@ -17,11 +17,15 @@
package org.apache.hop.marketplace.gui;
+import java.lang.reflect.InvocationTargetException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
+import org.apache.hop.core.IRunnableWithProgress;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.core.variables.Variables;
@@ -41,6 +45,7 @@ import org.apache.hop.marketplace.resolve.MavenCoordinates;
import org.apache.hop.ui.core.PropsUi;
import org.apache.hop.ui.core.dialog.ErrorDialog;
import org.apache.hop.ui.core.dialog.MessageBox;
+import org.apache.hop.ui.core.dialog.ProgressMonitorDialog;
import org.apache.hop.ui.core.gui.GuiResource;
import org.apache.hop.ui.core.gui.WindowProperty;
import org.apache.hop.ui.core.widget.ColumnInfo;
@@ -51,7 +56,6 @@ import org.eclipse.swt.custom.CTabFolder;
import org.eclipse.swt.custom.CTabItem;
import org.eclipse.swt.events.ShellAdapter;
import org.eclipse.swt.events.ShellEvent;
-import org.eclipse.swt.graphics.Cursor;
import org.eclipse.swt.layout.FormAttachment;
import org.eclipse.swt.layout.FormData;
import org.eclipse.swt.layout.FormLayout;
@@ -69,6 +73,19 @@ public class MarketplaceDialog extends Dialog {
private static final Class<?> PKG = MarketplaceGuiPlugin.class;
+ /** Quiet period after the last keystroke before the search box triggers a discovery call. */
+ private static final int SEARCH_DEBOUNCE_MS = 350;
+
+ /** Why the plugin table is being reloaded, which decides how the wait is presented. */
+ private enum RefreshReason {
+ /** Refresh button — the user is waiting, so show a cancellable progress dialog. */
+ USER_REFRESH,
+ /** Search box — background query, "Searching…" in the status line, no modal. */
+ SEARCH,
+ /** Dialog open, or after an install/uninstall — background query, status line untouched. */
+ AFTER_CHANGE
+ }
+
private final PropsUi props;
private final ILogChannel log = new LogChannel("Marketplace");
@@ -80,6 +97,12 @@ public class MarketplaceDialog extends Dialog {
private MarketplaceConfig config;
private MarketplaceRepositoriesPanel repositoriesPanel;
+ /** Incremented per discovery request so a slow, superseded answer can be discarded. */
+ private int queryGeneration;
+
+ private boolean searchPending;
+ private long searchRequestedAt;
+
public MarketplaceDialog(Shell parent) {
super(parent, SWT.DIALOG_TRIM | SWT.APPLICATION_MODAL | SWT.RESIZE | SWT.MAX);
this.props = PropsUi.getInstance();
@@ -154,7 +177,7 @@ public class MarketplaceDialog extends Dialog {
}
});
- refreshTable(false);
+ refreshTable(RefreshReason.AFTER_CHANGE);
BaseTransformDialog.setSize(shell);
shell.open();
@@ -188,7 +211,7 @@ public class MarketplaceDialog extends Dialog {
Button wRefresh = new Button(comp, SWT.PUSH);
wRefresh.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Button.Refresh"));
- wRefresh.addListener(SWT.Selection, e -> refreshTable(true));
+ wRefresh.addListener(SWT.Selection, e -> refreshTable(RefreshReason.USER_REFRESH));
BaseTransformDialog.positionBottomButtons(
comp, new Button[] {wInstall, wUninstall, wRefresh}, PropsUi.getMargin(), null);
@@ -219,7 +242,7 @@ public class MarketplaceDialog extends Dialog {
fdSearch.top = new FormAttachment(wlSearch, 0, SWT.CENTER);
fdSearch.right = new FormAttachment(wClearSearch, -PropsUi.getMargin());
wSearch.setLayoutData(fdSearch);
- wSearch.addListener(SWT.Modify, e -> refreshTable(false));
+ wSearch.addListener(SWT.Modify, e -> scheduleSearch());
Label wSearchSep = new Label(comp, SWT.HORIZONTAL | SWT.SEPARATOR);
FormData fdSearchSep = new FormData();
@@ -353,70 +376,175 @@ public class MarketplaceDialog extends Dialog {
/**
* Reload the plugins list.
*
- * @param userRefresh when true (Refresh button), prompt to save dirty repositories and show a
- * wait cursor; search filter updates pass false.
+ * <p>Discovery is a live Nexus search over every browse-enabled repository, so it never runs on
+ * the UI thread: {@link RefreshReason#USER_REFRESH} gets a cancellable progress dialog, the other
+ * reasons run quietly in the background and repopulate the table when they land.
*/
- private void refreshTable(boolean userRefresh) {
- if (userRefresh && !ensureRepositoriesSavedForRefresh()) {
+ private void refreshTable(RefreshReason reason) {
+ if (reason == RefreshReason.USER_REFRESH && !ensureRepositoriesSavedForRefresh()) {
return;
}
-
- Cursor waitCursor = null;
- if (userRefresh && shell != null && !shell.isDisposed()) {
- waitCursor = shell.getDisplay().getSystemCursor(SWT.CURSOR_WAIT);
- shell.setCursor(waitCursor);
+ // Keep shared in-memory config (after optional save/discard). Do not reload from disk on
+ // every search keystroke — that would drop unsaved repository edits.
+ if (repositoriesPanel != null) {
+ config = repositoriesPanel.getConfig();
+ }
+ // Filter only when more than 2 characters are typed (same as explorer/metadata search).
+ String searchText = wSearch != null ? wSearch.getText() : "";
+ String filter = searchText != null && searchText.trim().length() > 2 ? searchText.trim() : null;
+ MarketplaceConfig snapshot = config;
+
+ if (reason == RefreshReason.USER_REFRESH) {
+ queryWithProgress(filter, snapshot);
+ } else {
+ queryInBackground(filter, snapshot, reason == RefreshReason.SEARCH);
}
+ }
+
+ /** Refresh button: modal, cancellable, because the user explicitly asked and is waiting. */
+ private void queryWithProgress(String filter, MarketplaceConfig snapshot) {
+ List<OptionalPluginInfo> found = new ArrayList<>();
+ IRunnableWithProgress operation =
+ monitor -> {
+ // Discovery cannot report intermediate progress, so this is a one-step task: the bar
+ // shows "busy", the label says what for.
+ monitor.beginTask(BaseMessages.getString(PKG, "MarketplaceDialog.Progress.Searching"), 1);
+ try {
+ found.addAll(PluginDiscovery.query(filter, null, snapshot, log));
+ } catch (Exception e) {
+ // No monitor.done() before throwing — see the note in runInstall().
+ throw new InvocationTargetException(e, e.getMessage());
+ }
+ monitor.worked(1);
+ monitor.done();
+ };
try {
- // Keep shared in-memory config (after optional save/discard). Do not reload from disk on
- // every search keystroke — that would drop unsaved repository edits.
- if (repositoriesPanel != null) {
- config = repositoriesPanel.getConfig();
- }
- wTable.table.removeAll();
- // Filter only when more than 2 characters are typed (same as explorer/metadata search).
- String searchText = wSearch != null ? wSearch.getText() : "";
- String filter =
- searchText != null && searchText.trim().length() > 2 ? searchText.trim() : null;
- // Apache optional catalog + live list from every browse=true repository
- List<OptionalPluginInfo> plugins = PluginDiscovery.query(filter, null, config, log);
- for (OptionalPluginInfo info : plugins) {
- // Column 0 is the (hidden) index column; data columns are 1-based.
- TableItem item = new TableItem(wTable.table, SWT.NONE);
- String name = Const.NVL(info.getName(), info.getArtifactId());
- item.setText(1, Const.NVL(name, ""));
- item.setText(2, Const.NVL(info.getVersion(), ""));
- item.setText(3, Const.NVL(info.getArtifactId(), ""));
- item.setText(4, Const.NVL(info.getCategory(), ""));
- item.setText(5, repositoryDisplayName(info.getSource()));
- boolean onDisk = OptionalPluginCatalog.isInstalledOnDisk(hopHome, info);
- InstallReceipt receipt = null;
- try {
- receipt = PluginInstaller.readReceipt(hopHome, info.getArtifactId());
- } catch (Exception e) {
- log.logError("Unable to read receipt for " + info.getArtifactId(), e);
- }
- String status;
- if (onDisk && receipt != null) {
- status =
- BaseMessages.getString(PKG, "MarketplaceDialog.Status.Installed")
- + " ("
- + receipt.getVersion()
- + ")";
- } else if (onDisk) {
- status = BaseMessages.getString(PKG, "MarketplaceDialog.Status.Present");
- } else {
- status = BaseMessages.getString(PKG, "MarketplaceDialog.Status.NotInstalled");
- }
- item.setText(6, status);
- item.setText(7, Const.NVL(info.getDescription(), ""));
- item.setData(info);
+ new ProgressMonitorDialog(shell).run(true, operation);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ } catch (InvocationTargetException e) {
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG, "MarketplaceDialog.Error.Header"),
+ BaseMessages.getString(PKG, "MarketplaceDialog.Refresh.Error"),
+ e.getCause() == null ? e : e.getCause());
+ return;
+ }
+ populateTable(found);
+ }
+
+ /**
+ * Search box and post-install refreshes: a modal dialog per keystroke would be worse than the
+ * freeze it replaces, so query on a worker thread and swap the rows in when the answer arrives.
+ *
+ * <p>Results from a superseded query are dropped — with typing debounced there is rarely more
+ * than one in flight, but a slow repository could still finish after a newer query.
+ */
+ private void queryInBackground(String filter, MarketplaceConfig snapshot, boolean showSearching) {
+ Display display = shell.getDisplay();
+ final int generation = ++queryGeneration;
+ if (showSearching && wStatus != null && !wStatus.isDisposed()) {
+ wStatus.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Progress.Searching"));
+ }
+ Thread worker =
+ new Thread(
+ () -> {
+ List<OptionalPluginInfo> found;
+ try {
+ found = PluginDiscovery.query(filter, null, snapshot, log);
+ } catch (Exception e) {
+ log.logError("Marketplace discovery failed", e);
+ found = List.of();
+ }
+ List<OptionalPluginInfo> result = found;
+ display.asyncExec(
+ () -> {
+ // Stale answer, or the dialog closed while we were waiting on the network.
+ if (generation != queryGeneration || shell.isDisposed()) {
+ return;
+ }
+ populateTable(result);
+ if (showSearching && !wStatus.isDisposed()) {
+ wStatus.setText(
+ BaseMessages.getString(PKG, "MarketplaceDialog.Status.RestartHint"));
+ }
+ });
+ },
+ "Hop-Marketplace-Discovery");
+ worker.setDaemon(true);
+ worker.start();
+ }
+
+ /** Fill the table from discovery results. UI thread only. */
+ private void populateTable(List<OptionalPluginInfo> plugins) {
+ wTable.table.removeAll();
+ for (OptionalPluginInfo info : plugins) {
+ // Column 0 is the (hidden) index column; data columns are 1-based.
+ TableItem item = new TableItem(wTable.table, SWT.NONE);
+ String name = Const.NVL(info.getName(), info.getArtifactId());
+ item.setText(1, Const.NVL(name, ""));
+ item.setText(2, Const.NVL(info.getVersion(), ""));
+ item.setText(3, Const.NVL(info.getArtifactId(), ""));
+ item.setText(4, Const.NVL(info.getCategory(), ""));
+ item.setText(5, repositoryDisplayName(info.getSource()));
+ boolean onDisk = OptionalPluginCatalog.isInstalledOnDisk(hopHome, info);
+ InstallReceipt receipt = null;
+ try {
+ receipt = PluginInstaller.readReceipt(hopHome, info.getArtifactId());
+ } catch (Exception e) {
+ log.logError("Unable to read receipt for " + info.getArtifactId(), e);
}
- wTable.optimizeTableView();
- } finally {
- if (userRefresh && shell != null && !shell.isDisposed()) {
- shell.setCursor(null);
+ String status;
+ if (onDisk && receipt != null) {
+ status =
+ BaseMessages.getString(PKG, "MarketplaceDialog.Status.Installed")
+ + " ("
+ + receipt.getVersion()
+ + ")";
+ } else if (onDisk) {
+ status = BaseMessages.getString(PKG, "MarketplaceDialog.Status.Present");
+ } else {
+ status = BaseMessages.getString(PKG, "MarketplaceDialog.Status.NotInstalled");
}
+ item.setText(6, status);
+ item.setText(7, Const.NVL(info.getDescription(), ""));
+ item.setData(info);
}
+ wTable.optimizeTableView();
+ }
+
+ /**
+ * Coalesce keystrokes into one discovery call. Without this, every character typed in the search
+ * box fires a full Nexus search (paged, 60s per request) against every repository.
+ */
+ private void scheduleSearch() {
+ if (shell == null || shell.isDisposed()) {
+ return;
+ }
+ searchRequestedAt = System.currentTimeMillis();
+ if (searchPending) {
+ return;
+ }
+ searchPending = true;
+ shell.getDisplay().timerExec(SEARCH_DEBOUNCE_MS, this::fireDebouncedSearch);
+ }
+
+ private void fireDebouncedSearch() {
+ if (shell == null || shell.isDisposed()) {
+ searchPending = false;
+ return;
+ }
+ long quietFor = System.currentTimeMillis() - searchRequestedAt;
+ if (quietFor < SEARCH_DEBOUNCE_MS) {
+ // Still typing — check again once the remaining quiet period has elapsed.
+ shell
+ .getDisplay()
+ .timerExec((int) (SEARCH_DEBOUNCE_MS - quietFor), this::fireDebouncedSearch);
+ return;
+ }
+ searchPending = false;
+ refreshTable(RefreshReason.SEARCH);
}
/**
@@ -524,26 +652,88 @@ public class MarketplaceDialog extends Dialog {
StringUtils.isNotBlank(info.getSource()) && !"apache".equalsIgnoreCase(info.getSource())
? info.getSource()
: null;
+ // Same name the table shows, not the Maven coordinate.
+ String displayName = Const.NVL(info.getName(), info.getArtifactId());
wStatus.setText(
BaseMessages.getString(PKG, "MarketplaceDialog.Status.Installing", coords.gav()));
shell.update();
- new PluginInstaller(log, hopHome, config).install(coords, true, null, preferredRepo);
+ if (!runInstall(coords, preferredRepo, displayName)) {
+ // Cancelled by the user — no success popup, and the table still reflects reality.
+ refreshTable(RefreshReason.AFTER_CHANGE);
+ wStatus.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Status.Cancelled"));
+ return;
+ }
MessageBox box = new MessageBox(shell, SWT.OK | SWT.ICON_INFORMATION);
box.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Install.Done.Header"));
box.setMessage(
BaseMessages.getString(PKG, "MarketplaceDialog.Install.Done.Message", coords.gav()));
box.open();
- refreshTable(false);
+ refreshTable(RefreshReason.AFTER_CHANGE);
wStatus.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Status.RestartHint"));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ showInstallError(info, e);
} catch (Exception e) {
- new ErrorDialog(
- shell,
- BaseMessages.getString(PKG, "MarketplaceDialog.Error.Header"),
- BaseMessages.getString(PKG, "MarketplaceDialog.Install.Error", info.getArtifactId()),
- e);
+ showInstallError(info, e);
}
}
+ private void showInstallError(OptionalPluginInfo info, Exception e) {
+ // ProgressMonitorDialog wraps the real failure; unwrap so the dialog shows the cause.
+ Throwable cause =
+ e instanceof InvocationTargetException ite && ite.getCause() != null ? ite.getCause() : e;
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG, "MarketplaceDialog.Error.Header"),
+ BaseMessages.getString(PKG, "MarketplaceDialog.Install.Error", info.getArtifactId()),
+ cause);
+ }
+
+ /**
+ * Download and install on a worker thread behind a cancellable progress dialog, so the UI keeps
+ * repainting and the user can see the zip arrive.
+ *
+ * <p>{@link ProgressMonitorDialog#run(boolean, IRunnableWithProgress)} pumps the event loop until
+ * the work finishes and only then returns, so everything after the call is back on the UI thread
+ * and needs no marshalling.
+ *
+ * @return true when the plugin was installed, false when the user cancelled
+ * @throws InvocationTargetException wrapping whatever the install failed with
+ */
+ private boolean runInstall(MavenCoordinates coords, String preferredRepo, String displayName)
+ throws InvocationTargetException, InterruptedException {
+ PluginInstaller installer = new PluginInstaller(log, hopHome, config);
+ AtomicBoolean cancelled = new AtomicBoolean(false);
+ ProgressMonitorDialog dialog = new ProgressMonitorDialog(shell);
+
+ IRunnableWithProgress operation =
+ monitor -> {
+ ProgressMonitorInstallListener listener = new ProgressMonitorInstallListener(monitor);
+ listener.begin(
+ BaseMessages.getString(PKG, "MarketplaceDialog.Progress.Installing", displayName),
+ displayName);
+ try {
+ installer.install(coords, true, null, preferredRepo, listener);
+ } catch (Exception e) {
+ if (!monitor.isCanceled()) {
+ // Throw WITHOUT calling monitor.done(). done() disposes the progress shell, which
+ // ends ProgressMonitorDialog's pump loop before it has a chance to observe this
+ // exception — a failed install would then be reported as a success. Leaving the shell
+ // up lets the pump see the exception and rethrow it. Catching Exception (not just
+ // HopException) is what keeps a RuntimeException from killing the worker thread
+ // silently and hanging the dialog forever.
+ throw new InvocationTargetException(e, e.getMessage());
+ }
+ cancelled.set(true);
+ }
+ listener.complete();
+ monitor.done();
+ };
+
+ dialog.run(true, operation);
+ return !cancelled.get();
+ }
+
private void uninstallSelected() {
OptionalPluginInfo info = selected();
if (info == null) {
@@ -569,7 +759,7 @@ public class MarketplaceDialog extends Dialog {
return;
}
new PluginUninstaller(log, hopHome).uninstall(info.getArtifactId());
- refreshTable(false);
+ refreshTable(RefreshReason.AFTER_CHANGE);
wStatus.setText(BaseMessages.getString(PKG, "MarketplaceDialog.Status.RestartHint"));
} catch (Exception e) {
new ErrorDialog(
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListener.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListener.java
new file mode 100644
index 0000000000..f99baf21b7
--- /dev/null
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListener.java
@@ -0,0 +1,306 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.gui;
+
+import org.apache.hop.core.IProgressMonitor;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.marketplace.install.IInstallListener;
+import org.apache.hop.marketplace.install.TransferFormat;
+
+/**
+ * Drives an {@link IProgressMonitor} from plugin install progress: the bar shows overall percent
+ * across the batch, the subtask line shows the live transfer as "1.8MB/s - 42.0MB of 353MB, 3 mins
+ * left".
+ *
+ * <p>Two things this class exists to get right:
+ *
+ * <ul>
+ * <li><strong>Throttling.</strong> {@code IInstallListener.transferred} fires once per 64KB chunk
+ * — thousands of times for a large plugin — and {@code IProgressMonitor.subTask} posts an
+ * {@code asyncExec} per call. Updates are collapsed to one per {@link #UPDATE_INTERVAL_MS}.
+ * <li><strong>Scaling.</strong> An install is download + unzip + activate, and a batch is N of
+ * those. Bytes alone would peg the bar at 100% while the unzip still runs, so each phase owns
+ * a slice of its item's share.
+ * </ul>
+ *
+ * <p>All methods are called from the download thread. Only {@link IProgressMonitor} is touched, and
+ * that implementation already marshals to the UI thread.
+ */
+public class ProgressMonitorInstallListener implements IInstallListener {
+
+ private static final Class<?> PKG = MarketplaceGuiPlugin.class;
+
+ /** Total ticks reported to the monitor; the bar is driven in whole percent. */
+ public static final int TOTAL_WORK = 100;
+
+ /** Minimum gap between UI updates. Below ~100ms the labels are unreadable anyway. */
+ static final long UPDATE_INTERVAL_MS = 150L;
+
+ /**
+ * Percent-of-item boundaries for the install phases. Download dominates wall-clock time, so it
+ * gets the bulk; the rest keeps the bar moving while jars are written to disk.
+ */
+ private static final int PCT_AFTER_RESOLVE = 5;
+
+ private static final int PCT_AFTER_DOWNLOAD = 85;
+ private static final int PCT_AFTER_UNZIP = 95;
+
+ /** Speed is averaged over this window so the number does not flicker on a bursty link. */
+ private static final long SPEED_WINDOW_MS = 2000L;
+
+ private final IProgressMonitor monitor;
+ private final IClock clock;
+
+ /** Batch position: item {@code itemIndex} of {@code itemCount}. */
+ private int itemIndex;
+
+ private int itemCount = 1;
+
+ /**
+ * Short, user-facing name of the current artifact. The download layer only knows the Maven
+ * coordinate, which is too noisy for a dialog, so the name is supplied from the UI side instead.
+ */
+ private String itemName;
+
+ private Phase phase = Phase.RESOLVE;
+ private long lastUpdateMs;
+
+ /** Percent already reported to the monitor, so {@code worked()} can be fed deltas. */
+ private int reportedPercent;
+
+ private long transferStartMs;
+ private long windowStartMs;
+ private long windowStartBytes;
+ private long lastSpeedBytesPerSec;
+
+ /** Indirection over the clock so throttling and speed can be unit-tested. */
+ public interface IClock {
+ long millis();
+ }
+
+ public ProgressMonitorInstallListener(IProgressMonitor monitor) {
+ this(monitor, System::currentTimeMillis);
+ }
+
+ ProgressMonitorInstallListener(IProgressMonitor monitor, IClock clock) {
+ this.monitor = monitor;
+ this.clock = clock;
+ }
+
+ /**
+ * Call once before the work starts.
+ *
+ * @param taskName the overall operation name, shown on the persistent task line
+ * @param itemName short display name of what is being fetched, used in the download line. May be
+ * null for a batch, where {@link #item(String, int, int)} names each artifact instead.
+ */
+ public void begin(String taskName, String itemName) {
+ this.itemName = itemName;
+ monitor.beginTask(taskName, TOTAL_WORK);
+ }
+
+ @Override
+ public void item(String label, int index, int total) {
+ this.itemName = label;
+ this.itemIndex = index;
+ this.itemCount = Math.max(1, total);
+ this.phase = Phase.RESOLVE;
+ // The batch position belongs on the persistent task line; the subtask line is reused by every
+ // phase and transfer update below, so "3 of 12" would vanish within milliseconds there.
+ monitor.setTaskName(
+ BaseMessages.getString(
+ PKG,
+ "MarketplaceDialog.Progress.Item",
+ label,
+ Integer.toString(index + 1),
+ Integer.toString(this.itemCount)));
+ // Force the bar forward even if this item turns out to be already satisfied.
+ advanceTo(overallPercent(0));
+ }
+
+ @Override
+ public void phase(Phase newPhase, String detail) {
+ this.phase = newPhase;
+ advanceTo(overallPercent(phaseFloor(newPhase)));
+ monitor.subTask(phaseMessage(newPhase, detail));
+ // A new transfer is about to start; reset the speed window.
+ if (newPhase == Phase.DOWNLOAD) {
+ transferStartMs = 0L;
+ }
+ }
+
+ @Override
+ public void started(String label, long totalBytes) {
+ long now = clock.millis();
+ transferStartMs = now;
+ windowStartMs = now;
+ windowStartBytes = 0L;
+ lastSpeedBytesPerSec = 0L;
+ lastUpdateMs = 0L;
+ // `label` is the Maven coordinate (org.apache.hop:hop-x:2.19.0). Show the plugin's name.
+ monitor.subTask(
+ BaseMessages.getString(
+ PKG, "MarketplaceDialog.Progress.Downloading", itemName == null ? label : itemName));
+ }
+
+ @Override
+ public void transferred(long bytesSoFar, long totalBytes) {
+ long now = clock.millis();
+ if (now - lastUpdateMs < UPDATE_INTERVAL_MS) {
+ return;
+ }
+ lastUpdateMs = now;
+
+ updateSpeed(now, bytesSoFar);
+ if (totalBytes > 0) {
+ // Floor, matching the CLI bar: the phase ceiling is only reached on the last chunk.
+ int withinDownload = (int) Math.min(100L, bytesSoFar * 100L / totalBytes);
+ advanceTo(overallPercent(scaleIntoPhase(withinDownload)));
+ }
+ monitor.subTask(transferText(bytesSoFar, totalBytes));
+ }
+
+ /**
+ * The download read-out, laid out like a browser's: {@code 1.7MB/s - 42.0MB of 353MB, 3 mins
+ * left}. Parts that cannot be known yet are left out rather than filled with placeholders, so the
+ * line grows from "Starting download…" into the full form once a speed has been measured.
+ */
+ String transferText(long bytesSoFar, long totalBytes) {
+ boolean sizeKnown = totalBytes > 0;
+ boolean speedKnown = lastSpeedBytesPerSec > 0L;
+
+ if (!speedKnown) {
+ return sizeKnown
+ ? BaseMessages.getString(
+ PKG,
+ "MarketplaceDialog.Progress.TransferNoSpeed",
+ TransferFormat.bytes(bytesSoFar),
+ TransferFormat.bytes(totalBytes))
+ : BaseMessages.getString(PKG, "MarketplaceDialog.Progress.Starting");
+ }
+ if (!sizeKnown) {
+ // No Content-Length: an ETA or a percentage would be invented, so report volume and speed.
+ return BaseMessages.getString(
+ PKG,
+ "MarketplaceDialog.Progress.TransferUnknownSize",
+ TransferFormat.speed(lastSpeedBytesPerSec),
+ TransferFormat.bytes(bytesSoFar));
+ }
+ long remaining = totalBytes - bytesSoFar;
+ if (remaining <= 0) {
+ return BaseMessages.getString(
+ PKG,
+ "MarketplaceDialog.Progress.TransferNoSpeed",
+ TransferFormat.bytes(bytesSoFar),
+ TransferFormat.bytes(totalBytes));
+ }
+ return BaseMessages.getString(
+ PKG,
+ "MarketplaceDialog.Progress.Transfer",
+ TransferFormat.speed(lastSpeedBytesPerSec),
+ TransferFormat.bytes(bytesSoFar),
+ TransferFormat.bytes(totalBytes),
+ formatEta(remaining / lastSpeedBytesPerSec));
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return monitor.isCanceled();
+ }
+
+ /** Call when the work finished successfully to run the bar out to 100%. */
+ public void complete() {
+ advanceTo(TOTAL_WORK);
+ }
+
+ /**
+ * Bytes per second over the trailing {@link #SPEED_WINDOW_MS}, falling back to the average over
+ * the whole transfer until the first window closes.
+ */
+ private void updateSpeed(long now, long bytesSoFar) {
+ long windowMs = now - windowStartMs;
+ if (windowMs >= SPEED_WINDOW_MS) {
+ lastSpeedBytesPerSec = (bytesSoFar - windowStartBytes) * 1000L / windowMs;
+ windowStartMs = now;
+ windowStartBytes = bytesSoFar;
+ } else if (lastSpeedBytesPerSec == 0L) {
+ long elapsed = now - transferStartMs;
+ if (elapsed > 0) {
+ lastSpeedBytesPerSec = bytesSoFar * 1000L / elapsed;
+ }
+ }
+ }
+
+ /** Time remaining in words — "45 secs left", "3 mins left", "2 hrs left". */
+ private String formatEta(long seconds) {
+ TransferFormat.Eta remaining = TransferFormat.eta(seconds);
+ String key =
+ switch (remaining.unit()) {
+ case SECONDS -> remaining.isSingular() ? "Eta.Second" : "Eta.Seconds";
+ case MINUTES -> remaining.isSingular() ? "Eta.Minute" : "Eta.Minutes";
+ case HOURS -> remaining.isSingular() ? "Eta.Hour" : "Eta.Hours";
+ };
+ return BaseMessages.getString(
+ PKG, "MarketplaceDialog.Progress." + key, Long.toString(remaining.amount()));
+ }
+
+ /** Report the delta needed to reach {@code target}; never moves backwards. */
+ private void advanceTo(int target) {
+ int clamped = Math.min(TOTAL_WORK, Math.max(0, target));
+ if (clamped > reportedPercent) {
+ monitor.worked(clamped - reportedPercent);
+ reportedPercent = clamped;
+ }
+ }
+
+ /** Map a 0-100 percent within the current item onto the batch as a whole. */
+ private int overallPercent(int percentWithinItem) {
+ double perItem = TOTAL_WORK / (double) itemCount;
+ return (int) Math.round(itemIndex * perItem + percentWithinItem * perItem / 100.0);
+ }
+
+ /** Where a phase starts, as a percentage of one item. */
+ private static int phaseFloor(Phase phase) {
+ return switch (phase) {
+ case RESOLVE -> 0;
+ case DOWNLOAD -> PCT_AFTER_RESOLVE;
+ case UNZIP -> PCT_AFTER_DOWNLOAD;
+ case ACTIVATE -> PCT_AFTER_UNZIP;
+ };
+ }
+
+ /** Spread a 0-100 download percentage across the download phase's slice of one item. */
+ private int scaleIntoPhase(int withinDownload) {
+ if (phase != Phase.DOWNLOAD) {
+ return phaseFloor(phase);
+ }
+ int span = PCT_AFTER_DOWNLOAD - PCT_AFTER_RESOLVE;
+ return PCT_AFTER_RESOLVE + withinDownload * span / 100;
+ }
+
+ private String phaseMessage(Phase phase, String detail) {
+ String key =
+ switch (phase) {
+ case RESOLVE -> "MarketplaceDialog.Progress.Phase.Resolve";
+ case DOWNLOAD -> "MarketplaceDialog.Progress.Phase.Download";
+ case UNZIP -> "MarketplaceDialog.Progress.Phase.Unzip";
+ case ACTIVATE -> "MarketplaceDialog.Progress.Phase.Activate";
+ };
+ return BaseMessages.getString(PKG, key, detail == null ? "" : detail);
+ }
+}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/IInstallListener.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/IInstallListener.java
new file mode 100644
index 0000000000..05062f048c
--- /dev/null
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/IInstallListener.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.install;
+
+import org.apache.hop.marketplace.resolve.ITransferListener;
+
+/**
+ * Progress of a plugin install: which {@link Phase} it is in, plus the byte-level download progress
+ * inherited from {@link ITransferListener}.
+ *
+ * <p>An install is more than a download — resolving a SNAPSHOT costs a round trip, and unzipping a
+ * few hundred jars costs real time — so a progress bar driven by bytes alone would sit at 100%
+ * while the install looks hung. Phases let a listener spread the bar over the whole operation.
+ *
+ * <p>Like its parent, callbacks arrive off the UI thread.
+ */
+public interface IInstallListener extends ITransferListener {
+
+ /** A listener that reports nothing and never cancels. */
+ IInstallListener NONE = new IInstallListener() {};
+
+ /** Stages of a single plugin install, in the order they occur. */
+ enum Phase {
+ /** Contacting repositories, resolving a SNAPSHOT to a unique version. */
+ RESOLVE,
+ /** Streaming the plugin zip to {@code plugins/.staging/.download/}. */
+ DOWNLOAD,
+ /** Expanding the zip into {@code plugins/.staging/<artifactId>/}. */
+ UNZIP,
+ /** Copying staged files into the live Hop installation. */
+ ACTIVATE
+ }
+
+ /**
+ * The install moved to a new phase.
+ *
+ * @param phase the phase now starting
+ * @param detail extra context for the user (repository name, entry count), may be null
+ */
+ default void phase(Phase phase, String detail) {
+ // no-op
+ }
+
+ /**
+ * One of several artifacts in a batch is starting. Called by {@code EnvironmentApplier} so a
+ * listener can show "3 of 12" and scale the bar across the batch. Never called for a single
+ * install, which is equivalent to {@code item(label, 0, 1)}.
+ *
+ * @param label what is being installed
+ * @param index zero-based position in the batch
+ * @param total batch size
+ */
+ default void item(String label, int index, int total) {
+ // no-op
+ }
+}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/PluginInstaller.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/PluginInstaller.java
index 9bbb7e05c7..f7b635ecc2 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/PluginInstaller.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/PluginInstaller.java
@@ -102,19 +102,45 @@ public class PluginInstaller {
String forceRepoId,
String preferredRepoId)
throws HopException {
+ return install(
+ coordinates, activateImmediately, forceRepoId, preferredRepoId, IInstallListener.NONE);
+ }
+
+ /**
+ * @param listener receives phase changes and download byte counts, and can cancel the install
+ * between chunks. Pass {@link IInstallListener#NONE} for headless callers.
+ */
+ public InstallReceipt install(
+ MavenCoordinates coordinates,
+ boolean activateImmediately,
+ String forceRepoId,
+ String preferredRepoId,
+ IInstallListener listener)
+ throws HopException {
+ IInstallListener progress = listener == null ? IInstallListener.NONE : listener;
Path downloadDir = hopHome.resolve(STAGING_DIR).resolve(".download");
Path zipFile =
downloadDir.resolve(coordinates.artifactId() + "-" + coordinates.version() + ".zip");
try {
+ progress.phase(IInstallListener.Phase.RESOLVE, coordinates.gav());
List<MarketplaceRepository> repos = resolveRepositories(forceRepoId, preferredRepoId);
MarketplaceRepository used = null;
List<String> errors = new ArrayList<>();
for (MarketplaceRepository repo : repos) {
+ if (progress.isCancelled()) {
+ throw new HopException("Install of " + coordinates.gav() + " was cancelled");
+ }
try {
- client.downloadZip(repo, coordinates, zipFile);
+ progress.phase(IInstallListener.Phase.DOWNLOAD, repo.displayName());
+ client.downloadZip(repo, coordinates, zipFile, progress);
used = repo;
break;
} catch (HopException e) {
+ // A cancel is a user decision, not a repository failure — do not fall through to the
+ // next repository and start the whole download again.
+ if (progress.isCancelled()) {
+ throw e;
+ }
errors.add(
repo.getId()
+ " @ "
@@ -132,6 +158,7 @@ public class PluginInstaller {
+ String.join("\n - ", errors));
}
+ progress.phase(IInstallListener.Phase.UNZIP, coordinates.artifactId());
Path stageRoot = hopHome.resolve(STAGING_DIR).resolve(coordinates.artifactId());
deleteRecursive(stageRoot);
Files.createDirectories(stageRoot);
@@ -149,6 +176,7 @@ public class PluginInstaller {
writeReceipt(receipt);
if (activateImmediately) {
+ progress.phase(IInstallListener.Phase.ACTIVATE, coordinates.artifactId());
activateStaged(coordinates.artifactId(), relativePaths);
receipt.setPendingActivation(false);
writeReceipt(receipt);
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/TransferFormat.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/TransferFormat.java
new file mode 100644
index 0000000000..0b2a82c2c7
--- /dev/null
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/install/TransferFormat.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.install;
+
+import java.util.Locale;
+
+/**
+ * Shared number formatting for download progress, so the GUI dialog and the CLI report the same
+ * figures the same way.
+ *
+ * <p>Byte counts come out as symbols ({@code 42.0MB}), which need no translation. Time remaining is
+ * returned as an {@link Eta} value rather than a string, because the GUI words it through its
+ * message bundle while the CLI prints plain English — only the rounding rules are shared.
+ */
+public final class TransferFormat {
+
+ private static final String[] BYTE_UNITS = {"KB", "MB", "GB", "TB"};
+
+ private TransferFormat() {}
+
+ /** Unit an {@link Eta} is expressed in. */
+ public enum EtaUnit {
+ SECONDS,
+ MINUTES,
+ HOURS
+ }
+
+ /**
+ * Time remaining, already rounded to the coarsest useful unit.
+ *
+ * @param amount always at least 1
+ * @param unit which unit {@code amount} counts
+ */
+ public record Eta(long amount, EtaUnit unit) {
+ /** True when the amount needs a singular noun ("1 min" rather than "1 mins"). */
+ public boolean isSingular() {
+ return amount == 1;
+ }
+ }
+
+ /**
+ * A byte count the way a browser shows one: one decimal below 100, none above, so the width stays
+ * roughly stable as the number climbs — "42.0MB", then "353MB".
+ *
+ * <p>Hand-rolled rather than {@code StorageUnitConverter}, which rounds to whole units ("42MB")
+ * and so makes a slow download look frozen for seconds at a time.
+ */
+ public static String bytes(long bytes) {
+ if (bytes < 1024L) {
+ return bytes + "B";
+ }
+ double value = bytes;
+ int unit = -1;
+ do {
+ value /= 1024.0;
+ unit++;
+ } while (value >= 1024.0 && unit < BYTE_UNITS.length - 1);
+ return decimals(value) + BYTE_UNITS[unit];
+ }
+
+ public static String speed(long bytesPerSecond) {
+ return bytes(bytesPerSecond) + "/s";
+ }
+
+ /**
+ * Locale.ROOT deliberately: a technical size read-out uses a dot, as every download tool does.
+ * The default locale would render "17,6MB" on a comma-decimal machine, and worse, group thousands
+ * with a dot — making "1.750MB" ambiguous between 1750 and 1.75.
+ */
+ private static String decimals(double value) {
+ return value < 100.0
+ ? String.format(Locale.ROOT, "%.1f", value)
+ : String.format(Locale.ROOT, "%.0f", Math.floor(value));
+ }
+
+ /**
+ * Round a number of seconds to the coarsest unit that still says something useful. A precise
+ * mm:ss countdown reads as a promise the network cannot keep.
+ */
+ public static Eta eta(long seconds) {
+ if (seconds < 60) {
+ return new Eta(Math.max(1, seconds), EtaUnit.SECONDS);
+ }
+ long minutes = Math.round(seconds / 60.0);
+ if (minutes < 60) {
+ return new Eta(minutes, EtaUnit.MINUTES);
+ }
+ return new Eta(Math.round(minutes / 60.0), EtaUnit.HOURS);
+ }
+}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/ITransferListener.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/ITransferListener.java
new file mode 100644
index 0000000000..64f78684b8
--- /dev/null
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/ITransferListener.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.resolve;
+
+/**
+ * Byte-level progress of a single HTTP download, so callers can show a moving progress bar and
+ * cancel a transfer in flight.
+ *
+ * <p>Deliberately free of any UI type: the CLI can log lines, the GUI adapts it to an {@code
+ * IProgressMonitor}. Every method has a no-op default so {@link #NONE} and lambdas stay cheap.
+ *
+ * <p>Callbacks arrive on the thread doing the download, which is <strong>not</strong> the SWT UI
+ * thread. Implementations that touch widgets must marshal, and should throttle: {@link
+ * #transferred(long, long)} fires once per network chunk.
+ */
+public interface ITransferListener {
+
+ /** A listener that reports nothing and never cancels. */
+ ITransferListener NONE = new ITransferListener() {};
+
+ /**
+ * A transfer is about to begin.
+ *
+ * @param label human-readable name of what is being fetched (a GAV, a file name)
+ * @param totalBytes size from {@code Content-Length}, or -1 when the server does not say (chunked
+ * or compressed transfer). Listeners should switch to an indeterminate display for -1.
+ */
+ default void started(String label, long totalBytes) {
+ // no-op
+ }
+
+ /**
+ * Progress of the current transfer.
+ *
+ * @param bytesSoFar bytes written to the target file so far
+ * @param totalBytes same value as passed to {@link #started(String, long)}, or -1 if unknown
+ */
+ default void transferred(long bytesSoFar, long totalBytes) {
+ // no-op
+ }
+
+ /**
+ * Polled between chunks. Returning true aborts the download; the partially written file is
+ * removed and the caller sees a {@code HopException}.
+ */
+ default boolean isCancelled() {
+ return false;
+ }
+}
diff --git a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/MavenRepositoryClient.java b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/MavenRepositoryClient.java
index 56247c2773..5d52bd977a 100644
--- a/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/MavenRepositoryClient.java
+++ b/plugins/misc/marketplace/src/main/java/org/apache/hop/marketplace/resolve/MavenRepositoryClient.java
@@ -19,6 +19,7 @@ package org.apache.hop.marketplace.resolve;
import java.io.IOException;
import java.io.InputStream;
+import java.io.OutputStream;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
@@ -26,7 +27,6 @@ import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Base64;
@@ -41,6 +41,12 @@ import org.apache.hop.marketplace.config.MarketplaceRepository;
/** Downloads Maven layout artifacts over HTTP(S), with optional Basic authentication. */
public class MavenRepositoryClient {
+ /**
+ * Read chunk for the download loop. Small enough that cancel feels immediate on a slow link,
+ * large enough not to add syscall overhead on a fast one.
+ */
+ private static final int TRANSFER_BUFFER_SIZE = 64 * 1024;
+
private static final Pattern SNAPSHOT_ZIP_VALUE =
Pattern.compile(
"<snapshotVersion>\\s*<extension>zip</extension>\\s*<value>([^<]+)</value>",
@@ -70,10 +76,19 @@ public class MavenRepositoryClient {
public Path downloadZip(
MarketplaceRepository repository, MavenCoordinates coordinates, Path targetFile)
throws HopException {
+ return downloadZip(repository, coordinates, targetFile, ITransferListener.NONE);
+ }
+
+ public Path downloadZip(
+ MarketplaceRepository repository,
+ MavenCoordinates coordinates,
+ Path targetFile,
+ ITransferListener listener)
+ throws HopException {
String base = repository.normalizedUrl();
String relative = resolveZipRelativePath(repository, coordinates);
String url = base + relative;
- return download(url, repository, coordinates.gav(), targetFile);
+ return download(url, repository, coordinates.gav(), targetFile, listener);
}
/**
@@ -130,9 +145,19 @@ public class MavenRepositoryClient {
public void downloadArtifact(
MarketplaceRepository repository, String relativePath, String label, Path targetFile)
throws HopException {
+ downloadArtifact(repository, relativePath, label, targetFile, ITransferListener.NONE);
+ }
+
+ public void downloadArtifact(
+ MarketplaceRepository repository,
+ String relativePath,
+ String label,
+ Path targetFile,
+ ITransferListener listener)
+ throws HopException {
String base = repository.normalizedUrl();
String url = base + (relativePath.startsWith("/") ? relativePath.substring(1) : relativePath);
- download(url, repository, label, targetFile);
+ download(url, repository, label, targetFile, listener);
}
/**
@@ -238,9 +263,15 @@ public class MavenRepositoryClient {
}
}
- private Path download(String url, MarketplaceRepository repository, String label, Path targetFile)
+ private Path download(
+ String url,
+ MarketplaceRepository repository,
+ String label,
+ Path targetFile,
+ ITransferListener listener)
throws HopException {
log.logBasic("Downloading " + label + " from " + url);
+ ITransferListener progress = listener == null ? ITransferListener.NONE : listener;
try {
Files.createDirectories(targetFile.getParent());
HttpRequest.Builder builder =
@@ -255,16 +286,74 @@ public class MavenRepositoryClient {
throw new HopException(
"HTTP " + response.statusCode() + " downloading " + label + " from " + url);
}
- try (InputStream in = response.body()) {
- Files.copy(in, targetFile, StandardCopyOption.REPLACE_EXISTING);
- }
- log.logBasic("Downloaded " + targetFile + " (" + Files.size(targetFile) + " bytes)");
+ long totalBytes = contentLength(response);
+ progress.started(label, totalBytes);
+ long written = copyWithProgress(response.body(), targetFile, totalBytes, progress);
+ log.logBasic("Downloaded " + targetFile + " (" + written + " bytes)");
return targetFile;
} catch (IOException | InterruptedException e) {
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
+ deleteQuietly(targetFile);
throw new HopException("Failed to download " + label + " from " + url, e);
+ } catch (HopException e) {
+ // Cancelled or rejected mid-transfer: never leave a truncated zip behind for the unzip step.
+ deleteQuietly(targetFile);
+ throw e;
+ }
+ }
+
+ /**
+ * Stream the response body to {@code targetFile}, reporting every chunk so the caller can move a
+ * progress bar, and honouring cancellation between chunks.
+ *
+ * <p>This is why {@code Files.copy(in, target)} is not used: it consumes the whole stream in one
+ * call, leaving no hook for progress or cancel.
+ *
+ * @return the number of bytes written
+ */
+ private static long copyWithProgress(
+ InputStream body, Path targetFile, long totalBytes, ITransferListener listener)
+ throws IOException, HopException {
+ byte[] buffer = new byte[TRANSFER_BUFFER_SIZE];
+ long written = 0;
+ try (InputStream in = body;
+ OutputStream out = Files.newOutputStream(targetFile)) {
+ int read;
+ while ((read = in.read(buffer)) != -1) {
+ if (listener.isCancelled()) {
+ throw new HopException("Download cancelled");
+ }
+ out.write(buffer, 0, read);
+ written += read;
+ listener.transferred(written, totalBytes);
+ }
+ }
+ // One last callback carrying the exact byte count. A throttling listener must let this one
+ // through rather than coalesce it away, or its bar freezes just short of the total.
+ listener.transferred(written, totalBytes);
+ return written;
+ }
+
+ /**
+ * Size of the body from {@code Content-Length}, or -1 when the server does not say — chunked
+ * transfer encoding, or a compressed response whose header describes the encoded length rather
+ * than the file. Callers must treat -1 as "show an indeterminate bar", not as an error.
+ */
+ private static long contentLength(HttpResponse<InputStream> response) {
+ if (response.headers().firstValue("content-encoding").isPresent()) {
+ // Content-Length then counts encoded bytes, which would not match what we write to disk.
+ return -1L;
+ }
+ return response.headers().firstValueAsLong("content-length").orElse(-1L);
+ }
+
+ private void deleteQuietly(Path file) {
+ try {
+ Files.deleteIfExists(file);
+ } catch (IOException e) {
+ log.logDetailed("Unable to remove partial download " + file + ": " + e.getMessage());
}
}
diff --git a/plugins/misc/marketplace/src/main/resources/org/apache/hop/marketplace/gui/messages/messages_en_US.properties b/plugins/misc/marketplace/src/main/resources/org/apache/hop/marketplace/gui/messages/messages_en_US.properties
index ed67006ff2..2e893c857e 100644
--- a/plugins/misc/marketplace/src/main/resources/org/apache/hop/marketplace/gui/messages/messages_en_US.properties
+++ b/plugins/misc/marketplace/src/main/resources/org/apache/hop/marketplace/gui/messages/messages_en_US.properties
@@ -82,6 +82,28 @@ MarketplaceDialog.Status.Installing=Installing {0}...
MarketplaceDialog.Status.Validating=Validating environment file...
MarketplaceDialog.Status.Applying=Applying environment file...
MarketplaceDialog.Status.Drift=Environment drift detected — use Apply to fix
+MarketplaceDialog.Status.Cancelled=Cancelled.
+MarketplaceDialog.Progress.Installing=Installing {0}
+MarketplaceDialog.Progress.Applying=Applying environment file
+MarketplaceDialog.Progress.Searching=Searching repositories...
+MarketplaceDialog.Progress.Item={0} ({1} of {2})
+MarketplaceDialog.Progress.Phase.Resolve=Resolving {0}...
+MarketplaceDialog.Progress.Phase.Download=Contacting {0}...
+MarketplaceDialog.Progress.Phase.Unzip=Unpacking {0}...
+MarketplaceDialog.Progress.Phase.Activate=Installing files for {0}...
+MarketplaceDialog.Progress.Downloading=Downloading: {0}
+# Download read-out, laid out like a browser's: 1.7MB/s - 42.0MB of 353MB, 3 mins left
+MarketplaceDialog.Progress.Transfer={0} - {1} of {2}, {3}
+MarketplaceDialog.Progress.TransferNoSpeed={0} of {1}
+MarketplaceDialog.Progress.TransferUnknownSize={0} - {1} downloaded
+MarketplaceDialog.Progress.Starting=Starting download...
+MarketplaceDialog.Progress.Eta.Second={0} sec left
+MarketplaceDialog.Progress.Eta.Seconds={0} secs left
+MarketplaceDialog.Progress.Eta.Minute={0} min left
+MarketplaceDialog.Progress.Eta.Minutes={0} mins left
+MarketplaceDialog.Progress.Eta.Hour={0} hr left
+MarketplaceDialog.Progress.Eta.Hours={0} hrs left
+MarketplaceDialog.Refresh.Error=Failed to list plugins from the configured repositories
MarketplaceDialog.Error.Header=Marketplace error
MarketplaceDialog.Error.ResolveHome=Could not determine the Hop installation directory. Start Hop from the install root using the hop launcher.
MarketplaceDialog.Install.Done.Header=Plugin installed
diff --git a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/command/ConsoleInstallListenerTest.java b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/command/ConsoleInstallListenerTest.java
new file mode 100644
index 0000000000..11c8136711
--- /dev/null
+++ b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/command/ConsoleInstallListenerTest.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.command;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+import org.apache.hop.marketplace.install.IInstallListener;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The CLI progress bar: layout, throttling, in-place redraw, and the redirected-output fallback.
+ */
+class ConsoleInstallListenerTest {
+
+ private static final char[] ASCII = {'#', '-'};
+
+ /** Manually advanced clock so throttling is deterministic. */
+ private static class FakeClock implements ConsoleInstallListener.IClock {
+ private long now;
+
+ @Override
+ public long millis() {
+ return now;
+ }
+ }
+
+ private static class Capture {
+ private final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ private final PrintStream stream = new PrintStream(bytes, true, StandardCharsets.UTF_8);
+
+ private String text() {
+ return bytes.toString(StandardCharsets.UTF_8);
+ }
+ }
+
+ @Test
+ void drawsABarWithPercentSpeedAndEta() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+
+ listener.started("Data Vault", 353L * 1024 * 1024);
+ // 42MB in 24s ≈ 1.75MB/s, leaving 311MB ≈ 178s ≈ 3 mins.
+ clock.now += 24_000;
+ listener.transferred(42L * 1024 * 1024, 353L * 1024 * 1024);
+ listener.complete();
+
+ String output = capture.text();
+ assertTrue(
+ output.startsWith("Downloading: Data Vault (353MB)"),
+ "size should be announced up front: " + output);
+ assertTrue(
+ output.contains("[##----------------------] 11% 1.8MB/s - 42.0MB of 353MB, 3 mins left"),
+ "unexpected bar line: " + output);
+ }
+
+ @Test
+ void barFillsProportionallyAndCompletely() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+ listener.started("plugin", 100L);
+ clock.now += 1000;
+
+ assertTrue(listener.progressLine(50L, 100L).contains("[############------------] 50%"));
+ assertTrue(
+ listener.progressLine(100L, 100L).contains("[########################] 100%"),
+ "a finished transfer should show a full bar");
+ assertTrue(
+ listener.progressLine(0L, 100L).contains("[------------------------] 0%"),
+ "an empty bar at the start");
+ }
+
+ @Test
+ void redrawErasesTheTailOfALongerPreviousLine() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+
+ listener.started("plugin", 1000L);
+ clock.now += 1000;
+ listener.transferred(500L, 1000L);
+ int firstLineLength = lastCarriageReturnSegment(capture.text()).length();
+ // A shorter follow-up line (no ETA once the transfer is complete) must not leave debris behind.
+ clock.now += 1000;
+ listener.transferred(1000L, 1000L);
+
+ String tail = lastCarriageReturnSegment(capture.text());
+ assertTrue(
+ tail.length() >= firstLineLength,
+ "the shorter line must be padded out to erase the longer one it replaced");
+ assertTrue(tail.endsWith(" ") || tail.contains("100%"), "unexpected redraw: [" + tail + "]");
+ }
+
+ @Test
+ void throttlesRedrawsRegardlessOfChunkCount() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+ listener.started("plugin", 10_000_000L);
+
+ for (int i = 1; i <= 2000; i++) {
+ clock.now += 1; // 1ms apart, i.e. 2s of transfer
+ listener.transferred(i * 5000L, 10_000_000L);
+ }
+
+ long frames = capture.text().chars().filter(c -> c == '\r').count();
+ // 2000ms at a 200ms floor allows ~10 frames. Near 2000 would mean the throttle is broken.
+ assertTrue(frames <= 15, "expected the throttle to collapse 2000 chunks, got " + frames);
+ assertTrue(frames >= 3, "throttle should still produce visible movement, got " + frames);
+ }
+
+ @Test
+ void theFinalChunkIsNeverThrottledAway() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+ listener.started("plugin", 10_000L);
+
+ clock.now += 1000;
+ listener.transferred(9_700L, 10_000L);
+ // Arrives inside the redraw floor, but a bar left at 97% would be the last thing the user sees.
+ clock.now += 1;
+ listener.transferred(10_000L, 10_000L);
+
+ String output = capture.text();
+ assertTrue(
+ output.contains("[########################] 100%"), "bar should finish full: " + output);
+ assertTrue(
+ output.endsWith(System.lineSeparator()), "the bar line should be closed: [" + output + "]");
+ }
+
+ @Test
+ void aRepeatedFinalChunkIsNotDrawnTwice() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+ listener.started("plugin", 10_000L);
+
+ clock.now += 1000;
+ // The downloader reports the last chunk, then repeats the exact byte count once the stream
+ // closes. Only one full-bar line may reach the terminal.
+ listener.transferred(10_000L, 10_000L);
+ clock.now += 1000;
+ listener.transferred(10_000L, 10_000L);
+ listener.complete();
+
+ long frames = capture.text().chars().filter(c -> c == '\r').count();
+ assertEquals(1, frames, "the closing frame should be drawn once: " + capture.text());
+ }
+
+ @Test
+ void redirectedOutputGetsMilestonesButNoBarFrames() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ // interactive=false: stdout is a file or a pipe.
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, false, ASCII, clock::millis);
+
+ listener.started("Data Vault", 1000L);
+ for (int i = 1; i <= 100; i++) {
+ clock.now += 500;
+ listener.transferred(i * 10L, 1000L);
+ }
+ listener.phase(IInstallListener.Phase.UNZIP, "hop-datavault");
+ listener.complete();
+
+ String output = capture.text();
+ assertFalse(output.contains("\r"), "a redirected log must not be filled with bar frames");
+ assertTrue(output.contains("Downloading: Data Vault (1000B)"), output);
+ assertTrue(output.contains("Unpacking..."), output);
+ }
+
+ @Test
+ void batchItemsArePrefixedWithTheirPosition() {
+ Capture capture = new Capture();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, new FakeClock()::millis);
+
+ listener.item("hop-tech-parquet", 2, 12);
+ assertTrue(capture.text().contains("[3/12] hop-tech-parquet"), capture.text());
+ }
+
+ @Test
+ void singleInstallHasNoBatchPrefix() {
+ Capture capture = new Capture();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, new FakeClock()::millis);
+
+ listener.item("hop-datavault", 0, 1);
+ assertEquals("", capture.text(), "a lone artifact should not be numbered '1/1'");
+ }
+
+ @Test
+ void unknownSizeDropsTheBarAndTheEta() {
+ Capture capture = new Capture();
+ FakeClock clock = new FakeClock();
+ ConsoleInstallListener listener =
+ new ConsoleInstallListener(capture.stream, true, ASCII, clock::millis);
+
+ listener.started("plugin", -1L);
+ clock.now += 1000;
+ listener.transferred(500_000L, -1L);
+
+ String line = lastCarriageReturnSegment(capture.text());
+ assertFalse(line.contains("["), "no total means no meaningful bar: " + line);
+ assertFalse(line.contains("left"), "no total means no honest ETA: " + line);
+ assertTrue(line.contains("488KB downloaded"), line);
+ }
+
+ @Test
+ void asciiGlyphsAreUsedWhenTheEncodingCannotCarryBlocks() {
+ PrintStream ascii =
+ new PrintStream(new ByteArrayOutputStream(), true, StandardCharsets.US_ASCII);
+ PrintStream utf8 = new PrintStream(new ByteArrayOutputStream(), true, StandardCharsets.UTF_8);
+
+ assertEquals('#', ConsoleInstallListener.glyphsFor(ascii)[0], "US-ASCII cannot encode a block");
+ assertEquals('█', ConsoleInstallListener.glyphsFor(utf8)[0], "UTF-8 can");
+ }
+
+ /** Everything written after the final carriage return, i.e. the currently visible line. */
+ private static String lastCarriageReturnSegment(String output) {
+ int index = output.lastIndexOf('\r');
+ return index < 0 ? output : output.substring(index + 1);
+ }
+}
diff --git a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListenerTest.java b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListenerTest.java
new file mode 100644
index 0000000000..d572039741
--- /dev/null
+++ b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/gui/ProgressMonitorInstallListenerTest.java
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.gui;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.IProgressMonitor;
+import org.apache.hop.marketplace.install.IInstallListener;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The progress adapter is pure arithmetic over callbacks, so it is unit-testable without SWT. What
+ * matters: the bar never goes backwards, never exceeds 100, reaches 100 on completion, and UI
+ * updates stay throttled no matter how many chunks arrive.
+ */
+class ProgressMonitorInstallListenerTest {
+
+ /** Captures what the monitor is told, and lets the test drive a fake clock. */
+ private static class FakeMonitor implements IProgressMonitor {
+ private final List<String> subTasks = new ArrayList<>();
+ private int totalWorked;
+ private int beginWork;
+ private String taskName;
+ private boolean cancelled;
+
+ @Override
+ public void beginTask(String message, int nrWorks) {
+ taskName = message;
+ beginWork = nrWorks;
+ }
+
+ @Override
+ public void subTask(String message) {
+ subTasks.add(message);
+ }
+
+ @Override
+ public boolean isCanceled() {
+ return cancelled;
+ }
+
+ @Override
+ public void worked(int nrWorks) {
+ totalWorked += nrWorks;
+ }
+
+ @Override
+ public void done() {
+ // nothing to assert on
+ }
+
+ @Override
+ public void setTaskName(String taskName) {
+ this.taskName = taskName;
+ }
+ }
+
+ /** Manually advanced clock so throttling is deterministic. */
+ private static class FakeClock implements ProgressMonitorInstallListener.IClock {
+ private long now;
+
+ @Override
+ public long millis() {
+ return now;
+ }
+ }
+
+ @Test
+ void singleDownloadDrivesBarToDownloadPhaseCeiling() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ assertEquals(ProgressMonitorInstallListener.TOTAL_WORK, monitor.beginWork);
+
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("plugin", 1000L);
+ // Halfway through the bytes, with enough clock movement to defeat the throttle.
+ clock.now += 1000;
+ listener.transferred(500L, 1000L);
+
+ // Download owns 5..85% of a single-item install, so half the bytes is roughly 45%.
+ assertTrue(
+ monitor.totalWorked >= 40 && monitor.totalWorked <= 50,
+ "expected ~45% at half the bytes, got " + monitor.totalWorked);
+
+ clock.now += 1000;
+ listener.transferred(1000L, 1000L);
+ assertEquals(85, monitor.totalWorked, "all bytes transferred is the end of the download phase");
+ }
+
+ @Test
+ void completeRunsBarOutToFull() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.complete();
+
+ assertEquals(ProgressMonitorInstallListener.TOTAL_WORK, monitor.totalWorked);
+ }
+
+ @Test
+ void barNeverExceedsTotalOrMovesBackwards() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.ACTIVATE, "x");
+ int afterActivate = monitor.totalWorked;
+ // A late, out-of-order transfer callback must not drag the bar back.
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ clock.now += 1000;
+ listener.transferred(1L, 1000L);
+
+ assertEquals(afterActivate, monitor.totalWorked, "bar moved backwards");
+
+ listener.complete();
+ listener.complete();
+ assertEquals(
+ ProgressMonitorInstallListener.TOTAL_WORK,
+ monitor.totalWorked,
+ "repeated completion must not push past 100");
+ }
+
+ @Test
+ void batchScalesEachItemIntoItsOwnSlice() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Applying environment", null);
+ // Fourth of four items: the bar should already be around 75% before this item does any work.
+ listener.item("plugin-d", 3, 4);
+ assertTrue(
+ monitor.totalWorked >= 75 && monitor.totalWorked <= 76,
+ "expected ~75% entering item 4 of 4, got " + monitor.totalWorked);
+ assertTrue(monitor.taskName.contains("plugin-d"), "task name should name the current item");
+ assertTrue(monitor.taskName.contains("4"), "task name should show the batch position");
+
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ clock.now += 1000;
+ listener.transferred(1000L, 1000L);
+ // 75% + 85% of the final quarter ≈ 96%.
+ assertTrue(
+ monitor.totalWorked >= 95 && monitor.totalWorked <= 97,
+ "expected ~96% after the last item downloaded, got " + monitor.totalWorked);
+ }
+
+ @Test
+ void updatesAreThrottledRegardlessOfChunkCount() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("plugin", 10_000_000L);
+ int subTasksAfterStart = monitor.subTasks.size();
+
+ // 2000 chunks arriving over 1 second of wall clock — what a fast link actually looks like.
+ for (int i = 1; i <= 2000; i++) {
+ clock.now += 1; // 1ms apart
+ listener.transferred(i * 5000L, 10_000_000L);
+ }
+
+ int updates = monitor.subTasks.size() - subTasksAfterStart;
+ // 2000ms of transfer at a 150ms floor allows ~13 updates. Anything near 2000 means the
+ // throttle is broken and we would be flooding the display with asyncExec runnables.
+ assertTrue(updates <= 20, "expected the throttle to collapse 2000 chunks, got " + updates);
+ assertTrue(updates >= 5, "throttle should still produce visible movement, got " + updates);
+ }
+
+ @Test
+ void unknownSizeReportsVolumeWithoutPercentOrEta() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("plugin", -1L);
+ int barAtStart = monitor.totalWorked;
+ clock.now += 1000;
+ listener.transferred(500_000L, -1L);
+
+ assertEquals(barAtStart, monitor.totalWorked, "no size means no percentage to report");
+ String last = monitor.subTasks.get(monitor.subTasks.size() - 1);
+ assertFalse(last.contains("left"), "an ETA without a total would be invented: " + last);
+ assertTrue(last.contains("488KB"), "should show volume downloaded: " + last);
+ }
+
+ @Test
+ void downloadLineShowsTheDisplayNameNotTheMavenCoordinate() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+
+ listener.begin("Installing Data Vault", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ // The download layer only knows the coordinate; it must not leak into the dialog.
+ listener.started("org.apache.hop:hop-datavault:2.19.0", 1024L);
+
+ String line = monitor.subTasks.get(monitor.subTasks.size() - 1);
+ assertEquals("Downloading: Data Vault", line);
+ }
+
+ @Test
+ void downloadLineFallsBackToTheTransferLabelWhenNoNameIsKnown() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+
+ listener.begin("Applying environment", null);
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("org.postgresql:postgresql:42.7.3", 1024L);
+
+ assertEquals(
+ "Downloading: org.postgresql:postgresql:42.7.3",
+ monitor.subTasks.get(monitor.subTasks.size() - 1),
+ "better to show the coordinate than nothing at all");
+ }
+
+ @Test
+ void batchItemNameReachesTheDownloadLine() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+
+ listener.begin("Applying environment", null);
+ listener.item("hop-tech-parquet", 0, 3);
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("org.apache.hop:hop-tech-parquet:2.19.0", 1024L);
+
+ assertEquals(
+ "Downloading: hop-tech-parquet", monitor.subTasks.get(monitor.subTasks.size() - 1));
+ }
+
+ @Test
+ void transferTextReadsLikeABrowserDownload() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("plugin", 353L * 1024 * 1024);
+ // 42MB in 24s ≈ 1.75MB/s, leaving 311MB ≈ 178s ≈ 3 mins.
+ clock.now += 24_000;
+ listener.transferred(42L * 1024 * 1024, 353L * 1024 * 1024);
+
+ String text = monitor.subTasks.get(monitor.subTasks.size() - 1);
+ assertEquals("1.8MB/s - 42.0MB of 353MB, 3 mins left", text);
+ assertFalse(text.contains("--"), "the placeholder dashes should be gone");
+ }
+
+ @Test
+ void etaIsWordedAndSingularisedByUnit() {
+ FakeMonitor monitor = new FakeMonitor();
+ FakeClock clock = new FakeClock();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, clock::millis);
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+
+ // Drive a known speed of exactly 1000 bytes/s, then vary how much is left.
+ listener.started("plugin", 100_000L);
+ clock.now += 1000;
+ listener.transferred(1000L, 100_000L);
+
+ assertTrue(listener.transferText(99_999L, 100_000L).endsWith("1 sec left"));
+ assertTrue(listener.transferText(70_000L, 100_000L).endsWith("30 secs left"));
+ assertTrue(listener.transferText(40_000L, 100_000L).endsWith("1 min left"));
+ assertTrue(listener.transferText(0L, 3_700_000L).endsWith("1 hr left"));
+ assertTrue(listener.transferText(0L, 7_200_000L).endsWith("2 hrs left"));
+ // 2.5 hours rounds up rather than truncating to "2 hrs".
+ assertTrue(listener.transferText(0L, 9_000_000L).endsWith("3 hrs left"));
+ }
+
+ @Test
+ void startingIsShownUntilASpeedIsKnown() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+ listener.begin("Installing plugin", "Data Vault");
+ listener.phase(IInstallListener.Phase.DOWNLOAD, "nexus");
+ listener.started("plugin", -1L);
+
+ // No elapsed time yet, so no speed can be measured and no size was advertised.
+ assertEquals("Starting download...", listener.transferText(0L, -1L));
+ // Size known but still no speed: show the counts rather than a fake rate.
+ assertEquals("0B of 1.0KB", listener.transferText(0L, 1024L));
+ }
+
+ @Test
+ void cancellationIsDelegatedToTheMonitor() {
+ FakeMonitor monitor = new FakeMonitor();
+ ProgressMonitorInstallListener listener =
+ new ProgressMonitorInstallListener(monitor, new FakeClock()::millis);
+
+ assertFalse(listener.isCancelled());
+ monitor.cancelled = true;
+ assertTrue(listener.isCancelled(), "Cancel button must reach the download loop");
+ }
+}
diff --git a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/PluginInstallerTest.java b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/PluginInstallerTest.java
index 0e85de4156..9a67d9f43d 100644
--- a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/PluginInstallerTest.java
+++ b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/PluginInstallerTest.java
@@ -19,6 +19,7 @@ package org.apache.hop.marketplace.install;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import com.sun.net.httpserver.HttpServer;
@@ -28,8 +29,11 @@ import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
+import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.HopLogStore;
import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.marketplace.config.MarketplaceConfig;
@@ -204,6 +208,144 @@ class PluginInstallerTest {
}
}
+ @Test
+ void installReportsPhasesAndDownloadBytes() throws Exception {
+ byte[] zipBytes = buildPluginZip();
+ HttpServer server = zipServer(zipBytes);
+ server.start();
+ try {
+ Path hopHome = tempDir.resolve("hop-progress");
+ Files.createDirectories(hopHome.resolve("plugins"));
+ MarketplaceConfig config = localRepoConfig(server.getAddress().getPort());
+ RecordingInstallListener listener = new RecordingInstallListener();
+
+ new PluginInstaller(new LogChannel("test"), hopHome, config)
+ .install(
+ new MavenCoordinates("org.apache.hop", "hop-test-plugin", "1.0.0"),
+ true,
+ null,
+ null,
+ listener);
+
+ assertEquals(
+ List.of(
+ IInstallListener.Phase.RESOLVE,
+ IInstallListener.Phase.DOWNLOAD,
+ IInstallListener.Phase.UNZIP,
+ IInstallListener.Phase.ACTIVATE),
+ listener.phases,
+ "the progress bar depends on phases arriving in install order");
+ assertEquals(zipBytes.length, listener.startedTotal, "size must reach the listener");
+ assertEquals(
+ zipBytes.length,
+ listener.lastBytes,
+ "the final byte callback must equal the file size so the bar completes");
+ assertTrue(Files.isRegularFile(hopHome.resolve("plugins/tech/test/plugin.jar")));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void cancelDuringDownloadInstallsNothingAndDoesNotTryOtherRepositories() throws Exception {
+ byte[] zipBytes = buildPluginZip();
+ HttpServer server = zipServer(zipBytes);
+ server.start();
+ try {
+ Path hopHome = tempDir.resolve("hop-cancel");
+ Files.createDirectories(hopHome.resolve("plugins"));
+ int port = server.getAddress().getPort();
+
+ // Two working repositories: a cancel must not silently restart against the second one.
+ MarketplaceConfig config = new MarketplaceConfig();
+ config.getRepositories().clear();
+ config.getRepositories().add(new MarketplaceRepository("one", localUrl(port), true));
+ config.getRepositories().add(new MarketplaceRepository("two", localUrl(port), false));
+
+ RecordingInstallListener listener = new RecordingInstallListener();
+ listener.cancelOnceDownloading = true;
+
+ assertThrows(
+ HopException.class,
+ () ->
+ new PluginInstaller(new LogChannel("test"), hopHome, config)
+ .install(
+ new MavenCoordinates("org.apache.hop", "hop-test-plugin", "1.0.0"),
+ true,
+ null,
+ null,
+ listener));
+
+ assertFalse(
+ Files.exists(hopHome.resolve("plugins/tech/test/plugin.jar")),
+ "a cancelled install must not leave plugin files behind");
+ assertFalse(
+ Files.exists(
+ hopHome.resolve(PluginInstaller.RECEIPTS_DIR).resolve("hop-test-plugin.json")),
+ "a cancelled install must not write a receipt");
+ assertEquals(
+ 1,
+ listener.phases.stream().filter(p -> p == IInstallListener.Phase.DOWNLOAD).count(),
+ "cancel must not fall through to the second repository");
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ /** Captures the phase sequence and byte counts an install reports. */
+ private static class RecordingInstallListener implements IInstallListener {
+ private final List<Phase> phases = new ArrayList<>();
+ private long startedTotal = Long.MIN_VALUE;
+ private long lastBytes = -1;
+ private boolean cancelOnceDownloading;
+
+ @Override
+ public void phase(Phase phase, String detail) {
+ phases.add(phase);
+ }
+
+ @Override
+ public void started(String label, long totalBytes) {
+ startedTotal = totalBytes;
+ }
+
+ @Override
+ public void transferred(long bytesSoFar, long totalBytes) {
+ lastBytes = bytesSoFar;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ // Only once the download has actually started, so this exercises cancel mid-transfer rather
+ // than the cheaper pre-flight check.
+ return cancelOnceDownloading && phases.contains(Phase.DOWNLOAD);
+ }
+ }
+
+ private static String localUrl(int port) {
+ return "http://127.0.0.1:" + port + "/";
+ }
+
+ private static MarketplaceConfig localRepoConfig(int port) {
+ MarketplaceConfig config = new MarketplaceConfig();
+ config.getRepositories().clear();
+ config.getRepositories().add(new MarketplaceRepository("local", localUrl(port), true));
+ return config;
+ }
+
+ private static HttpServer zipServer(byte[] zipBytes) throws IOException {
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/org/apache/hop/hop-test-plugin/1.0.0/hop-test-plugin-1.0.0.zip",
+ exchange -> {
+ exchange.getResponseHeaders().add("Content-Type", "application/zip");
+ exchange.sendResponseHeaders(200, zipBytes.length);
+ exchange.getResponseBody().write(zipBytes);
+ exchange.close();
+ });
+ return server;
+ }
+
private static byte[] buildPluginZip() throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try (ZipOutputStream zos = new ZipOutputStream(bos)) {
diff --git a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/TransferFormatTest.java b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/TransferFormatTest.java
new file mode 100644
index 0000000000..e884c250a8
--- /dev/null
+++ b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/install/TransferFormatTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.install;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Locale;
+import org.apache.hop.marketplace.install.TransferFormat.EtaUnit;
+import org.junit.jupiter.api.Test;
+
+/** Formatting shared by the GUI progress dialog and the CLI progress bar. */
+class TransferFormatTest {
+
+ @Test
+ void bytesUseOneDecimalBelowOneHundred() {
+ // Same rule a browser uses: keeps the width stable as the number climbs.
+ assertEquals("512B", TransferFormat.bytes(512));
+ assertEquals("1.0KB", TransferFormat.bytes(1024));
+ assertEquals("488KB", TransferFormat.bytes(500_000));
+ assertEquals("42.0MB", TransferFormat.bytes(42L * 1024 * 1024));
+ assertEquals("353MB", TransferFormat.bytes(353L * 1024 * 1024));
+ assertEquals("1.5GB", TransferFormat.bytes(1536L * 1024 * 1024));
+ }
+
+ @Test
+ void bytesNeverPromotePastTheLargestUnit() {
+ // 4096TB rather than a unit label we do not have.
+ assertTrue(TransferFormat.bytes(4096L * 1024 * 1024 * 1024 * 1024).endsWith("TB"));
+ }
+
+ @Test
+ void speedIsAByteCountPerSecond() {
+ assertEquals("1.8MB/s", TransferFormat.speed(1_835_008L));
+ assertEquals("0B/s", TransferFormat.speed(0L));
+ }
+
+ @Test
+ void etaRoundsToTheCoarsestUsefulUnit() {
+ assertEquals(new TransferFormat.Eta(30, EtaUnit.SECONDS), TransferFormat.eta(30));
+ assertEquals(new TransferFormat.Eta(1, EtaUnit.MINUTES), TransferFormat.eta(60));
+ assertEquals(new TransferFormat.Eta(3, EtaUnit.MINUTES), TransferFormat.eta(178));
+ assertEquals(new TransferFormat.Eta(1, EtaUnit.HOURS), TransferFormat.eta(3700));
+ assertEquals(new TransferFormat.Eta(2, EtaUnit.HOURS), TransferFormat.eta(7200));
+ // 2.5 hours rounds up rather than truncating to "2 hrs".
+ assertEquals(new TransferFormat.Eta(3, EtaUnit.HOURS), TransferFormat.eta(9000));
+ }
+
+ @Test
+ void etaNeverReportsZero() {
+ // "0 secs left" reads as finished when it is not; the last tick should say "1 sec left".
+ assertEquals(new TransferFormat.Eta(1, EtaUnit.SECONDS), TransferFormat.eta(0));
+ }
+
+ @Test
+ void byteFormattingIgnoresTheDefaultLocale() {
+ // A comma-decimal locale must not turn "17.6MB" into "17,6MB", nor group thousands with a dot
+ // where "1.750MB" would be ambiguous. Guards against the suite passing only on dot-decimal
+ // machines while the shipped binary prints something else.
+ Locale original = Locale.getDefault();
+ try {
+ Locale.setDefault(Locale.GERMANY);
+ assertEquals("17.6MB", TransferFormat.bytes(18_454_938L));
+ assertEquals("353MB", TransferFormat.bytes(353L * 1024 * 1024));
+ assertEquals("1.8MB/s", TransferFormat.speed(1_835_008L));
+ } finally {
+ Locale.setDefault(original);
+ }
+ }
+
+ @Test
+ void etaFlagsSingularAmounts() {
+ assertTrue(TransferFormat.eta(60).isSingular());
+ assertFalse(TransferFormat.eta(120).isSingular());
+ }
+}
diff --git a/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/resolve/MavenRepositoryClientDownloadTest.java b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/resolve/MavenRepositoryClientDownloadTest.java
new file mode 100644
index 0000000000..16feac9ea4
--- /dev/null
+++ b/plugins/misc/marketplace/src/test/java/org/apache/hop/marketplace/resolve/MavenRepositoryClientDownloadTest.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.marketplace.resolve;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpClient;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.marketplace.config.MarketplaceRepository;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Covers the streaming download loop: size reporting, progress callbacks, cancellation and partial
+ * file cleanup. These are the guarantees the marketplace progress dialog is built on.
+ */
+class MavenRepositoryClientDownloadTest {
+
+ private static final MavenCoordinates COORDS =
+ new MavenCoordinates("org.apache.hop", "hop-datavault", "1.0.0");
+
+ @TempDir private Path tempDir;
+
+ /** Records everything a listener is told, so assertions can inspect the whole sequence. */
+ private static class RecordingListener implements ITransferListener {
+ private final List<Long> progress = new ArrayList<>();
+ private String label;
+ private long startedTotal = Long.MIN_VALUE;
+ private long cancelAfterBytes = -1;
+
+ @Override
+ public void started(String label, long totalBytes) {
+ this.label = label;
+ this.startedTotal = totalBytes;
+ }
+
+ @Override
+ public void transferred(long bytesSoFar, long totalBytes) {
+ progress.add(bytesSoFar);
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return cancelAfterBytes >= 0
+ && !progress.isEmpty()
+ && progress.get(progress.size() - 1) >= cancelAfterBytes;
+ }
+ }
+
+ @Test
+ void reportsTotalSizeAndProgressFromContentLength() throws Exception {
+ byte[] payload = payload(200_000);
+ MavenRepositoryClient client = clientReturning(payload, Map.of("content-length", "200000"));
+ RecordingListener listener = new RecordingListener();
+ Path target = tempDir.resolve("plugin.zip");
+
+ Path result = client.downloadZip(repo(), COORDS, target, listener);
+
+ assertEquals(target, result);
+ assertEquals(200_000L, Files.size(target), "whole payload must reach disk");
+ assertEquals(200_000L, listener.startedTotal, "size comes from Content-Length");
+ assertEquals("org.apache.hop:hop-datavault:1.0.0", listener.label);
+ assertFalse(listener.progress.isEmpty(), "progress must be reported per chunk");
+ assertEquals(
+ 200_000L,
+ listener.progress.get(listener.progress.size() - 1),
+ "final callback must land exactly on the total, so the bar reaches 100%");
+ assertTrue(
+ listener.progress.size() > 2,
+ "a 200KB payload spans several 64KB chunks; got " + listener.progress.size());
+ }
+
+ @Test
+ void unknownSizeWhenContentLengthMissing() throws Exception {
+ byte[] payload = payload(1000);
+ MavenRepositoryClient client = clientReturning(payload, Map.of());
+ RecordingListener listener = new RecordingListener();
+ Path target = tempDir.resolve("plugin.zip");
+
+ client.downloadZip(repo(), COORDS, target, listener);
+
+ assertEquals(-1L, listener.startedTotal, "no Content-Length must report -1, not 0");
+ assertEquals(1000L, Files.size(target));
+ }
+
+ @Test
+ void unknownSizeWhenResponseIsEncoded() throws Exception {
+ // Content-Length would describe the gzipped length, not what we write to disk.
+ byte[] payload = payload(1000);
+ MavenRepositoryClient client =
+ clientReturning(payload, Map.of("content-length", "400", "content-encoding", "gzip"));
+ RecordingListener listener = new RecordingListener();
+
+ client.downloadZip(repo(), COORDS, tempDir.resolve("plugin.zip"), listener);
+
+ assertEquals(-1L, listener.startedTotal, "encoded responses must not trust Content-Length");
+ }
+
+ @Test
+ void cancelStopsTransferAndRemovesPartialFile() throws Exception {
+ byte[] payload = payload(500_000);
+ MavenRepositoryClient client = clientReturning(payload, Map.of("content-length", "500000"));
+ RecordingListener listener = new RecordingListener();
+ listener.cancelAfterBytes = 64 * 1024;
+ Path target = tempDir.resolve("plugin.zip");
+
+ HopException e =
+ assertThrows(
+ HopException.class, () -> client.downloadZip(repo(), COORDS, target, listener));
+
+ assertTrue(e.getMessage().contains("cancelled"), "message was: " + e.getMessage());
+ assertFalse(
+ Files.exists(target),
+ "a truncated zip must never survive — the unzip step would fail confusingly");
+ }
+
+ @Test
+ void failedDownloadRemovesPartialFile() throws Exception {
+ MavenRepositoryClient client = clientFailingMidStream(payload(300_000));
+ Path target = tempDir.resolve("plugin.zip");
+
+ assertThrows(
+ HopException.class,
+ () -> client.downloadZip(repo(), COORDS, target, ITransferListener.NONE));
+
+ assertFalse(Files.exists(target), "partial download left behind after an I/O failure");
+ }
+
+ @Test
+ void downloadWorksWithoutAListener() throws Exception {
+ // The legacy signature must keep working for the CLI and for existing callers.
+ MavenRepositoryClient client = clientReturning(payload(5000), Map.of("content-length", "5000"));
+ Path target = tempDir.resolve("plugin.zip");
+
+ Path result = client.downloadZip(repo(), COORDS, target);
+
+ assertNotNull(result);
+ assertEquals(5000L, Files.size(target));
+ }
+
+ private static byte[] payload(int size) {
+ byte[] data = new byte[size];
+ for (int i = 0; i < size; i++) {
+ data[i] = (byte) (i % 251);
+ }
+ return data;
+ }
+
+ private static MarketplaceRepository repo() {
+ return new MarketplaceRepository("test", "https://example.com/repository/hop/");
+ }
+
+ private static MavenRepositoryClient clientReturning(byte[] payload, Map<String, String> headers)
+ throws Exception {
+ return new MavenRepositoryClient(
+ log(), stubClient(new ByteArrayInputStream(payload), headers, 200));
+ }
+
+ /** A stream that dies part way through, simulating a dropped connection. */
+ private static MavenRepositoryClient clientFailingMidStream(byte[] payload) throws Exception {
+ InputStream failing =
+ new InputStream() {
+ private int served;
+
+ @Override
+ public int read() throws IOException {
+ return read(new byte[1], 0, 1);
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (served >= 128 * 1024) {
+ throw new IOException("connection reset");
+ }
+ int n = Math.min(len, payload.length - served);
+ System.arraycopy(payload, served, b, off, n);
+ served += n;
+ return n;
+ }
+ };
+ return new MavenRepositoryClient(
+ log(), stubClient(failing, Map.of("content-length", String.valueOf(payload.length)), 200));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static HttpClient stubClient(
+ InputStream body, Map<String, String> headers, int statusCode) throws Exception {
+ HttpResponse<InputStream> response = mock(HttpResponse.class);
+ when(response.statusCode()).thenReturn(statusCode);
+ when(response.body()).thenReturn(body);
+ Map<String, List<String>> headerMap = new java.util.LinkedHashMap<>();
+ headers.forEach((k, v) -> headerMap.put(k, List.of(v)));
+ when(response.headers()).thenReturn(HttpHeaders.of(headerMap, (a, b) -> true));
+
+ HttpClient client = mock(HttpClient.class);
+ when(client.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)))
+ .thenReturn((HttpResponse<Object>) (HttpResponse<?>) response);
+ return client;
+ }
+
+ /**
+ * Mocked rather than a real {@link org.apache.hop.core.logging.LogChannel}, which would need the
+ * central log store initialised for these plain unit tests.
+ */
+ private static ILogChannel log() {
+ return mock(ILogChannel.class);
+ }
+}