Merge pull request #40 from qiniu/develop

Release 6.0.1

Author: longbai

CHANGELOG.md

@@ -1,5 +1,12 @@
 ## CHANGE LOG
 
+### v6.0.1
+2014-04-03 issue [#40](https://github.com/qiniu/android-sdk/pull/40)
+
+- [#35] fix bugs and close idle connection
+- [#36] 增加连接超时处理
+
+
 ### v6.0.0
 
 增加 SDK 实现规范

README.md

@@ -32,9 +32,8 @@ Qiniu Resource Storage SDK for Android
 
 ## 许可证
 
-Copyright (c) 2013 qiniu.com
+Copyright (c) 2012-2014 qiniu.com
 
 基于 MIT 协议发布:
 
 * [www.opensource.org/licenses/MIT](http://www.opensource.org/licenses/MIT)
-* 

docs/README.md

@@ -22,6 +22,7 @@
 import org.apache.http.util.EntityUtils;
 
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
 public class Client {
 
@@ -31,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);
@@ -77,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;
@@ -90,22 +97,16 @@ protected Object doInBackground(Object... objects) {
 			try {
 				HttpResponse resp = roundtrip(mHttpRequest);
 				int statusCode = resp.getStatusLine().getStatusCode();
-				if (statusCode == 401) { // android 2.3 will not response
-					return new Exception("unauthorized!");
-				}
-				byte[] data = EntityUtils.toByteArray(resp.getEntity());
+				String xl = resp.getFirstHeader("X-Log").getValue();
 
-				if (statusCode / 100 != 2) {
-					if (data.length == 0) {
-						String xlog = resp.getFirstHeader("X-Log").getValue();
-						if (xlog.length() > 0) {
-							return new Exception(xlog);
-						}
-						return new Exception(resp.getStatusLine().getReasonPhrase());
-					}
-					return new Exception(new String(data));
-				}
-				return data;
+				if (statusCode == 401) return new Exception("unauthorized!"); // android 2.3 will not response
+				if (xl.contains("invalid BlockCtx")) return new Exception(xl);
+
+				byte[] data = EntityUtils.toByteArray(resp.getEntity());
+				if (statusCode / 100 == 2) return data;
+				if (data.length > 0) return new Exception(new String(data));
+				if (xl.length() > 0) return new Exception(xl);
+				return new Exception(resp.getStatusLine().getStatusCode() + ":" + resp.getStatusLine().getReasonPhrase());
 			} catch (IOException e) {
 				e.printStackTrace();
 				return e;
@@ -114,17 +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) {
+			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/auth/Client.java

@@ -12,6 +12,7 @@
 import org.json.JSONObject;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Map;
@@ -21,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;
 	}
 
@@ -44,9 +51,18 @@ private static Client defaultClient() {
 	public void put(String key, InputStreamAt isa, PutExtra extra, JSONObjectRet ret) {
 		MultipartEntity m = new MultipartEntity();
 		if (key != null) m.addField("key", key);
-		if (extra.checkCrc == PutExtra.AUTO_CRC32) extra.crc32 = isa.crc32();
+		if (extra.checkCrc == PutExtra.AUTO_CRC32) {
+			try {
+				extra.crc32 = isa.crc32();
+			} catch (IOException e) {
+				ret.onFailure(e);
+				return;
+			}
+		}
 		if (extra.checkCrc != PutExtra.UNUSE_CRC32) m.addField("crc32", extra.crc32 + "");
-		for (Map.Entry<String, String> i: extra.params.entrySet()) m.addField(i.getKey(), i.getValue());
+		for (Map.Entry<String, String> i: extra.params.entrySet()) {
+			m.addField(i.getKey(), i.getValue());
+		}
 
 		m.addField("token", mUptoken);
 		m.addFile("file", extra.mimeType, key == null ? "?" : key, isa);
@@ -58,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/io/IO.java

@@ -47,14 +47,24 @@ public void onInit(int flag) {
 
 			public void putInit() {
 				int chunkSize = Math.min(writeNeed, CHUNK_SIZE);
-				crc32 = input.getCrc32(offset, chunkSize);
+				try {
+					crc32 = input.getCrc32(offset, chunkSize);
+				} catch (IOException e) {
+					onFailure(e);
+					return;
+				}
 				canceler[0] = mkblk(input, offset, writeNeed, chunkSize, this);
 			}
 
 			public void putNext() {
 				wrote = putRet.offset;
 				int remainLength = Math.min((int) (input.length() - offset - putRet.offset), CHUNK_SIZE);
-				crc32 = input.getCrc32(offset+putRet.offset, remainLength);
+				try {
+					crc32 = input.getCrc32(offset+putRet.offset, remainLength);
+				} catch (IOException e) {
+					onFailure(e);
+					return;
+				}
 				canceler[0] = bput(putRet.host, input, putRet.ctx, offset, putRet.offset, remainLength, this);
 			}
 

src/com/qiniu/resumableio/ResumableClient.java

@@ -36,7 +36,7 @@ private synchronized void removeTask(Integer id) {
 		idCancels.remove(id);
 	}
 
-	public int putAndClose(final String key, final InputStreamAt input, final PutExtra extra, final JSONObjectRet ret) {
+	private int putAndClose(final String key, final InputStreamAt input, final PutExtra extra, final JSONObjectRet ret) {
 		return put(key, input, extra, new JSONObjectRet() {
 			@Override
 			public void onSuccess(JSONObject obj) {
@@ -121,6 +121,10 @@ public void onProcess(long current, long total) {
 
 				@Override
 				public void onFailure(Exception ex) {
+					if (failure[0]) {
+						ex.printStackTrace();
+						return;
+					}
 					if (--retryTime <= 0 || (ex.getMessage() != null && ex.getMessage().contains("Unauthorized"))) {
 						removeTask(taskId);
 						failure[0] = true;

src/com/qiniu/resumableio/ResumableIO.java

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

src/com/qiniu/utils/IOnProcess.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 {
@@ -58,14 +57,14 @@ public InputStreamAt(byte[] data) {
 		mData = data;
 	}
 
-	public long getCrc32(long offset, int length) {
+	public long getCrc32(long offset, int length) throws IOException {
 		CRC32 crc32 = new CRC32();
 		byte[] data = read(offset, length);
 		crc32.update(data);
 		return crc32.getValue();
 	}
 
-	public long crc32() {
+	public long crc32() throws IOException {
 		if (mCrc32 >= 0) return mCrc32;
 		CRC32 crc32 = new CRC32();
 		long index = 0;
@@ -120,45 +119,50 @@ protected static File storeToFile(Context context, InputStream is) {
 		}
 	}
 
-	public byte[] read(long offset, int length) {
-		if (mClosed) return null;
-		try {
-			if (mFileStream != null) {
-				return fileStreamRead(offset, length);
-			}
-			if (mData != null) {
-				byte[] ret = new byte[length];
-				System.arraycopy(mData, (int) offset, ret, 0, length);
-				return ret;
-			}
-		} catch (IOException e) {
-			e.printStackTrace();
+	public byte[] read(long offset, int length) throws IOException {
+		if (mClosed) throw new IOException("inputStreamAt closed");
+		if (mFileStream != null) {
+			return fileStreamRead(offset, length);
 		}
-
-		return null;
+		if (mData != null) {
+			byte[] ret = new byte[length];
+			System.arraycopy(mData, (int) offset, ret, 0, length);
+			return ret;
+		}
+		throw new IOException("inputStreamAt not init");
 	}
 
 	protected byte[] fileStreamRead(long offset, int length) throws IOException {
 		if (mFileStream == null) return null;
+		long fileLength = mFileStream.length();
+		if (length + offset > fileLength) length = (int) (fileLength - offset);
 		byte[] data = new byte[length];
 
 		int read;
 		int totalRead = 0;
 		synchronized (data) {
 			mFileStream.seek(offset);
 			do {
-				read = mFileStream.read(data, totalRead, length);
+				read = mFileStream.read(data, totalRead, length - totalRead);
 				if (read <= 0) break;
 				totalRead += read;
 			} while (length > totalRead);
 		}
 
 		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/InputStreamAt.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();