From efd5985d91c8c1b50aa782c3a938b30db9e27a0b Mon Sep 17 00:00:00 2001 From: Konstantin Kolosovsky Date: Sun, 24 Nov 2013 17:23:47 +0400 Subject: [PATCH] IDEA-113730 Implemented "cat" command to read content as binary data (not to read as text data and then convert to bytes using some encoding) --- .../process/BaseOSProcessHandler.java | 25 +++- .../com/intellij/util/io/BaseDataReader.java | 139 ++++++++++++++++++ .../intellij/util/io/BaseOutputReader.java | 114 +------------- .../intellij/util/io/BinaryOutputReader.java | 58 ++++++++ .../idea/svn/commandLine/CommandExecutor.java | 52 ++++++- .../idea/svn/content/CmdContentClient.java | 9 +- 6 files changed, 277 insertions(+), 120 deletions(-) create mode 100644 platform/util/src/com/intellij/util/io/BaseDataReader.java create mode 100644 platform/util/src/com/intellij/util/io/BinaryOutputReader.java diff --git a/platform/util/src/com/intellij/execution/process/BaseOSProcessHandler.java b/platform/util/src/com/intellij/execution/process/BaseOSProcessHandler.java index a44d8f76f04d..463708e1b9af 100644 --- a/platform/util/src/com/intellij/execution/process/BaseOSProcessHandler.java +++ b/platform/util/src/com/intellij/execution/process/BaseOSProcessHandler.java @@ -20,6 +20,7 @@ import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.util.Key; import com.intellij.util.ConcurrencyUtil; import com.intellij.util.Consumer; +import com.intellij.util.io.BaseDataReader; import com.intellij.util.io.BaseOutputReader; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -28,7 +29,7 @@ import java.io.*; import java.nio.charset.Charset; import java.util.concurrent.*; -import static com.intellij.util.io.BaseOutputReader.AdaptiveSleepingPolicy; +import static com.intellij.util.io.BaseDataReader.AdaptiveSleepingPolicy; public class BaseOSProcessHandler extends ProcessHandler implements TaskExecutor { private static final Logger LOG = Logger.getInstance("#com.intellij.execution.process.OSProcessHandlerBase"); @@ -82,13 +83,10 @@ public class BaseOSProcessHandler extends ProcessHandler implements TaskExecutor @Override public void startNotified(final ProcessEvent event) { try { - BaseOutputReader.SleepingPolicy sleepingPolicy = - useAdaptiveSleepingPolicyWhenReadingOutput() ? new AdaptiveSleepingPolicy() : BaseOutputReader.SleepingPolicy.SIMPLE; - final BaseOutputReader stdoutReader = new SimpleOutputReader(createProcessOutReader(), ProcessOutputTypes.STDOUT, sleepingPolicy); - final BaseOutputReader stderrReader = processHasSeparateErrorStream() - ? new SimpleOutputReader(createProcessErrReader(), ProcessOutputTypes.STDERR, - sleepingPolicy) - : null; + BaseDataReader.SleepingPolicy sleepingPolicy = + useAdaptiveSleepingPolicyWhenReadingOutput() ? new AdaptiveSleepingPolicy() : BaseDataReader.SleepingPolicy.SIMPLE; + final BaseDataReader stdoutReader = createOutputDataReader(sleepingPolicy); + final BaseDataReader stderrReader = processHasSeparateErrorStream() ? createErrorDataReader(sleepingPolicy) : null; myWaitFor.setTerminationCallback(new Consumer() { @Override @@ -120,6 +118,17 @@ public class BaseOSProcessHandler extends ProcessHandler implements TaskExecutor super.startNotify(); } + @NotNull + protected BaseDataReader createErrorDataReader(BaseDataReader.SleepingPolicy sleepingPolicy) { + return new SimpleOutputReader(createProcessErrReader(), ProcessOutputTypes.STDERR, + sleepingPolicy); + } + + @NotNull + protected BaseDataReader createOutputDataReader(BaseDataReader.SleepingPolicy sleepingPolicy) { + return new SimpleOutputReader(createProcessOutReader(), ProcessOutputTypes.STDOUT, sleepingPolicy); + } + protected void onOSProcessTerminated(final int exitCode) { notifyProcessTerminated(exitCode); } diff --git a/platform/util/src/com/intellij/util/io/BaseDataReader.java b/platform/util/src/com/intellij/util/io/BaseDataReader.java new file mode 100644 index 000000000000..5b0226667edd --- /dev/null +++ b/platform/util/src/com/intellij/util/io/BaseDataReader.java @@ -0,0 +1,139 @@ +/* + * Copyright 2000-2013 JetBrains s.r.o. + * + * Licensed 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 com.intellij.util.io; + +import com.intellij.openapi.diagnostic.Logger; +import com.intellij.util.TimeoutUtil; +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +/** + * @author Konstantin Kolosovsky. + */ +public abstract class BaseDataReader { + private static final Logger LOG = Logger.getInstance(BaseDataReader.class); + + protected volatile boolean isStopped = false; + + private Future myFinishedFuture = null; + @NotNull protected final SleepingPolicy mySleepingPolicy; + + public BaseDataReader(SleepingPolicy sleepingPolicy) { + mySleepingPolicy = sleepingPolicy != null ? sleepingPolicy: SleepingPolicy.SIMPLE; + } + + protected void start() { + if (myFinishedFuture == null) { + myFinishedFuture = executeOnPooledThread(new Runnable() { + @Override + public void run() { + doRun(); + } + }); + } + } + + protected abstract Future executeOnPooledThread(Runnable runnable); + + public interface SleepingPolicy { + int sleepTimeWhenWasActive = 1; + int sleepTimeWhenIdle = 5; + + SleepingPolicy SIMPLE = new SleepingPolicy() { + @Override + public int getTimeToSleep(boolean wasActive) { + return wasActive ? sleepTimeWhenWasActive : sleepTimeWhenIdle; + } + }; + + int getTimeToSleep(boolean wasActive); + } + + public static class AdaptiveSleepingPolicy implements SleepingPolicy { + private static final int maxSleepTimeWhenIdle = 200; + private static final int maxIterationsWithCurrentSleepTime = 50; + + private volatile int myIterationsWithCurrentTime; + private volatile int myCurrentSleepTime = sleepTimeWhenIdle; + + @Override + public int getTimeToSleep(boolean wasActive) { + int currentSleepTime = myCurrentSleepTime; // volatile read + if (wasActive) currentSleepTime = sleepTimeWhenWasActive; + else if (currentSleepTime == sleepTimeWhenWasActive) { + currentSleepTime = sleepTimeWhenIdle; + myIterationsWithCurrentTime = 0; + } + else { + int iterationsWithCurrentTime = ++myIterationsWithCurrentTime; + if (iterationsWithCurrentTime >= maxIterationsWithCurrentSleepTime) { + myIterationsWithCurrentTime = 0; + currentSleepTime = Math.min(2* currentSleepTime, maxSleepTimeWhenIdle); + } + } + + myCurrentSleepTime = currentSleepTime; // volatile write + return currentSleepTime; + } + } + + protected void doRun() { + try { + while (true) { + boolean read = readAvailable(); + + if (isStopped) { + break; + } + + TimeoutUtil.sleep(mySleepingPolicy.getTimeToSleep(read)); + } + } + catch (IOException e) { + LOG.info(e); + } + catch (Exception e) { + LOG.error(e); + } + finally { + try { + close(); + } + catch (IOException e) { + LOG.error("Can't close stream", e); + } + } + } + + protected abstract boolean readAvailable() throws IOException; + protected abstract void close() throws IOException; + + public void stop() { + isStopped = true; + } + + public void waitFor() throws InterruptedException { + try { + myFinishedFuture.get(); + } + catch (ExecutionException e) { + LOG.error(e); + } + } +} diff --git a/platform/util/src/com/intellij/util/io/BaseOutputReader.java b/platform/util/src/com/intellij/util/io/BaseOutputReader.java index ba1e187b8363..d6684414110c 100644 --- a/platform/util/src/com/intellij/util/io/BaseOutputReader.java +++ b/platform/util/src/com/intellij/util/io/BaseOutputReader.java @@ -15,121 +15,29 @@ */ package com.intellij.util.io; -import com.intellij.openapi.diagnostic.Logger; -import com.intellij.util.TimeoutUtil; import org.jetbrains.annotations.NotNull; import java.io.IOException; import java.io.Reader; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; /** * @author traff */ -public abstract class BaseOutputReader { - private static final Logger LOG = Logger.getInstance("#com.intellij.util.io.BaseOutputReader"); +public abstract class BaseOutputReader extends BaseDataReader { protected final Reader myReader; - protected volatile boolean isStopped = false; private final char[] myBuffer = new char[8192]; private final StringBuilder myTextBuffer = new StringBuilder(); private boolean skipLF = false; - private Future myFinishedFuture = null; - @NotNull protected final SleepingPolicy mySleepingPolicy; - public BaseOutputReader(@NotNull Reader reader) { this(reader, null); } public BaseOutputReader(@NotNull Reader reader, SleepingPolicy sleepingPolicy) { + super(sleepingPolicy); myReader = reader; - mySleepingPolicy = sleepingPolicy != null ? sleepingPolicy: SleepingPolicy.SIMPLE; - } - - protected void start() { - if (myFinishedFuture == null) { - myFinishedFuture = executeOnPooledThread(new Runnable() { - @Override - public void run() { - doRun(); - } - }); - } - } - - protected abstract Future executeOnPooledThread(Runnable runnable); - - public interface SleepingPolicy { - int sleepTimeWhenWasActive = 1; - int sleepTimeWhenIdle = 5; - - SleepingPolicy SIMPLE = new SleepingPolicy() { - @Override - public int getTimeToSleep(boolean wasActive) { - return wasActive ? sleepTimeWhenWasActive : sleepTimeWhenIdle; - } - }; - - int getTimeToSleep(boolean wasActive); - } - - public static class AdaptiveSleepingPolicy implements SleepingPolicy { - private static final int maxSleepTimeWhenIdle = 200; - private static final int maxIterationsWithCurrentSleepTime = 50; - - private volatile int myIterationsWithCurrentTime; - private volatile int myCurrentSleepTime = sleepTimeWhenIdle; - - @Override - public int getTimeToSleep(boolean wasActive) { - int currentSleepTime = myCurrentSleepTime; // volatile read - if (wasActive) currentSleepTime = sleepTimeWhenWasActive; - else if (currentSleepTime == sleepTimeWhenWasActive) { - currentSleepTime = sleepTimeWhenIdle; - myIterationsWithCurrentTime = 0; - } - else { - int iterationsWithCurrentTime = ++myIterationsWithCurrentTime; - if (iterationsWithCurrentTime >= maxIterationsWithCurrentSleepTime) { - myIterationsWithCurrentTime = 0; - currentSleepTime = Math.min(2* currentSleepTime, maxSleepTimeWhenIdle); - } - } - - myCurrentSleepTime = currentSleepTime; // volatile write - return currentSleepTime; - } - } - - protected void doRun() { - try { - while (true) { - boolean read = readAvailable(); - - if (isStopped) { - break; - } - - TimeoutUtil.sleep(mySleepingPolicy.getTimeToSleep(read)); - } - } - catch (IOException e) { - LOG.info(e); - } - catch (Exception e) { - LOG.error(e); - } - finally { - try { - myReader.close(); - } - catch (IOException e) { - LOG.error("Can't close stream", e); - } - } } /** @@ -176,18 +84,10 @@ public abstract class BaseOutputReader { return read; } + @Override + protected void close() throws IOException { + myReader.close(); + } + protected abstract void onTextAvailable(@NotNull String text); - - public void stop() { - isStopped = true; - } - - public void waitFor() throws InterruptedException { - try { - myFinishedFuture.get(); - } - catch (ExecutionException e) { - LOG.error(e); - } - } } diff --git a/platform/util/src/com/intellij/util/io/BinaryOutputReader.java b/platform/util/src/com/intellij/util/io/BinaryOutputReader.java new file mode 100644 index 000000000000..a80fb50796cd --- /dev/null +++ b/platform/util/src/com/intellij/util/io/BinaryOutputReader.java @@ -0,0 +1,58 @@ +/* + * Copyright 2000-2013 JetBrains s.r.o. + * + * Licensed 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 com.intellij.util.io; + +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.io.InputStream; + +/** +* @author Konstantin Kolosovsky. +*/ +public abstract class BinaryOutputReader extends BaseDataReader { + + @NotNull private final InputStream myStream; + @NotNull private final byte[] myBuffer = new byte[8192]; + + public BinaryOutputReader(@NotNull InputStream stream, SleepingPolicy sleepingPolicy) { + super(sleepingPolicy); + myStream = stream; + } + + @Override + protected boolean readAvailable() throws IOException { + byte[] buffer = myBuffer; + + boolean read = false; + while (myStream.available() > 0) { + int n = myStream.read(buffer); + if (n <= 0) break; + read = true; + + onBinaryAvailable(buffer, n); + } + + return read; + } + + protected abstract void onBinaryAvailable(@NotNull byte[] data, int size); + + @Override + protected void close() throws IOException { + myStream.close(); + } +} diff --git a/plugins/svn4idea/src/org/jetbrains/idea/svn/commandLine/CommandExecutor.java b/plugins/svn4idea/src/org/jetbrains/idea/svn/commandLine/CommandExecutor.java index 556691a06c6f..a7580aedd4f8 100644 --- a/plugins/svn4idea/src/org/jetbrains/idea/svn/commandLine/CommandExecutor.java +++ b/plugins/svn4idea/src/org/jetbrains/idea/svn/commandLine/CommandExecutor.java @@ -22,13 +22,18 @@ import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.util.Key; import com.intellij.openapi.util.text.StringUtil; import com.intellij.util.EventDispatcher; +import com.intellij.util.io.BaseDataReader; +import com.intellij.util.io.BinaryOutputReader; import org.jetbrains.annotations.NonNls; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.tmatesoft.svn.core.SVNCancelException; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStreamWriter; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -117,7 +122,13 @@ public class CommandExecutor { @NotNull protected OSProcessHandler createProcessHandler() { - return new OSProcessHandler(myProcess, myCommandLine.getCommandLineString()); + return needsBinaryOutput() + ? new BinaryOSProcessHandler(myProcess, myCommandLine.getCommandLineString()) + : new OSProcessHandler(myProcess, myCommandLine.getCommandLineString()); + } + + private boolean needsBinaryOutput() { + return SvnCommandName.cat.equals(myCommand.getName()); } @NotNull @@ -142,6 +153,11 @@ public class CommandExecutor { return outputAdapter.getOutput().getStderr(); } + @Nullable + public ByteArrayOutputStream getBinaryOutput() { + return myHandler instanceof BinaryOSProcessHandler ? ((BinaryOSProcessHandler)myHandler).myBinaryOutput : null; + } + // TODO: Carefully here - do not modify command from threads other than the one started command execution @NotNull public Command getCommand() { @@ -344,4 +360,38 @@ public class CommandExecutor { } } } + + private static class BinaryOSProcessHandler extends OSProcessHandler { + + @NotNull private final ByteArrayOutputStream myBinaryOutput; + + public BinaryOSProcessHandler(@NotNull final Process process, @Nullable final String commandLine) { + super(process, commandLine); + myBinaryOutput = new ByteArrayOutputStream(); + } + + @NotNull + @Override + protected BaseDataReader createOutputDataReader(BaseDataReader.SleepingPolicy sleepingPolicy) { + return new SimpleBinaryOutputReader(myProcess.getInputStream(), sleepingPolicy); + } + + private class SimpleBinaryOutputReader extends BinaryOutputReader { + + public SimpleBinaryOutputReader(@NotNull InputStream stream, SleepingPolicy sleepingPolicy) { + super(stream, sleepingPolicy); + start(); + } + + @Override + protected void onBinaryAvailable(@NotNull byte[] data, int size) { + myBinaryOutput.write(data, 0, size); + } + + @Override + protected Future executeOnPooledThread(Runnable runnable) { + return BinaryOSProcessHandler.this.executeOnPooledThread(runnable); + } + } + } } diff --git a/plugins/svn4idea/src/org/jetbrains/idea/svn/content/CmdContentClient.java b/plugins/svn4idea/src/org/jetbrains/idea/svn/content/CmdContentClient.java index 2b1c22ee5c1a..d654253bf426 100644 --- a/plugins/svn4idea/src/org/jetbrains/idea/svn/content/CmdContentClient.java +++ b/plugins/svn4idea/src/org/jetbrains/idea/svn/content/CmdContentClient.java @@ -10,6 +10,7 @@ import org.jetbrains.idea.svn.commandLine.*; import org.tmatesoft.svn.core.wc.SVNRevision; import org.tmatesoft.svn.core.wc2.SvnTarget; +import java.io.ByteArrayOutputStream; import java.util.ArrayList; import java.util.List; @@ -22,17 +23,17 @@ public class CmdContentClient extends BaseSvnClient implements ContentClient { public byte[] getContent(@NotNull SvnTarget target, @Nullable SVNRevision revision, @Nullable SVNRevision pegRevision) throws VcsException, FileTooBigRuntimeException { // TODO: rewrite this to provide output as Stream - // TODO: rewrite without conversion from String to byte[] // TODO: Also implement max size constraint like in SvnKitContentClient - // TODO: Could not use export to get content of scheduled for deletion file - use cat command, but write special binary handler + // NOTE: Export could not be used to get content of scheduled for deletion file List parameters = new ArrayList(); CommandUtil.put(parameters, target.getPathOrUrlString(), pegRevision); CommandUtil.put(parameters, revision); CommandExecutor command = CommandUtil.execute(myVcs, target, SvnCommandName.cat, parameters, null); - - byte[] bytes = CharsetToolkit.getUtf8Bytes(command.getOutput()); + // TODO: currently binary output will be null for terminal mode - use text output in this case + ByteArrayOutputStream output = command.getBinaryOutput(); + byte[] bytes = output != null ? output.toByteArray() : CharsetToolkit.getUtf8Bytes(command.getOutput()); ContentRevisionCache.checkContentsSize(target.getPathOrUrlString(), bytes.length);