Merge pull request #36 from chzyer/feature/v6

fix

Author: xushiwei

src/com/qiniu/auth/Client.java

@@ -32,6 +32,11 @@ public Client(HttpClient client) {
 		mClient = client;
 	}
 
+	public void close() {
+		mClient.getConnectionManager().closeExpiredConnections();
+		mClient.getConnectionManager().shutdown();
+	}
+
 	public static ClientExecutor get(String url, CallRet ret) {
 		Client client = Client.defaultClient();
 		return client.get(client.makeClientExecutor(), url, ret);
@@ -78,6 +83,7 @@ protected HttpResponse roundtrip(HttpRequestBase httpRequest) throws IOException
 	public class ClientExecutor extends AsyncTask<Object, Object, Object> implements ICancel {
 		HttpRequestBase mHttpRequest;
 		CallRet mRet;
+		boolean failed;
 		public void setup(HttpRequestBase httpRequest, CallRet ret) {
 			mHttpRequest = httpRequest;
 			mRet = ret;
@@ -109,18 +115,29 @@ protected Object doInBackground(Object... objects) {
 
 		@Override
 		protected void onProgressUpdate(Object... values) {
+			if (failed) return;
+			if (values.length == 1 && values[0] instanceof Exception) {
+				mRet.onFailure((Exception) values[0]);
+				failed = true;
+				return;
+			}
 			mRet.onProcess((Long) values[0], (Long) values[1]);
 		}
 
 		@Override
 		protected void onPostExecute(Object o) {
-			mClient.getConnectionManager().closeIdleConnections(30, TimeUnit.SECONDS);
+			if (failed) return;
 			if (o instanceof Exception) {
 				mRet.onFailure((Exception) o);
 				return;
 			}
 			mRet.onSuccess((byte[]) o);
 		}
+
+		public void onFailure(Exception ex) {
+			publishProgress(ex);
+			cancel(true);
+		}
 	};
 
 	public static Client defaultClient() {

src/com/qiniu/io/IO.java

@@ -22,15 +22,21 @@ public class IO {
 	public static String UNDEFINED_KEY = null;
 	private static Client mClient;
 	private static String mUptoken;
+	private static long mClientUseTime;
 	public IO(Client client, String uptoken) {
 		mClient = client;
 		mUptoken = uptoken;
 	}
 
 	private static Client defaultClient() {
+		if (mClient != null && System.currentTimeMillis() - mClientUseTime > 3 * 60 * 1000) { // 1 minute
+			mClient.close();
+			mClient = null;
+		}
 		if (mClient == null) {
 			mClient = Client.defaultClient();
 		}
+		mClientUseTime = System.currentTimeMillis();
 		return mClient;
 	}
 
@@ -68,6 +74,11 @@ public void put(String key, InputStreamAt isa, PutExtra extra, JSONObjectRet ret
 			public void onProcess(long current, long total) {
 				executor.upload(current, total);
 			}
+
+			@Override
+			public void onFailure(Exception ex) {
+				executor.onFailure(ex);
+			}
 		});
 		client.call(executor, Conf.UP_HOST, m, ret);
 	}

src/com/qiniu/utils/IOnProcess.java

@@ -2,4 +2,5 @@
 
 public interface IOnProcess {
 	public void onProcess(long current, long total);
+	public void onFailure(Exception ex);
 }

src/com/qiniu/utils/InputStreamAt.java

@@ -7,7 +7,6 @@
 import org.apache.http.entity.AbstractHttpEntity;
 
 import java.io.*;
-import java.util.Arrays;
 import java.util.zip.CRC32;
 
 public class InputStreamAt implements Closeable {
@@ -151,11 +150,19 @@ protected byte[] fileStreamRead(long offset, int length) throws IOException {
 		}
 
 		if (totalRead != data.length) {
-			data = Arrays.copyOfRange(data, 0, totalRead);
+			data = copyOfRange(data, 0, totalRead);
 		}
 		return data;
 	}
 
+	public static byte[] copyOfRange(byte[] original, int from, int to) {
+		int newLength = to - from;
+		if (newLength < 0) throw new IllegalArgumentException(from + " > " + to);
+		byte[] copy = new byte[newLength];
+		System.arraycopy(original, from, copy, 0, Math.min(original.length - from, newLength));
+	    return copy;
+	}
+
 	@Override
 	public synchronized void close(){
 		if (mClosed) return;

src/com/qiniu/utils/MultipartEntity.java

@@ -8,6 +8,7 @@
 import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.Random;
+import java.util.concurrent.*;
 
 public class MultipartEntity extends AbstractHttpEntity  {
 	private String mBoundary;
@@ -33,7 +34,7 @@ public void addFile(String field, String contentType, String fileName, InputStre
 
 	@Override
 	public boolean isRepeatable() {
-		return false;
+		return true;
 	}
 
 	@Override
@@ -55,6 +56,7 @@ public InputStream getContent() throws IOException, IllegalStateException {
 
 	@Override
 	public void writeTo(OutputStream outputStream) throws IOException {
+		writed = 0;
 		outputStream.write(mData.toString().getBytes());
 		outputStream.flush();
 		writed += mData.toString().getBytes().length;
@@ -80,6 +82,7 @@ public boolean isStreaming() {
 	public void setProcessNotify(IOnProcess ret) {
 		mNotify = ret;
 	}
+	ExecutorService executor = Executors.newFixedThreadPool(1);
 
 	class FileInfo {
 
@@ -112,12 +115,18 @@ public void writeTo(OutputStream outputStream) throws IOException {
 
 			int blockSize = (int) (getContentLength() / 100);
 			if (blockSize > 256 * 1024) blockSize = 256 * 1024;
-			if (blockSize < 16 * 1024) blockSize = 16 * 1024;
+			if (blockSize < 32 * 1024) blockSize = 32 * 1024;
 			long index = 0;
 			long length = mIsa.length();
 			while (index < length) {
 				int readLength = (int) StrictMath.min((long) blockSize, mIsa.length() - index);
-				outputStream.write(mIsa.read(index, readLength));
+				int timeout = readLength * 2;
+				try {
+					write(timeout, outputStream, mIsa.read(index, readLength));
+				} catch (Exception e) {
+					mNotify.onFailure(e);
+					return;
+				}
 				index += blockSize;
 				outputStream.flush();
 				writed += readLength;
@@ -130,6 +139,18 @@ public void writeTo(OutputStream outputStream) throws IOException {
 		}
 	}
 
+	private void write(int timeout, final OutputStream outputStream, final byte[] data) throws InterruptedException, ExecutionException, TimeoutException {
+		Callable<Object> readTask = new Callable<Object>() {
+			@Override
+			public Object call() throws Exception {
+				outputStream.write(data);
+				return null;
+			}
+		};
+		Future<Object> future = executor.submit(readTask);
+		future.get(timeout, TimeUnit.MILLISECONDS);
+	}
+
 	private static String getRandomString(int length) {
 		String base = "abcdefghijklmnopqrstuvwxyz0123456789";
 		Random random = new Random();