This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 4.0.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 0a97dded6720e83507811900df8f9c4bf2ff6471 Author: Andriy Redko <[email protected]> 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) --- core/pom.xml | 5 + .../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(+), 4 deletions(-) diff --git a/core/pom.xml b/core/pom.xml index 54709eba21..8ac8722025 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -180,6 +180,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..04a258a558 --- /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 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 jakarta.annotation.Resource; +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 7abb5a478d..d17717c1f6 100644 --- a/parent/pom.xml +++ b/parent/pom.xml @@ -275,7 +275,6 @@ <cxf.wsdl4j.bundle.version>1.6.3_1</cxf.wsdl4j.bundle.version> <cxf.xmlresolver.bundle.version>1.2_5</cxf.xmlresolver.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 47dfd025f4..89e551e21a 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, jakarta.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/>";
