You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by re...@apache.org on 2024/09/23 01:30:56 UTC
(cxf) branch 3.6.x-fixes updated (9835dfc01d -> b544d0eb69)
This is an automated email from the ASF dual-hosted git repository.
reta pushed a change to branch 3.6.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
from 9835dfc01d Recording .gitmergeinfo Changes
new 71075539b1 Revert temporary excludion of some build agents from the label (due to persistent issues)
new 43b4eee96c CXF-7396: CachedOutputStream doesn't delete temp files (#2048)
new b544d0eb69 Recording .gitmergeinfo Changes
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitmergeinfo | 1 +
Jenkinsfile | 6 +-
core/pom.xml | 6 +-
.../java/org/apache/cxf/io/CachedConstants.java | 8 +
.../java/org/apache/cxf/io/CachedOutputStream.java | 22 +-
.../apache/cxf/io/CachedOutputStreamCleaner.java | 25 ++-
.../cxf/io/DelayedCachedOutputStreamCleaner.java | 239 +++++++++++++++++++++
.../main/resources/META-INF/cxf/bus-extensions.txt | 2 +-
.../io/DelayedCachedOutputStreamCleanerTest.java | 201 +++++++++++++++++
parent/pom.xml | 1 -
.../apache/cxf/systest/jaxws/ClientServerTest.java | 44 +++-
11 files changed, 537 insertions(+), 18 deletions(-)
copy services/xkms/xkms-client/src/main/java/org/apache/cxf/xkms/cache/XKMSClientCache.java => core/src/main/java/org/apache/cxf/io/CachedOutputStreamCleaner.java (61%)
create mode 100644 core/src/main/java/org/apache/cxf/io/DelayedCachedOutputStreamCleaner.java
create mode 100644 core/src/test/java/org/apache/cxf/io/DelayedCachedOutputStreamCleanerTest.java
(cxf) 02/03: CXF-7396: CachedOutputStream doesn't delete temp files (#2048)
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.6.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 43b4eee96cbffdf17218c235aa7f5956a1df053f
Author: Andriy Redko <dr...@gmail.com>
AuthorDate: Sun Sep 22 10:13:32 2024 -0400
CXF-7396: CachedOutputStream doesn't delete temp files (#2048)
* CXF-7396: CachedOutputStream doesn't delete temp files
* Refactor the cleaner implementation and add guardrails for cleaner delay
(cherry picked from commit 03a85a52f78e6eb5826e4b8bfc4992e0f4e0414b)
(cherry picked from commit 0a97dded6720e83507811900df8f9c4bf2ff6471)
# Conflicts:
# core/pom.xml
---
core/pom.xml | 6 +-
.../java/org/apache/cxf/io/CachedConstants.java | 8 +
.../java/org/apache/cxf/io/CachedOutputStream.java | 22 +-
.../apache/cxf/io/CachedOutputStreamCleaner.java | 43 ++++
.../cxf/io/DelayedCachedOutputStreamCleaner.java | 239 +++++++++++++++++++++
.../main/resources/META-INF/cxf/bus-extensions.txt | 2 +-
.../io/DelayedCachedOutputStreamCleanerTest.java | 201 +++++++++++++++++
parent/pom.xml | 1 -
.../apache/cxf/systest/jaxws/ClientServerTest.java | 44 +++-
9 files changed, 561 insertions(+), 5 deletions(-)
diff --git a/core/pom.xml b/core/pom.xml
index da3ba85f23..617e170c38 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -230,7 +230,11 @@
<artifactId>saaj-impl</artifactId>
<scope>test</scope>
</dependency>
-
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/core/src/main/java/org/apache/cxf/io/CachedConstants.java b/core/src/main/java/org/apache/cxf/io/CachedConstants.java
index 24ba8d8347..1b23f66e99 100644
--- a/core/src/main/java/org/apache/cxf/io/CachedConstants.java
+++ b/core/src/main/java/org/apache/cxf/io/CachedConstants.java
@@ -71,6 +71,14 @@ public final class CachedConstants {
public static final String CIPHER_TRANSFORMATION_BUS_PROP =
"bus.io.CachedOutputStream.CipherTransformation";
+ /**
+ * The delay (in ms) for cleaning up unclosed {@code CachedOutputStream} instances. 30 minutes
+ * is specified by default, the minimum value is 2 seconds. If the value of the delay is set to
+ * 0 (or is negative), the cleaner will be deactivated.
+ */
+ public static final String CLEANER_DELAY_BUS_PROP =
+ "bus.io.CachedOutputStreamCleaner.Delay";
+
private CachedConstants() {
// complete
}
diff --git a/core/src/main/java/org/apache/cxf/io/CachedOutputStream.java b/core/src/main/java/org/apache/cxf/io/CachedOutputStream.java
index ea8ce0d625..3ba937d04b 100644
--- a/core/src/main/java/org/apache/cxf/io/CachedOutputStream.java
+++ b/core/src/main/java/org/apache/cxf/io/CachedOutputStream.java
@@ -22,6 +22,7 @@ package org.apache.cxf.io;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
@@ -93,6 +94,7 @@ public class CachedOutputStream extends OutputStream {
private List<CachedOutputStreamCallback> callbacks;
private List<Object> streamList = new ArrayList<>();
+ private CachedOutputStreamCleaner cachedOutputStreamCleaner;
public CachedOutputStream() {
this(defaultThreshold);
@@ -127,6 +129,8 @@ public class CachedOutputStream extends OutputStream {
outputDir = f;
}
}
+
+ cachedOutputStreamCleaner = b.getExtension(CachedOutputStreamCleaner.class);
}
}
@@ -279,6 +283,9 @@ public class CachedOutputStream extends OutputStream {
}
} finally {
streamList.remove(currentStream);
+ if (cachedOutputStreamCleaner != null) {
+ cachedOutputStreamCleaner.unregister(currentStream);
+ }
deleteTempFile();
inmem = true;
}
@@ -481,6 +488,9 @@ public class CachedOutputStream extends OutputStream {
bout.writeTo(currentStream);
inmem = false;
streamList.add(currentStream);
+ if (cachedOutputStreamCleaner != null) {
+ cachedOutputStreamCleaner.register(this);
+ }
} catch (Exception ex) {
//Could be IOException or SecurityException or other issues.
//Don't care what, just keep it in memory.
@@ -512,6 +522,10 @@ public class CachedOutputStream extends OutputStream {
try {
InputStream fileInputStream = new TransferableFileInputStream(tempFile);
streamList.add(fileInputStream);
+ if (cachedOutputStreamCleaner != null) {
+ cachedOutputStreamCleaner.register(fileInputStream);
+ }
+
if (cipherTransformation != null) {
fileInputStream = new CipherInputStream(fileInputStream, ciphers.getDecryptor()) {
boolean closed;
@@ -537,7 +551,7 @@ public class CachedOutputStream extends OutputStream {
FileUtils.delete(file);
}
}
- private boolean maybeDeleteTempFile(Object stream) {
+ private boolean maybeDeleteTempFile(Closeable stream) {
boolean postClosedInvoked = false;
streamList.remove(stream);
if (!inmem && tempFile != null && streamList.isEmpty() && allowDeleteOfFile) {
@@ -549,6 +563,9 @@ public class CachedOutputStream extends OutputStream {
//ignore
}
postClosedInvoked = true;
+ if (cachedOutputStreamCleaner != null) {
+ cachedOutputStreamCleaner.unregister(this);
+ }
}
deleteTempFile();
currentStream = new LoadingByteArrayOutputStream(1024);
@@ -665,6 +682,9 @@ public class CachedOutputStream extends OutputStream {
if (!closed) {
super.close();
maybeDeleteTempFile(this);
+ if (cachedOutputStreamCleaner != null) {
+ cachedOutputStreamCleaner.unregister(this);
+ }
}
closed = true;
}
diff --git a/core/src/main/java/org/apache/cxf/io/CachedOutputStreamCleaner.java b/core/src/main/java/org/apache/cxf/io/CachedOutputStreamCleaner.java
new file mode 100644
index 0000000000..3d6361a4c9
--- /dev/null
+++ b/core/src/main/java/org/apache/cxf/io/CachedOutputStreamCleaner.java
@@ -0,0 +1,43 @@
+/**
+ * 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.cxf.io;
+
+import java.io.Closeable;
+
+/**
+ * The {@link Bus} extension to clean up unclosed {@link CachedOutputStream} instances (and alike) backed by
+ * temporary files (leading to disk fill, see https://issues.apache.org/jira/browse/CXF-7396.
+ */
+public interface CachedOutputStreamCleaner {
+ /**
+ * Run the clean up
+ */
+ void clean();
+
+ /**
+ * Register the stream instance for the clean up
+ */
+ void unregister(Closeable closeable);
+
+ /**
+ * Unregister the stream instance from the clean up (closed properly)
+ */
+ void register(Closeable closeable);
+}
diff --git a/core/src/main/java/org/apache/cxf/io/DelayedCachedOutputStreamCleaner.java b/core/src/main/java/org/apache/cxf/io/DelayedCachedOutputStreamCleaner.java
new file mode 100644
index 0000000000..50d94b6737
--- /dev/null
+++ b/core/src/main/java/org/apache/cxf/io/DelayedCachedOutputStreamCleaner.java
@@ -0,0 +1,239 @@
+/**
+ * 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.cxf.io;
+
+import javax.annotation.Resource;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.buslifecycle.BusLifeCycleListener;
+import org.apache.cxf.buslifecycle.BusLifeCycleManager;
+import org.apache.cxf.common.logging.LogUtils;
+
+public final class DelayedCachedOutputStreamCleaner implements CachedOutputStreamCleaner, BusLifeCycleListener {
+ private static final Logger LOG = LogUtils.getL7dLogger(DelayedCachedOutputStreamCleaner.class);
+ private static final long MIN_DELAY = 2000; /* 2 seconds */
+ private static final DelayedCleaner NOOP_CLEANER = new DelayedCleaner() {
+ // NOOP
+ };
+
+ private DelayedCleaner cleaner = NOOP_CLEANER;
+
+ private interface DelayedCleaner extends CachedOutputStreamCleaner, Closeable {
+ @Override
+ default void register(Closeable closeable) {
+ }
+
+ @Override
+ default void unregister(Closeable closeable) {
+ }
+
+ @Override
+ default void close() {
+ }
+
+ @Override
+ default void clean() {
+ }
+
+ default void forceClean() {
+ }
+ }
+
+ private static final class DelayedCleanerImpl implements DelayedCleaner {
+ private final long delay; /* default is 30 minutes, in milliseconds */
+ private final DelayQueue<DelayedCloseable> queue = new DelayQueue<>();
+ private final Timer timer;
+
+ DelayedCleanerImpl(final long delay) {
+ this.delay = delay;
+ this.timer = new Timer("DelayedCachedOutputStreamCleaner", true);
+ this.timer.scheduleAtFixedRate(new TimerTask() {
+ @Override
+ public void run() {
+ clean();
+ }
+ }, 0, Math.max(MIN_DELAY, delay >> 1));
+ }
+
+ @Override
+ public void register(Closeable closeable) {
+ queue.put(new DelayedCloseable(closeable, delay));
+ }
+
+ @Override
+ public void unregister(Closeable closeable) {
+ queue.remove(new DelayedCloseable(closeable, delay));
+ }
+
+ @Override
+ public void clean() {
+ final Collection<DelayedCloseable> closeables = new ArrayList<>();
+ queue.drainTo(closeables);
+ clean(closeables);
+ }
+
+ @Override
+ public void forceClean() {
+ clean(queue);
+ }
+
+ @Override
+ public void close() {
+ timer.cancel();
+ queue.clear();
+ }
+
+ private void clean(Collection<DelayedCloseable> closeables) {
+ final Iterator<DelayedCloseable> iterator = closeables.iterator();
+ while (iterator.hasNext()) {
+ final DelayedCloseable next = iterator.next();
+ try {
+ iterator.remove();
+ LOG.warning("Unclosed (leaked?) stream detected: " + next.closeable);
+ next.closeable.close();
+ } catch (final IOException | RuntimeException ex) {
+ LOG.warning("Unable to close (leaked?) stream: " + ex.getMessage());
+ }
+ }
+ }
+ }
+
+ private static final class DelayedCloseable implements Delayed {
+ private final Closeable closeable;
+ private final long expireAt;
+
+ DelayedCloseable(final Closeable closeable, final long delay) {
+ this.closeable = closeable;
+ this.expireAt = System.nanoTime() + delay;
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ return Long.compare(expireAt, ((DelayedCloseable) o).expireAt);
+ }
+
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return unit.convert(expireAt - System.nanoTime(), TimeUnit.NANOSECONDS);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(closeable);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+
+ if (obj == null) {
+ return false;
+ }
+
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+
+ final DelayedCloseable other = (DelayedCloseable) obj;
+ return Objects.equals(closeable, other.closeable);
+ }
+ }
+
+ @Resource
+ public void setBus(Bus bus) {
+ Number delayValue = null;
+ BusLifeCycleManager busLifeCycleManager = null;
+
+ if (bus != null) {
+ delayValue = (Number) bus.getProperty(CachedConstants.CLEANER_DELAY_BUS_PROP);
+ busLifeCycleManager = bus.getExtension(BusLifeCycleManager.class);
+ }
+
+ if (cleaner != null) {
+ cleaner.close();
+ }
+
+ if (delayValue == null) {
+ // Default delay is set to 30 mins
+ cleaner = new DelayedCleanerImpl(TimeUnit.MILLISECONDS.convert(30, TimeUnit.MINUTES));
+ } else {
+ final long value = delayValue.longValue();
+ if (value > 0 && value >= MIN_DELAY) {
+ cleaner = new DelayedCleanerImpl(value); /* already in milliseconds */
+ } else {
+ cleaner = NOOP_CLEANER;
+ if (value != 0) {
+ throw new IllegalArgumentException("The value of " + CachedConstants.CLEANER_DELAY_BUS_PROP
+ + " property is invalid: " + value + " (should be >= " + MIN_DELAY + ", 0 to deactivate)");
+ }
+ }
+ }
+
+ if (busLifeCycleManager != null) {
+ busLifeCycleManager.registerLifeCycleListener(this);
+ }
+ }
+
+ @Override
+ public void register(Closeable closeable) {
+ cleaner.register(closeable);
+ }
+
+ @Override
+ public void unregister(Closeable closeable) {
+ cleaner.unregister(closeable);
+ }
+
+ @Override
+ public void clean() {
+ cleaner.clean();
+ }
+
+ @Override
+ public void initComplete() {
+ }
+
+ @Override
+ public void postShutdown() {
+ }
+
+ @Override
+ public void preShutdown() {
+ cleaner.close();
+ }
+
+ public void forceClean() {
+ cleaner.forceClean();
+ }
+}
diff --git a/core/src/main/resources/META-INF/cxf/bus-extensions.txt b/core/src/main/resources/META-INF/cxf/bus-extensions.txt
index 3677c28b5a..93b8a6b3aa 100644
--- a/core/src/main/resources/META-INF/cxf/bus-extensions.txt
+++ b/core/src/main/resources/META-INF/cxf/bus-extensions.txt
@@ -11,4 +11,4 @@ org.apache.cxf.bus.resource.ResourceManagerImpl:org.apache.cxf.resource.Resource
org.apache.cxf.catalog.OASISCatalogManager:org.apache.cxf.catalog.OASISCatalogManager:true
org.apache.cxf.common.util.ASMHelperImpl:org.apache.cxf.common.util.ASMHelper:true
org.apache.cxf.common.spi.ClassLoaderProxyService:org.apache.cxf.common.spi.ClassLoaderService:true
-
+org.apache.cxf.io.DelayedCachedOutputStreamCleaner:org.apache.cxf.io.CachedOutputStreamCleaner:true
diff --git a/core/src/test/java/org/apache/cxf/io/DelayedCachedOutputStreamCleanerTest.java b/core/src/test/java/org/apache/cxf/io/DelayedCachedOutputStreamCleanerTest.java
new file mode 100644
index 0000000000..3f0603aa7f
--- /dev/null
+++ b/core/src/test/java/org/apache/cxf/io/DelayedCachedOutputStreamCleanerTest.java
@@ -0,0 +1,201 @@
+/**
+ * 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.cxf.io;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.bus.extension.ExtensionManagerBus;
+
+import org.junit.After;
+import org.junit.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class DelayedCachedOutputStreamCleanerTest {
+ private Bus bus;
+
+ @After
+ public void tearDown() {
+ if (bus != null) {
+ bus.shutdown(true);
+ bus = null;
+ }
+ }
+
+ @Test
+ public void testNoop() {
+ final Map<String, Object> properties = Collections.singletonMap(CachedConstants.CLEANER_DELAY_BUS_PROP, 0);
+ bus = new ExtensionManagerBus(new HashMap<>(), properties);
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class)); /* noop */
+
+ assertNoopCleaner(cleaner);
+ }
+
+ @Test
+ public void testForceClean() throws InterruptedException {
+ bus = new ExtensionManagerBus();
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class));
+
+ final AtomicBoolean latch = new AtomicBoolean(false);
+ final Closeable closeable = () -> latch.compareAndSet(false, true);
+ cleaner.register(closeable);
+
+ final DelayedCachedOutputStreamCleaner delayedCleaner = (DelayedCachedOutputStreamCleaner) cleaner;
+ delayedCleaner.forceClean();
+
+ // Await for Closeable::close to be called
+ assertThat(latch.get(), is(true));
+ }
+
+ @Test
+ public void testClean() throws InterruptedException {
+ final AtomicInteger latch = new AtomicInteger();
+ final Closeable closeable1 = () -> latch.incrementAndGet();
+ final Closeable closeable2 = () -> latch.incrementAndGet();
+
+ /* Delay of 2.5 seconds */
+ final Map<String, Object> properties = Collections.singletonMap(CachedConstants.CLEANER_DELAY_BUS_PROP, 2500);
+ bus = new ExtensionManagerBus(new HashMap<>(), properties);
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ cleaner.register(closeable1);
+ cleaner.register(closeable2);
+
+ // Await for Closeable::close to be called on schedule
+ await().atMost(5, TimeUnit.SECONDS).untilAtomic(latch, equalTo(2));
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class));
+ }
+
+ @Test
+ public void testForceCleanForEmpty() throws InterruptedException {
+ bus = new ExtensionManagerBus();
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class));
+
+ final AtomicBoolean latch = new AtomicBoolean(false);
+ final Closeable closeable = () -> latch.compareAndSet(false, true);
+
+ cleaner.register(closeable);
+ cleaner.unregister(closeable);
+
+ final DelayedCachedOutputStreamCleaner delayedCleaner = (DelayedCachedOutputStreamCleaner) cleaner;
+ delayedCleaner.forceClean();
+
+ // Closeable::close should not be called
+ assertThat(latch.get(), is(false));
+ }
+
+ @Test
+ public void testForceCleanException() throws InterruptedException {
+ bus = new ExtensionManagerBus();
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class));
+
+ final AtomicInteger latch = new AtomicInteger();
+ final Closeable closeable2 = () -> latch.incrementAndGet();
+ final Closeable closeable1 = () -> {
+ latch.incrementAndGet();
+ throw new IOException("Simulated");
+ };
+ cleaner.register(closeable1);
+ cleaner.register(closeable2);
+
+ final DelayedCachedOutputStreamCleaner delayedCleaner = (DelayedCachedOutputStreamCleaner) cleaner;
+ delayedCleaner.forceClean();
+
+ // Try to call force clean one more time
+ delayedCleaner.forceClean();
+
+ // Await for Closeable::close to be called
+ assertThat(latch.get(), equalTo(2));
+ }
+
+ @Test
+ public void testBusLifecycle() throws InterruptedException {
+ /* Delay of 2.5 seconds */
+ final Map<String, Object> properties = Collections.singletonMap(CachedConstants.CLEANER_DELAY_BUS_PROP, 2500);
+ bus = new ExtensionManagerBus(new HashMap<>(), properties);
+
+ final AtomicBoolean latch = new AtomicBoolean();
+ final Closeable closeable = () -> latch.compareAndSet(false, true);
+
+ bus.setProperty(CachedConstants.CLEANER_DELAY_BUS_PROP, 2500); /* 2.5 seconds */
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ cleaner.register(closeable);
+
+ // Closes the bus, the cleaner should cancel the internal timer(s)
+ bus.shutdown(true);
+
+ // The Closeable::close should not be called since timer(s) is cancelled
+ await().during(3, TimeUnit.SECONDS).untilAtomic(latch, is(false));
+ }
+
+ @Test
+ public void testNegativeDelay() throws InterruptedException {
+ final Map<String, Object> properties = Collections.singletonMap(CachedConstants.CLEANER_DELAY_BUS_PROP, -1);
+ bus = new ExtensionManagerBus(new HashMap<>(), properties);
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class)); /* noop */
+
+ assertNoopCleaner(cleaner);
+ }
+
+ @Test
+ public void testTooSmallDelay() throws InterruptedException {
+ final Map<String, Object> properties = Collections.singletonMap(CachedConstants.CLEANER_DELAY_BUS_PROP, 1500);
+ bus = new ExtensionManagerBus(new HashMap<>(), properties);
+
+ final CachedOutputStreamCleaner cleaner = bus.getExtension(CachedOutputStreamCleaner.class);
+ assertThat(cleaner, instanceOf(DelayedCachedOutputStreamCleaner.class)); /* noop */
+
+ assertNoopCleaner(cleaner);
+ }
+
+ private void assertNoopCleaner(final CachedOutputStreamCleaner cleaner) {
+ final AtomicBoolean latch = new AtomicBoolean(false);
+ final Closeable closeable = () -> latch.compareAndSet(false, true);
+ cleaner.register(closeable);
+
+ final DelayedCachedOutputStreamCleaner delayedCleaner = (DelayedCachedOutputStreamCleaner) cleaner;
+ delayedCleaner.forceClean();
+
+ // Noop, Closeable::close should not be called
+ assertThat(latch.get(), is(false));
+ }
+}
diff --git a/parent/pom.xml b/parent/pom.xml
index 81f39ba670..c40d0ee08a 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -277,7 +277,6 @@
<cxf.xmlresolver.bundle.version>1.2_5</cxf.xmlresolver.bundle.version>
<cxf.xmlsec.bundle.version>2.3.4</cxf.xmlsec.bundle.version>
<cxf.xpp3.bundle.version>1.1.4c_6</cxf.xpp3.bundle.version>
- <!-- Downgrade to 4.2.0 due to https://github.com/awaitility/awaitility/pull/279 -->
<cxf.awaitility.version>4.2.2</cxf.awaitility.version>
</properties>
<build>
diff --git a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
index b892ae8618..72a2734f6c 100644
--- a/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
+++ b/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
@@ -68,6 +68,8 @@ import org.apache.cxf.ext.logging.LoggingFeature;
import org.apache.cxf.frontend.ClientProxy;
import org.apache.cxf.helpers.FileUtils;
import org.apache.cxf.io.CachedOutputStream;
+import org.apache.cxf.io.CachedOutputStreamCleaner;
+import org.apache.cxf.io.DelayedCachedOutputStreamCleaner;
import org.apache.cxf.jaxws.DispatchImpl;
import org.apache.cxf.message.Message;
import org.apache.cxf.staxutils.StaxUtils;
@@ -126,9 +128,10 @@ public class ClientServerTest extends AbstractBusClientServerTestBase {
public static void startServers() throws Exception {
// set up configuration to enable schema validation
URL url = ClientServerTest.class.getResource("fault-stack-trace.xml");
+ // Create bus first so it will be shared between the server and clients
+ createStaticBus(url.toString());
assertNotNull("cannot find test resource", url);
assertTrue("server did not launch correctly", launchServer(Server.class, true));
- createStaticBus(url.toString());
}
@Test
@@ -1076,6 +1079,45 @@ public class ClientServerTest extends AbstractBusClientServerTestBase {
FileUtils.removeDir(f);
}
+ @Test
+ public void testEchoProviderThresholdTimeout() throws Exception {
+ final File f = Files.createTempDir();
+ LOG.info("Using temp folder: " + f.getAbsolutePath());
+
+ System.setProperty("org.apache.cxf.io.CachedOutputStream.OutputDirectory", f.getAbsolutePath());
+ CachedOutputStream.setDefaultThreshold(5);
+
+ String requestString = "<echo/>";
+ Service service = Service.create(serviceName);
+ service.addPort(fakePortName, javax.xml.ws.soap.SOAPBinding.SOAP11HTTP_BINDING,
+ "http://localhost:" + PORT + "/SoapContext/AsyncEchoProvider");
+ Dispatch<StreamSource> dispatcher = service.createDispatch(fakePortName,
+ StreamSource.class,
+ Service.Mode.PAYLOAD);
+ dispatcher.getRequestContext().put("jakarta.xml.ws.client.receiveTimeout", "1000");
+ dispatcher.getRequestContext().put("jakarta.xml.ws.client.connectionTimeout", "1000");
+
+ StreamSource request = new StreamSource(new ByteArrayInputStream(requestString.getBytes()));
+ try {
+ // Expecting java.net.SocketTimeoutException: Read timed out
+ StreamSource response = dispatcher.invoke(request);
+ assertEquals(requestString, StaxUtils.toString(response));
+ } catch (final WebServiceException ex) {
+ ((DispatchImpl<StreamSource>)dispatcher).getClient().close();
+ }
+
+ //give the server side a little time to process it's part and close the files
+ if (f.list().length > 0) {
+ final CachedOutputStreamCleaner cleaner = getBus().getExtension(CachedOutputStreamCleaner.class);
+ if (cleaner instanceof DelayedCachedOutputStreamCleaner) {
+ ((DelayedCachedOutputStreamCleaner) cleaner).forceClean();
+ }
+ }
+
+ assertEquals("Expected no files but there is at list one", 0, f.list().length);
+ FileUtils.removeDir(f);
+ }
+
@Test
public void testEchoProviderAsyncDecoupledEndpoints() throws Exception {
String requestString = "<echo/>";
(cxf) 01/03: Revert temporary excludion of some build agents from the label (due to persistent issues)
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.6.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 71075539b10db9c7dce387e7b4ef12945993dbf2
Author: Andriy Redko <dr...@gmail.com>
AuthorDate: Sat Sep 21 10:09:02 2024 -0400
Revert temporary excludion of some build agents from the label (due to persistent issues)
(cherry picked from commit 359808906918a77f696fe08a8db154e91c513411)
(cherry picked from commit 2cf863e831ac65fe85e0da40190ff3dbb181483b)
---
Jenkinsfile | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/Jenkinsfile b/Jenkinsfile
index 6d3e0fb079..cfb9d8a4a7 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -15,7 +15,7 @@ pipeline {
stages {
stage('Prepare') {
agent {
- label 'ubuntu && !builds55 && !builds56 && !builds57 && !builds58 && !builds59 && !builds60'
+ label 'ubuntu'
}
stages {
stage('Clean up') {
@@ -28,7 +28,7 @@ pipeline {
stage('Build') {
matrix {
agent {
- label 'ubuntu && !builds55 && !builds56 && !builds57 && !builds58 && !builds59 && !builds60'
+ label 'ubuntu'
}
axes {
axis {
@@ -39,7 +39,7 @@ pipeline {
stages {
stage('JDK specific build') {
agent {
- label 'ubuntu && !builds55 && !builds56 && !builds57 && !builds58 && !builds59 && !builds60'
+ label 'ubuntu'
}
tools {
jdk "${JAVA_VERSION}"
(cxf) 03/03: Recording .gitmergeinfo Changes
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.6.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit b544d0eb6966cb81b33095e6c7ae03ea12e0d31c
Author: Andriy Redko <dr...@gmail.com>
AuthorDate: Sun Sep 22 21:25:33 2024 -0400
Recording .gitmergeinfo Changes
---
.gitmergeinfo | 1 +
1 file changed, 1 insertion(+)
diff --git a/.gitmergeinfo b/.gitmergeinfo
index e2ce494997..8702a0641c 100644
--- a/.gitmergeinfo
+++ b/.gitmergeinfo
@@ -284,6 +284,7 @@ M 0a106ca8313663f324b57606a72ef6da76aa1ce0
M 0a1d2512da62b7d650d8ae6bef6a131f24746059
M 0a232967c6911527b83c0612d8d91b2e322d911e
M 0a35e128c6986a6c91e89b774b9a13841c397311
+M 0a97dded6720e83507811900df8f9c4bf2ff6471
M 0b44327f67cbc2874cb0a425659aab863a1d4caf
M 0b4a6fa2659ff62db8d8cc3f057ede60fccc9af7
M 0b53df25a3002b972a4cf65b890fa78e1e4989cc