From 8f5dac99b75c69076e9fdfe311ee9c05c369323c Mon Sep 17 00:00:00 2001 From: Lars Vogel Date: Fri, 7 Aug 2026 07:05:15 +0200 Subject: [PATCH] Remove unused org.eclipse.team.internal.core.streams Every caller of this package lived in org.eclipse.team.cvs.core or org.eclipse.team.cvs.ssh2, and all of them disappeared with 7592b18cbb "Retire CVS from master". The package was moved into org.eclipse.team.core by 68ca7959a2 precisely so the CVS core bundle and the ssh fragments could share one copy of the stream filters, so serving CVS was its only reason to live here. No non-CVS caller ever existed outside the package's own test. The eight stream classes form a closed cluster: the five orphaned by CVS plus PollingInputStream, which only SizeConstrainedInputStream used, and the two line-delimiter filters, whose only remaining client was StreamTests. That test goes away with them, along with its entry in AllTeamTests, the x-friends export of the package and the five message keys nothing references any more. The package was exported only via x-friends, naming the three deleted CVS bundles plus org.eclipse.team.ui and org.eclipse.team.tests.core, neither of which imports it. Removing it is therefore not an API change. Fixes https://github.com/eclipse-platform/eclipse.platform/issues/2856 --- .../META-INF/MANIFEST.MF | 6 - .../eclipse/team/internal/core/Messages.java | 5 - .../team/internal/core/messages.properties | 6 - .../core/streams/CRLFtoLFInputStream.java | 171 -------- .../core/streams/LFtoCRLFInputStream.java | 160 -------- .../core/streams/PollingInputStream.java | 236 ----------- .../core/streams/PollingOutputStream.java | 216 ----------- .../streams/ProgressMonitorInputStream.java | 149 ------- .../streams/SizeConstrainedInputStream.java | 161 -------- .../core/streams/TimeoutInputStream.java | 365 ------------------ .../core/streams/TimeoutOutputStream.java | 310 --------------- .../eclipse/team/tests/core/AllTeamTests.java | 1 - .../eclipse/team/tests/core/StreamTests.java | 81 ---- 13 files changed, 1867 deletions(-) delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/CRLFtoLFInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/LFtoCRLFInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingOutputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/ProgressMonitorInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/SizeConstrainedInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutInputStream.java delete mode 100644 team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutOutputStream.java delete mode 100644 team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/StreamTests.java diff --git a/team/bundles/org.eclipse.team.core/META-INF/MANIFEST.MF b/team/bundles/org.eclipse.team.core/META-INF/MANIFEST.MF index c563e3b9c3f..7c0997e0e1f 100644 --- a/team/bundles/org.eclipse.team.core/META-INF/MANIFEST.MF +++ b/team/bundles/org.eclipse.team.core/META-INF/MANIFEST.MF @@ -76,12 +76,6 @@ Export-Package: org.eclipse.team.core; org.eclipse.team.cvs.ui, org.eclipse.team.ui, org.eclipse.team.tests.core", - org.eclipse.team.internal.core.streams; - x-friends:="org.eclipse.team.cvs.core, - org.eclipse.team.cvs.ssh2, - org.eclipse.team.cvs.ui, - org.eclipse.team.ui, - org.eclipse.team.tests.core", org.eclipse.team.internal.core.subscribers; x-friends:="org.eclipse.team.cvs.core, org.eclipse.team.cvs.ssh2, diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/Messages.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/Messages.java index 4ca8fa91ccc..84f5b39acd2 100644 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/Messages.java +++ b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/Messages.java @@ -49,11 +49,6 @@ public class Messages extends NLS { public static String SubscriberDiffTreeEventHandler_0; public static String Team_readError; - public static String PollingInputStream_readTimeout; - public static String PollingInputStream_closeTimeout; - public static String PollingOutputStream_writeTimeout; - public static String PollingOutputStream_closeTimeout; - public static String TimeoutOutputStream_cannotWriteToStream; public static String RemoteSyncElement_delimit; public static String RemoteSyncElement_insync; diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/messages.properties b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/messages.properties index 3c4d13e9527..183308cb640 100644 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/messages.properties +++ b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/messages.properties @@ -34,12 +34,6 @@ RepositoryProvider_toString={0}:{1} Team_readError=An error occurred reading the state file ''{0}'' -PollingInputStream_readTimeout=Timeout while reading from input stream -PollingInputStream_closeTimeout=Timeout while closing input stream -PollingOutputStream_writeTimeout=Timeout while writing to output stream -PollingOutputStream_closeTimeout=Timeout while closing output stream -TimeoutOutputStream_cannotWriteToStream=Cannot write to output stream - RemoteSyncElement_delimit=[{0}] RemoteSyncElement_insync=in-sync RemoteSyncElement_conflicting=conflicting diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/CRLFtoLFInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/CRLFtoLFInputStream.java deleted file mode 100644 index b75d537d950..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/CRLFtoLFInputStream.java +++ /dev/null @@ -1,171 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -/** - * Converts CR/LFs in the underlying input stream to LF. - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public class CRLFtoLFInputStream extends FilterInputStream { - private boolean pendingByte = false; - private int lastByte = -1; - - /** - * Creates a new filtered input stream. - * @param in the underlying input stream - */ - public CRLFtoLFInputStream(InputStream in) { - super(in); - } - - /** - * Wraps the underlying stream's method. - * Translates CR/LF sequences to LFs transparently. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read() throws IOException { - if (! pendingByte) { - lastByte = in.read(); // ok if this throws - pendingByte = true; // remember the byte in case we throw an exception later on - } - if (lastByte == '\r') { - lastByte = in.read(); // ok if this throws - if (lastByte != '\n') { - if (lastByte == -1) { - pendingByte = false; - } - return '\r'; // leaves the byte pending for later - } - } - pendingByte = false; - return lastByte; - } - - /** - * Wraps the underlying stream's method. - * Translates CR/LF sequences to LFs transparently. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read(byte[] buffer, int off, int len) throws IOException { - // handle boundary cases cleanly - if (len == 0) { - return 0; - } else if (len == 1) { - int b = read(); - if (b == -1) { - return -1; - } - buffer[off] = (byte) b; - return 1; - } - // read some bytes from the stream - // prefix with pending byte from last read if any - int count = 0; - if (pendingByte) { - buffer[off] = (byte) lastByte; - pendingByte = false; - count = 1; - } - InterruptedIOException iioe = null; - try { - len = in.read(buffer, off + count, len - count); - if (len == -1) { - return (count == 0) ? -1 : count; - } - } catch (InterruptedIOException e) { - len = e.bytesTransferred; - iioe = e; - } - count += len; - // strip out CR's in CR/LF pairs - // pendingByte will be true iff previous byte was a CR - int j = off; - for (int i = off; i < off + count; ++i) { // invariant: j <= i - lastByte = buffer[i]; - if (lastByte == '\r') { - if (pendingByte) { - buffer[j++] = '\r'; // write out orphan CR - } else { - pendingByte = true; - } - } else { - if (pendingByte) { - if (lastByte != '\n') { - buffer[j++] = '\r'; // if LF, don't write the CR - } - pendingByte = false; - } - buffer[j++] = (byte) lastByte; - } - } - if (iioe != null) { - iioe.bytesTransferred = j - off; - throw iioe; - } - return j - off; - } - - /** - * Calls read() to skip the specified number of bytes - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public long skip(long count) throws IOException { - int actualCount = 0; // assumes count < Integer.MAX_INT - try { - while (count-- > 0 && read() != -1) { - actualCount++; // skip the specified number of bytes - } - return actualCount; - } catch (InterruptedIOException e) { - e.bytesTransferred = actualCount; - throw e; - } - } - - /** - * Wraps the underlying stream's method. - * Returns the number of bytes that can be read without blocking; accounts for - * possible translation of CR/LF sequences to LFs in these bytes. - * @throws IOException if an i/o error occurs - */ - @Override - public int available() throws IOException { - return in.available() / 2; // we can guarantee at least this amount after contraction - } - - /** - * Mark is not supported by the wrapper even if the underlying stream does, returns false. - */ - @Override - public boolean markSupported() { - return false; - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/LFtoCRLFInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/LFtoCRLFInputStream.java deleted file mode 100644 index a36b8edb1a4..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/LFtoCRLFInputStream.java +++ /dev/null @@ -1,160 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -/** - * Converts LFs in the underlying input stream to CR/LF. - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public class LFtoCRLFInputStream extends FilterInputStream { - private boolean mustReturnLF = false; - - /** - * Creates a new filtered input stream. - * @param in the underlying input stream - */ - public LFtoCRLFInputStream(InputStream in) { - super(in); - } - - /** - * Wraps the underlying stream's method. - * Translates LFs to CR/LF sequences transparently. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read() throws IOException { - if (mustReturnLF) { - mustReturnLF = false; - return '\n'; - } - int b = in.read(); // ok if this throws - if (b == '\n') { - mustReturnLF = true; - b = '\r'; - } - return b; - } - - /** - * Wraps the underlying stream's method. - * Translates LFs to CR/LF sequences transparently. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read(byte[] buffer, int off, int len) throws IOException { - // handle boundary cases cleanly - if (len == 0) { - return 0; - } else if (len == 1) { - int b = read(); - if (b == -1) { - return -1; - } - buffer[off] = (byte) b; - return 1; - } - // prefix with remembered \n from last read, but don't expand it a second time - int count = 0; - if (mustReturnLF) { - mustReturnLF = false; - buffer[off++] = '\n'; - --len; - count = 1; - if (len < 2) { - return count; // is there still enough room to expand more? - } - } - // read some bytes from the stream into the back half of the buffer - // this guarantees that there is always room to expand - len /= 2; - int j = off + len; - InterruptedIOException iioe = null; - try { - len = in.read(buffer, j, len); - if (len == -1) { - return (count == 0) ? -1 : count; - } - } catch (InterruptedIOException e) { - len = e.bytesTransferred; - iioe = e; - } - count += len; - // copy bytes from the middle to the front of the array, expanding LF->CR/LF - while (len-- > 0) { - byte b = buffer[j++]; - if (b == '\n') { - buffer[off++] = '\r'; - count++; - } - buffer[off++] = b; - } - if (iioe != null) { - iioe.bytesTransferred = count; - throw iioe; - } - return count; - } - - /** - * Calls read() to skip the specified number of bytes - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public long skip(long count) throws IOException { - int actualCount = 0; // assumes count < Integer.MAX_INT - try { - while (count-- > 0 && read() != -1) { - actualCount++; // skip the specified number of bytes - } - return actualCount; - } catch (InterruptedIOException e) { - e.bytesTransferred = actualCount; - throw e; - } - } - - /** - * Wraps the underlying stream's method. - * Returns the number of bytes that can be read without blocking; accounts for - * possible translation of LFs to CR/LF sequences in these bytes. - * @throws IOException if an i/o error occurs - */ - @Override - public int available() throws IOException { - return in.available(); // we can guarantee at least this amount after expansion - } - - /** - * Mark is not supported by the wrapper even if the underlying stream does, returns false. - */ - @Override - public boolean markSupported() { - return false; - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingInputStream.java deleted file mode 100644 index 22a8a951223..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingInputStream.java +++ /dev/null @@ -1,236 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -import org.eclipse.core.runtime.IProgressMonitor; -import org.eclipse.core.runtime.IStatus; -import org.eclipse.core.runtime.OperationCanceledException; -import org.eclipse.team.internal.core.Messages; -import org.eclipse.team.internal.core.Policy; -import org.eclipse.team.internal.core.TeamPlugin; - -/** - * Polls a progress monitor periodically and handles timeouts over extended durations. - * For this class to be effective, a high numAttempts should be specified, and the - * underlying stream should time out frequently on reads (every second or so). - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public class PollingInputStream extends FilterInputStream { - private final int numAttempts; - private final IProgressMonitor monitor; - private boolean cancellable; - - /** - * Creates a new polling input stream. - * @param in the underlying input stream - * @param numAttempts the number of attempts before issuing an InterruptedIOException, - * if 0, retries indefinitely until canceled - * @param monitor the progress monitor to be polled for cancellation - */ - public PollingInputStream(InputStream in, int numAttempts, IProgressMonitor monitor) { - super(in); - this.numAttempts = numAttempts; - this.monitor = monitor; - this.cancellable = true; - } - - /** - * Wraps the underlying stream's method. - * It may be important to wait for an input stream to be closed because it - * holds an implicit lock on a system resource (such as a file) while it is - * open. Closing a stream may take time if the underlying stream is still - * servicing a previous request. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times - */ - @Override - public void close() throws InterruptedIOException { - int attempts = 0; - try { - readPendingInput(); - } catch (IOException e) { - // We shouldn't get an exception when we're getting the available input. - // If we do, just log it so we can close. - TeamPlugin.log(IStatus.ERROR, e.getMessage(), e); - } finally { - boolean stop = false; - while (!stop) { - try { - if (in != null) { - in.close(); - } - stop = true; - } catch (InterruptedIOException e) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingInputStream_closeTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("close retry=" + attempts); //$NON-NLS-1$ - } - } catch (IOException e) { - // ignore it - see https://bugs.eclipse.org/bugs/show_bug.cgi?id=203423#c10 - } - } - } - } - - /** - * Wraps the underlying stream's method. - * @return the next byte of data, or -1 if the end of the stream is reached. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times - * and no data was received, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read() throws IOException { - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - return in.read(); - } catch (InterruptedIOException e) { - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingInputStream_readTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("read retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Wraps the underlying stream's method. - * @param buffer - the buffer into which the data is read. - * @param off - the start offset of the data. - * @param len - the maximum number of bytes read. - * @return the total number of bytes read into the buffer, or -1 if there is no more data because the end of the stream has been reached. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times - * and no data was received, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read(byte[] buffer, int off, int len) throws IOException { - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - return in.read(buffer, off, len); - } catch (InterruptedIOException e) { - if (e.bytesTransferred != 0) { - return e.bytesTransferred; // keep partial transfer - } - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingInputStream_readTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("read retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Wraps the underlying stream's method. - * @param count - the number of bytes to be skipped. - * @return the actual number of bytes skipped. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times - * and no data was received, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public long skip(long count) throws IOException { - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - return in.skip(count); - } catch (InterruptedIOException e) { - if (e.bytesTransferred != 0) { - return e.bytesTransferred; // keep partial transfer - } - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingInputStream_readTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("read retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Reads any pending input from the input stream so that - * the stream can savely be closed. - */ - protected void readPendingInput() throws IOException { - byte[] buffer= new byte[2048]; - while (true) { - int available = in.available(); - if (available < 1) { - break; - } - if (available > buffer.length) { - available = buffer.length; - } - if (in.read(buffer, 0, available) < 1) { - break; - } - } - } - - /** - * Called to set whether cancellation will be checked by this stream. Turning cancellation checking - * off can be very useful for protecting critical portions of a protocol that shouldn't be interrupted. - * For example, it is often necessary to protect login sequences. - * @param cancellable a flag controlling whether this stream will check for cancellation. - */ - public void setIsCancellable(boolean cancellable) { - this.cancellable = cancellable; - } - - /** - * Checked whether the monitor for this stream has been cancelled. If the cancellable - * flag is false then the monitor is never cancelled. - * @return true if the monitor has been cancelled and false - * otherwise. - */ - private boolean checkCancellation() { - if(cancellable) { - return monitor.isCanceled(); - } else { - return false; - } - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingOutputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingOutputStream.java deleted file mode 100644 index dbc52edf110..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/PollingOutputStream.java +++ /dev/null @@ -1,216 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterOutputStream; -import java.io.IOException; -import java.io.InterruptedIOException; -import java.io.OutputStream; - -import org.eclipse.core.runtime.IProgressMonitor; -import org.eclipse.core.runtime.OperationCanceledException; -import org.eclipse.team.internal.core.Messages; -import org.eclipse.team.internal.core.Policy; - -/** - * Polls a progress monitor periodically and handles timeouts over extended durations. - * For this class to be effective, a high numAttempts should be specified, and the - * underlying stream should time out frequently on writes (every second or so). - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public class PollingOutputStream extends FilterOutputStream { - private final int numAttempts; - private final IProgressMonitor monitor; - private boolean cancellable; - - /** - * Creates a new polling output stream. - * @param out the underlying output stream - * @param numAttempts the number of attempts before issuing an InterruptedIOException, - * if 0, retries indefinitely until canceled - * @param monitor the progress monitor to be polled for cancellation - */ - public PollingOutputStream(OutputStream out, int numAttempts, IProgressMonitor monitor) { - super(out); - this.numAttempts = numAttempts; - this.monitor = monitor; - this.cancellable = true; - } - - /** - * Wraps the underlying stream's method. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times - * and no data was sent, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public void write(int b) throws IOException { - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - out.write(b); - return; - } catch (InterruptedIOException e) { - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingOutputStream_writeTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("write retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Wraps the underlying stream's method. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times, - * bytesTransferred will reflect the number of bytes sent - * @throws IOException if an i/o error occurs - */ - @Override - public void write(byte[] buffer, int off, int len) throws IOException { - int count = 0; - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - out.write(buffer, off, len); - return; - } catch (InterruptedIOException e) { - int amount = e.bytesTransferred; - if (amount != 0) { // keep partial transfer - len -= amount; - if (len <= 0) { - return; - } - off += amount; - count += amount; - attempts = 0; // made some progress, don't time out quite yet - } - if (++attempts == numAttempts) { - e = new InterruptedIOException(Messages.PollingOutputStream_writeTimeout); - e.bytesTransferred = count; - throw e; - } - if (Policy.DEBUG_STREAMS) { - System.out.println("write retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Wraps the underlying stream's method. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times, - * bytesTransferred will reflect the number of bytes sent - * @throws IOException if an i/o error occurs - */ - @Override - public void flush() throws IOException { - int count = 0; - int attempts = 0; - for (;;) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - try { - out.flush(); - return; - } catch (InterruptedIOException e) { - int amount = e.bytesTransferred; - if (amount != 0) { // keep partial transfer - count += amount; - attempts = 0; // made some progress, don't time out quite yet - } - if (++attempts == numAttempts) { - e = new InterruptedIOException(Messages.PollingOutputStream_writeTimeout); - e.bytesTransferred = count; - throw e; - } - if (Policy.DEBUG_STREAMS) { - System.out.println("write retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - - /** - * Calls flush() then close() on the underlying stream. - * @throws OperationCanceledException if the progress monitor is canceled - * @throws InterruptedIOException if the underlying operation times out numAttempts times, - * bytesTransferred will reflect the number of bytes sent during the flush() - * @throws IOException if an i/o error occurs - */ - @Override - public void close() throws IOException { - int attempts = numAttempts - 1; // fail fast if flush() does times out - try { - out.flush(); - attempts = 0; - } finally { - boolean stop = false; - while (!stop) { - try { - out.close(); - stop = true; - } catch (InterruptedIOException e) { - if (checkCancellation()) { - throw new OperationCanceledException(); - } - if (++attempts == numAttempts) { - throw new InterruptedIOException(Messages.PollingOutputStream_closeTimeout); - } - if (Policy.DEBUG_STREAMS) { - System.out.println("close retry=" + attempts); //$NON-NLS-1$ - } - } - } - } - } - - /** - * Called to set whether cancellation will be checked by this stream. Turning cancellation checking - * off can be very useful for protecting critical portions of a protocol that shouldn't be interrupted. - * For example, it is often necessary to protect login sequences. - * @param cancellable a flag controlling whether this stream will check for cancellation. - */ - public void setIsCancellable(boolean cancellable) { - this.cancellable = cancellable; - } - - /** - * Checked whether the monitor for this stream has been cancelled. If the cancellable - * flag is false then the monitor is never cancelled. - * @return true if the monitor has been cancelled and false - * otherwise. - */ - private boolean checkCancellation() { - if(cancellable) { - return monitor.isCanceled(); - } else { - return false; - } - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/ProgressMonitorInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/ProgressMonitorInputStream.java deleted file mode 100644 index b02fb3a7205..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/ProgressMonitorInputStream.java +++ /dev/null @@ -1,149 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -import org.eclipse.core.runtime.IProgressMonitor; - -/** - * Updates a progress monitor as bytes are read from the input stream. - * Also starts a background thread to provide responsive cancellation on read(). - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public abstract class ProgressMonitorInputStream extends FilterInputStream { - private final IProgressMonitor monitor; - private final int updateIncrement; - private final long bytesTotal; - private long bytesRead = 0; - private long lastUpdate = -1; - private long nextUpdate = 0; - - /** - * Creates a progress monitoring input stream. - * @param in the underlying input stream - * @param bytesTotal the number of bytes to read in total (passed to updateMonitor()) - * @param updateIncrement the number of bytes read between updates - * @param monitor the progress monitor - */ - public ProgressMonitorInputStream(InputStream in, long bytesTotal, int updateIncrement, IProgressMonitor monitor) { - super(in); - this.bytesTotal = bytesTotal; - this.updateIncrement = updateIncrement; - this.monitor = monitor; - update(true); - } - - protected abstract void updateMonitor(long bytesRead, long size, IProgressMonitor monitor); - - /** - * Wraps the underlying stream's method. - * Updates the progress monitor to the final number of bytes read. - * @throws IOException if an i/o error occurs - */ - @Override - public void close() throws IOException { - try { - in.close(); - } finally { - update(true); - } - } - - /** - * Wraps the underlying stream's method. - * Updates the progress monitor if the next update increment has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read() throws IOException { - int b = in.read(); - if (b != -1) { - bytesRead += 1; - update(false); - } - return b; - } - - /** - * Wraps the underlying stream's method. - * Updates the progress monitor if the next update increment has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read(byte[] buffer, int offset, int length) throws IOException { - try { - int count = in.read(buffer, offset, length); - if (count != -1) { - bytesRead += count; - update(false); - } - return count; - } catch (InterruptedIOException e) { - bytesRead += e.bytesTransferred; - update(false); - throw e; - } - } - - /** - * Wraps the underlying stream's method. - * Updates the progress monitor if the next update increment has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public long skip(long amount) throws IOException { - try { - long count = in.skip(amount); - bytesRead += count; - update(false); - return count; - } catch (InterruptedIOException e) { - bytesRead += e.bytesTransferred; - update(false); - throw e; - } - } - - /** - * Mark is not supported by the wrapper even if the underlying stream does, returns false. - */ - @Override - public boolean markSupported() { - return false; - } - - private void update(boolean now) { - if (bytesRead >= nextUpdate || now) { - nextUpdate = bytesRead - (bytesRead % updateIncrement); - if (nextUpdate != lastUpdate) { - updateMonitor(nextUpdate, bytesTotal, monitor); - } - lastUpdate = nextUpdate; - nextUpdate += updateIncrement; - } - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/SizeConstrainedInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/SizeConstrainedInputStream.java deleted file mode 100644 index 9bc2a1b2616..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/SizeConstrainedInputStream.java +++ /dev/null @@ -1,161 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -import org.eclipse.core.runtime.OperationCanceledException; - -/** - * Simulates a stream that represents only a portion of the underlying stream. - * Will report EOF when this portion has been fully read and prevent further reads. - * The underlying stream is not closed on close(), but the remaining unread input - * may optionally be skip()'d. - * - * Supports resuming partially completed operations after an InterruptedIOException - * if the underlying stream does. Check the bytesTransferred field to determine how - * much of the operation completed; conversely, at what point to resume. - */ -public class SizeConstrainedInputStream extends FilterInputStream { - private final boolean discardOnClose; - private long bytesRemaining; - - /** - * Creates a size constrained input stream. - * @param in the underlying input stream, never actually closed by this filter - * @param size the maximum number of bytes of the underlying input stream that - * can be read through this filter - * @param discardOnClose if true, discards remaining unread bytes on close() - */ - public SizeConstrainedInputStream(InputStream in, long size, boolean discardOnClose) { - super(in); - this.bytesRemaining = size; - this.discardOnClose = discardOnClose; - } - - /** - * Prevents further reading from the stream but does not close the underlying stream. - * If discardOnClose, skip()'s over any remaining unread bytes in the constrained region. - * @throws IOException if an i/o error occurs - */ - @Override - public void close() throws IOException { - try { - if (discardOnClose) { - while (bytesRemaining != 0 && skip(bytesRemaining) != 0) { - ; - } - } - } catch (OperationCanceledException e) { - // The receiver is likely wrapping a PollingInputStream which could throw - // an OperationCanceledException on a skip. - // Since we're closing, just ignore the cancel and let the caller check the monitor - } finally { - bytesRemaining = 0; - } - } - - /** - * Wraps the underlying stream's method. - * Simulates an end-of-file condition if the end of the constrained region has been reached. - * @throws IOException if an i/o error occurs - */ - @Override - public int available() throws IOException { - int amount = in.available(); - if (amount > bytesRemaining) { - amount = (int) bytesRemaining; - } - return amount; - } - - /** - * Wraps the underlying stream's method. - * Simulates an end-of-file condition if the end of the constrained region has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read() throws IOException { - if (bytesRemaining == 0) { - return -1; - } - int b = in.read(); - if (b != -1) { - bytesRemaining -= 1; - } - return b; - } - - /** - * Wraps the underlying stream's method. - * Simulates an end-of-file condition if the end of the constrained region has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public int read(byte[] buffer, int offset, int length) throws IOException { - if (length > bytesRemaining) { - if (bytesRemaining == 0) { - return -1; - } - length = (int) bytesRemaining; - } - try { - int count = in.read(buffer, offset, length); - if (count != -1) { - bytesRemaining -= count; - } - return count; - } catch (InterruptedIOException e) { - bytesRemaining -= e.bytesTransferred; - throw e; - } - } - - /** - * Wraps the underlying stream's method. - * Simulates an end-of-file condition if the end of the constrained region has been reached. - * @throws InterruptedIOException if the operation was interrupted before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public long skip(long amount) throws IOException { - if (amount > bytesRemaining) { - amount = bytesRemaining; - } - try { - long count = in.skip(amount); - bytesRemaining -= count; - return count; - } catch (InterruptedIOException e) { - bytesRemaining -= e.bytesTransferred; - throw e; - } - } - - /** - * Mark is not supported by the wrapper even if the underlying stream does, returns false. - */ - @Override - public boolean markSupported() { - return false; - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutInputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutInputStream.java deleted file mode 100644 index 39b5485cf7e..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutInputStream.java +++ /dev/null @@ -1,365 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.FilterInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.InterruptedIOException; - -import org.eclipse.team.internal.core.Policy; - -/** - * Wraps an input stream that blocks indefinitely to simulate timeouts on read(), - * skip(), and close(). The resulting input stream is buffered and supports - * retrying operations that failed due to an InterruptedIOException. - * - * Supports resuming partially completed operations after an InterruptedIOException - * REGARDLESS of whether the underlying stream does unless the underlying stream itself - * generates InterruptedIOExceptions in which case it must also support resuming. - * Check the bytesTransferred field to determine how much of the operation completed; - * conversely, at what point to resume. - */ -public class TimeoutInputStream extends FilterInputStream { - // unsynchronized variables - private final long readTimeout; // read() timeout in millis - private final long closeTimeout; // close() timeout in millis, or -1 - - // requests for the thread (synchronized) - private boolean closeRequested = false; // if true, close requested - - // responses from the thread (synchronized) - private Thread thread; // if null, thread has terminated - private byte[] iobuffer; // circular buffer - private int head = 0; // points to first unread byte - private int length = 0; // number of remaining unread bytes - private IOException ioe = null; // if non-null, contains a pending exception - private boolean waitingForClose = false; // if true, thread is waiting for close() - - private boolean growWhenFull = false; // if true, buffer will grow when it is full - - /** - * Creates a timeout wrapper for an input stream. - * @param in the underlying input stream - * @param bufferSize the buffer size in bytes; should be large enough to mitigate - * Thread synchronization and context switching overhead - * @param readTimeout the number of milliseconds to block for a read() or skip() before - * throwing an InterruptedIOException; 0 blocks indefinitely - * @param closeTimeout the number of milliseconds to block for a close() before throwing - * an InterruptedIOException; 0 blocks indefinitely, -1 closes the stream in the background - */ - public TimeoutInputStream(InputStream in, int bufferSize, long readTimeout, long closeTimeout) { - super(in); - this.readTimeout = readTimeout; - this.closeTimeout = closeTimeout; - this.iobuffer = new byte[bufferSize]; - thread = new Thread((Runnable) this::runThread, "TimeoutInputStream");//$NON-NLS-1$ - thread.setDaemon(true); - thread.start(); - } - - public TimeoutInputStream(InputStream in, int bufferSize, long readTimeout, long closeTimeout, boolean growWhenFull) { - this(in, bufferSize, readTimeout, closeTimeout); - this.growWhenFull = growWhenFull; - } - - /** - * Wraps the underlying stream's method. - * It may be important to wait for a stream to actually be closed because it - * holds an implicit lock on a system resoure (such as a file) while it is - * open. Closing a stream may take time if the underlying stream is still - * servicing a previous request. - * @throws InterruptedIOException if the timeout expired - * @throws IOException if an i/o error occurs - */ - @Override - public void close() throws IOException { - Thread oldThread; - synchronized (this) { - if (thread == null) { - return; - } - oldThread = thread; - closeRequested = true; - thread.interrupt(); - checkError(); - } - if (closeTimeout == -1) { - return; - } - try { - oldThread.join(closeTimeout); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); // we weren't expecting to be interrupted - } - synchronized (this) { - checkError(); - if (thread != null) { - throw new InterruptedIOException(); - } - } - } - - /** - * Returns the number of unread bytes in the buffer. - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized int available() throws IOException { - if (length == 0) { - checkError(); - } - return length > 0 ? length : 0; - } - - /** - * Reads a byte from the stream. - * @throws InterruptedIOException if the timeout expired and no data was received, - * bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized int read() throws IOException { - if (! syncFill()) { - return -1; // EOF reached - } - int b = iobuffer[head++] & 255; - if (head == iobuffer.length) { - head = 0; - } - length--; - notify(); - return b; - } - - /** - * Reads multiple bytes from the stream. - * @throws InterruptedIOException if the timeout expired and no data was received, - * bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized int read(byte[] buffer, int off, int len) throws IOException { - if (! syncFill()) { - return -1; // EOF reached - } - int pos = off; - if (len > length) { - len = length; - } - while (len-- > 0) { - buffer[pos++] = iobuffer[head++]; - if (head == iobuffer.length) { - head = 0; - } - length--; - } - notify(); - return pos - off; - } - - /** - * Skips multiple bytes in the stream. - * @throws InterruptedIOException if the timeout expired before all of the - * bytes specified have been skipped, bytesTransferred may be non-zero - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized long skip(long count) throws IOException { - long amount = 0; - try { - do { - if (! syncFill()) { - break; // EOF reached - } - int skip = (int) Math.min(count - amount, length); - head = (head + skip) % iobuffer.length; - length -= skip; - amount += skip; - } while (amount < count); - } catch (InterruptedIOException e) { - e.bytesTransferred = (int) amount; // assumes amount < Integer.MAX_INT - throw e; - } - notify(); - return amount; - } - - /** - * Mark is not supported by the wrapper even if the underlying stream does, returns false. - */ - @Override - public boolean markSupported() { - return false; - } - - /** - * Waits for the buffer to fill if it is empty and the stream has not reached EOF. - * @return true if bytes are available, false if EOF has been reached - * @throws InterruptedIOException if EOF not reached but no bytes are available - */ - private boolean syncFill() throws IOException { - if (length != 0) { - return true; - } - checkError(); // check errors only after we have read all remaining bytes - if (waitingForClose) { - return false; - } - notify(); - try { - wait(readTimeout); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); // we weren't expecting to be interrupted - } - if (length != 0) { - return true; - } - checkError(); // check errors only after we have read all remaining bytes - if (waitingForClose) { - return false; - } - throw new InterruptedIOException(); - } - - /** - * If an exception is pending, throws it. - */ - private void checkError() throws IOException { - if (ioe != null) { - IOException e = ioe; - ioe = null; - throw e; - } - } - - /** - * Runs the thread in the background. - */ - private void runThread() { - try { - readUntilDone(); - } catch (IOException e) { - synchronized (this) { ioe = e; } - } finally { - waitUntilClosed(); - try { - in.close(); - } catch (IOException e) { - synchronized (this) { ioe = e; } - } finally { - synchronized (this) { - thread = null; - notify(); - } - } - } - } - - /** - * Waits until we have been requested to close the stream. - */ - private synchronized void waitUntilClosed() { - waitingForClose = true; - notify(); - while (! closeRequested) { - try { - wait(); - } catch (InterruptedException e) { - closeRequested = true; // alternate quit signal - } - } - } - - /** - * Reads bytes into the buffer until EOF, closed, or error. - */ - private void readUntilDone() throws IOException { - for (;;) { - int off, len; - synchronized (this) { - while (isBufferFull()) { - if (closeRequested) { - return; // quit signal - } - waitForRead(); - } - off = (head + length) % iobuffer.length; - len = ((head > off) ? head : iobuffer.length) - off; - } - int count; - try { - // the i/o operation might block without releasing the lock, - // so we do this outside of the synchronized block - count = in.read(iobuffer, off, len); - if (count == -1) { - return; // EOF encountered - } - } catch (InterruptedIOException e) { - count = e.bytesTransferred; // keep partial transfer - } - synchronized (this) { - length += count; - notify(); - } - } - } - - /* - * Wait for a read when the buffer is full (with the implication - * that space will become available in the buffer after the read - * takes place). - */ - private synchronized void waitForRead() { - try { - if (growWhenFull) { - // wait a second before growing to let reads catch up - wait(readTimeout); - } else { - wait(); - } - } catch (InterruptedException e) { - closeRequested = true; // alternate quit signal - } - // If the buffer is still full, give it a chance to grow - if (growWhenFull && isBufferFull()) { - growBuffer(); - } - } - - private synchronized void growBuffer() { - int newSize = 2 * iobuffer.length; - if (newSize > iobuffer.length) { - if (Policy.DEBUG_STREAMS) { - System.out.println("InputStream growing to " + newSize + " bytes"); //$NON-NLS-1$ //$NON-NLS-2$ - } - byte[] newBuffer = new byte[newSize]; - int pos = 0; - int len = length; - while (len-- > 0) { - newBuffer[pos++] = iobuffer[head++]; - if (head == iobuffer.length) { - head = 0; - } - } - iobuffer = newBuffer; - head = 0; - // length instance variable was not changed by this method - } - } - - private boolean isBufferFull() { - return length == iobuffer.length; - } -} diff --git a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutOutputStream.java b/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutOutputStream.java deleted file mode 100644 index 190c805f37d..00000000000 --- a/team/bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/streams/TimeoutOutputStream.java +++ /dev/null @@ -1,310 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2017 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.internal.core.streams; - -import java.io.BufferedOutputStream; -import java.io.FilterOutputStream; -import java.io.IOException; -import java.io.InterruptedIOException; -import java.io.OutputStream; - -import org.eclipse.team.internal.core.Messages; - -/** - * Wraps an output stream that blocks indefinitely to simulate timeouts on write(), - * flush(), and close(). The resulting output stream is buffered and supports - * retrying operations that failed due to an InterruptedIOException. - * - * Supports resuming partially completed operations after an InterruptedIOException - * REGARDLESS of whether the underlying stream does unless the underlying stream itself - * generates InterruptedIOExceptions in which case it must also support resuming. - * Check the bytesTransferred field to determine how much of the operation completed; - * conversely, at what point to resume. - */ -public class TimeoutOutputStream extends FilterOutputStream { - // unsynchronized variables - private final long writeTimeout; // write() timeout in millis - private final long closeTimeout; // close() timeout in millis, or -1 - - // requests for the thread (synchronized) - private final byte[] iobuffer; // circular buffer - private int head = 0; // points to first unwritten byte - private int length = 0; // number of remaining unwritten bytes - private boolean closeRequested = false; // if true, close requested - private boolean flushRequested = false; // if true, flush requested - - // responses from the thread (synchronized) - private Thread thread; - private boolean waitingForClose = false; // if true, the thread is waiting for close() - private IOException ioe = null; - - /** - * Creates a timeout wrapper for an output stream. - * @param out the underlying input stream - * @param bufferSize the buffer size in bytes; should be large enough to mitigate - * Thread synchronization and context switching overhead - * @param writeTimeout the number of milliseconds to block for a write() or flush() before - * throwing an InterruptedIOException; 0 blocks indefinitely - * @param closeTimeout the number of milliseconds to block for a close() before throwing - * an InterruptedIOException; 0 blocks indefinitely, -1 closes the stream in the background - */ - public TimeoutOutputStream(OutputStream out, int bufferSize, long writeTimeout, long closeTimeout) { - super(new BufferedOutputStream(out, bufferSize)); - this.writeTimeout = writeTimeout; - this.closeTimeout = closeTimeout; - this.iobuffer = new byte[bufferSize]; - thread = new Thread((Runnable) this::runThread, "TimeoutOutputStream");//$NON-NLS-1$ - thread.setDaemon(true); - thread.start(); - } - - /** - * Wraps the underlying stream's method. - * It may be important to wait for a stream to actually be closed because it - * holds an implicit lock on a system resoure (such as a file) while it is - * open. Closing a stream may take time if the underlying stream is still - * servicing a previous request. - * @throws InterruptedIOException if the timeout expired, bytesTransferred will - * reflect the number of bytes flushed from the buffer - * @throws IOException if an i/o error occurs - */ - @Override - public void close() throws IOException { - Thread oldThread; - synchronized (this) { - if (thread == null) { - return; - } - oldThread = thread; - closeRequested = true; - thread.interrupt(); - checkError(); - } - if (closeTimeout == -1) { - return; - } - try { - oldThread.join(closeTimeout); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); // we weren't expecting to be interrupted - } - synchronized (this) { - checkError(); - if (thread != null) { - throw new InterruptedIOException(); - } - } - } - - /** - * Writes a byte to the stream. - * @throws InterruptedIOException if the timeout expired and no data was sent, - * bytesTransferred will be zero - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized void write(int b) throws IOException { - syncCommit(true); - iobuffer[(head + length) % iobuffer.length] = (byte) b; - length++; - notify(); - } - - /** - * Writes multiple bytes to the stream. - * @throws InterruptedIOException if the timeout expired, bytesTransferred will - * reflect the number of bytes sent - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized void write(byte[] buffer, int off, int len) throws IOException { - int amount = 0; - try { - do { - syncCommit(true); - while (amount < len && length != iobuffer.length) { - iobuffer[(head + length) % iobuffer.length] = buffer[off++]; - length++; - amount++; - } - } while (amount < len); - } catch (InterruptedIOException e) { - e.bytesTransferred = amount; - throw e; - } - notify(); - } - - /** - * Flushes the stream. - * @throws InterruptedIOException if the timeout expired, bytesTransferred will - * reflect the number of bytes flushed from the buffer - * @throws IOException if an i/o error occurs - */ - @Override - public synchronized void flush() throws IOException { - int oldLength = length; - flushRequested = true; - try { - syncCommit(false); - } catch (InterruptedIOException e) { - e.bytesTransferred = oldLength - length; - throw e; - } - notify(); - } - - /** - * Waits for the buffer to drain if it is full. - * @param partial if true, waits until the buffer is partially empty, else drains it entirely - * @throws InterruptedIOException if the buffer could not be drained as requested - */ - private void syncCommit(boolean partial) throws IOException { - checkError(); // check errors before allowing the addition of new bytes - if (partial && length != iobuffer.length || length == 0) { - return; - } - if (waitingForClose) { - throw new IOException(Messages.TimeoutOutputStream_cannotWriteToStream); - } - notify(); - try { - wait(writeTimeout); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); // we weren't expecting to be interrupted - } - checkError(); // check errors before allowing the addition of new bytes - if (partial && length != iobuffer.length || length == 0) { - return; - } - throw new InterruptedIOException(); - } - - /** - * If an exception is pending, throws it. - */ - private void checkError() throws IOException { - if (ioe != null) { - IOException e = ioe; - ioe = null; - throw e; - } - } - - /** - * Runs the thread in the background. - */ - private void runThread() { - try { - writeUntilDone(); - } catch (IOException e) { - synchronized (this) { ioe = e; } - } finally { - waitUntilClosed(); - try { - out.close(); - } catch (IOException e) { - synchronized (this) { ioe = e; } - } finally { - synchronized (this) { - thread = null; - notify(); - } - } - } - } - - /** - * Waits until we have been requested to close the stream. - */ - private synchronized void waitUntilClosed() { - waitingForClose = true; - notify(); - while (! closeRequested) { - try { - wait(); - } catch (InterruptedException e) { - closeRequested = true; // alternate quit signal - } - } - } - - /** - * Writes bytes from the buffer until closed and buffer is empty - */ - private void writeUntilDone() throws IOException { - int bytesUntilFlush = -1; // if > 0, then we will flush after that many bytes have been written - for (;;) { - int off, len; - synchronized (this) { - for (;;) { - if (closeRequested && length == 0) { - return; // quit signal - } - if (length != 0 || flushRequested) { - break; - } - try { - wait(); - } catch (InterruptedException e) { - closeRequested = true; // alternate quit signal - } - } - off = head; - len = iobuffer.length - head; - if (len > length) { - len = length; - } - if (flushRequested && bytesUntilFlush < 0) { - flushRequested = false; - bytesUntilFlush = length; - } - } - - // If there are bytes to be written, write them - if (len != 0) { - // write out all remaining bytes from the buffer before flushing - try { - // the i/o operation might block without releasing the lock, - // so we do this outside of the synchronized block - out.write(iobuffer, off, len); - } catch (InterruptedIOException e) { - len = e.bytesTransferred; - } - } - - // If there was a pending flush, do it - if (bytesUntilFlush >= 0) { - bytesUntilFlush -= len; - if (bytesUntilFlush <= 0) { - // flush the buffer now - try { - out.flush(); - } catch (InterruptedIOException e) { - } - bytesUntilFlush = -1; // might have been 0 - } - } - - // If bytes were written, update the circular buffer - if (len != 0) { - synchronized (this) { - head = (head + len) % iobuffer.length; - length -= len; - notify(); - } - } - } - } -} diff --git a/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/AllTeamTests.java b/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/AllTeamTests.java index 8030468e172..94aa760a7a0 100644 --- a/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/AllTeamTests.java +++ b/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/AllTeamTests.java @@ -24,7 +24,6 @@ RefreshSubscriberParticipantJobProgressTests.class, // RepositoryProviderTests.class, // StorageMergerTests.class, // - StreamTests.class, // UserMappingTest.class, // }) public class AllTeamTests { diff --git a/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/StreamTests.java b/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/StreamTests.java deleted file mode 100644 index 455967df4e5..00000000000 --- a/team/tests/org.eclipse.team.tests.core/src/org/eclipse/team/tests/core/StreamTests.java +++ /dev/null @@ -1,81 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2000, 2006 IBM Corporation and others. - * - * This program and the accompanying materials - * are made available under the terms of the Eclipse Public License 2.0 - * which accompanies this distribution, and is available at - * https://www.eclipse.org/legal/epl-2.0/ - * - * SPDX-License-Identifier: EPL-2.0 - * - * Contributors: - * IBM Corporation - initial API and implementation - *******************************************************************************/ -package org.eclipse.team.tests.core; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; - -import org.eclipse.team.internal.core.streams.CRLFtoLFInputStream; -import org.eclipse.team.internal.core.streams.LFtoCRLFInputStream; -import org.junit.jupiter.api.Test; - -public class StreamTests { - - @Test - public void testCRLFtoLFInputStream() throws IOException { - testCRLFtoLFTranslation("", ""); - testCRLFtoLFTranslation("a", "a"); - testCRLFtoLFTranslation("abc", "abc"); - testCRLFtoLFTranslation("\n", "\n"); - testCRLFtoLFTranslation("\r", "\r"); - testCRLFtoLFTranslation("\r\n", "\n"); - testCRLFtoLFTranslation("x\r\r\n\rx", "x\r\n\rx"); - testCRLFtoLFTranslation("The \r\n quick brown \n fox \r\n\n\r\r\n jumped \n\n over \r\n the \n lazy dog.\r\n", - "The \n quick brown \n fox \n\n\r\n jumped \n\n over \n the \n lazy dog.\n"); - } - - private void testCRLFtoLFTranslation(String pre, String post) throws IOException { - ByteArrayInputStream bin = new ByteArrayInputStream(pre.getBytes()); - InputStream in = new CRLFtoLFInputStream(bin); - InputStream inExpected = new ByteArrayInputStream(post.getBytes()); - assertStreamEquals(inExpected, in); - } - - @Test - public void testLFtoCRLFInputStream() throws IOException { - testLFtoCRLFTranslation("", ""); - testLFtoCRLFTranslation("a", "a"); - testLFtoCRLFTranslation("abc", "abc"); - testLFtoCRLFTranslation("\n", "\r\n"); - testLFtoCRLFTranslation("\r", "\r"); - testLFtoCRLFTranslation("\r\n", "\r\r\n"); - testLFtoCRLFTranslation("x\r\r\n\rx", "x\r\r\r\n\rx"); - testLFtoCRLFTranslation("The \r\n quick brown \n fox \r\n\n\r\r\n jumped \n\n over \r\n the \n lazy dog.\r\n", - "The \r\r\n quick brown \r\n fox \r\r\n\r\n\r\r\r\n jumped \r\n\r\n over \r\r\n the \r\n lazy dog.\r\r\n"); - } - - private void testLFtoCRLFTranslation(String pre, String post) throws IOException { - ByteArrayInputStream bin = new ByteArrayInputStream(pre.getBytes()); - InputStream in = new LFtoCRLFInputStream(bin); - InputStream inExpected = new ByteArrayInputStream(post.getBytes()); - assertStreamEquals(inExpected, in); - } - - private void assertStreamEquals(InputStream in1, InputStream in2) throws IOException { - try { - for (;;) { - int byte1 = in1.read(); - int byte2 = in2.read(); - assertEquals(byte1, byte2, "Streams not equal"); - if (byte1 == -1) break; - } - } finally { - in1.close(); - in2.close(); - } - } -}