Sulkta-OSS/straw
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

main commit

Browse source 
Post-processing infrastructure
* remove interfaces with one implementation
* fix download resources with unknow length
* marquee style for ProgressDrawable
* "view details" option in mission context menu
* notification for finished downloads
* postprocessing infrastructure: sub-missions, circular file, layers for layers of abstractions for Java IO streams
* Mp4 muxing (only DASH brand)
* WebM muxing
* Captions downloading
* alert dialog for overwrite existing downloads finished or not.

Misc changes
* delete SQLiteDownloadDataSource.java
* delete DownloadMissionSQLiteHelper.java
* implement Localization from #114

Misc fixes (this branch)
* restore old mission listeners variables. Prevents registered listeners get de-referenced on low-end devices
* DownloadManagerService.checkForRunningMission() now return false if the mission has a error.
* use Intent.FLAG_ACTIVITY_NEW_TASK when launching an activity from gigaget threads (apparently it is required in old versions of android)

More changes
* proper error handling "infrastructure"
* queue instead of multiple downloads
* move serialized pending downloads (.giga files) to app data
* stop downloads when swicthing to mobile network (never works, see 2nd point)
* save the thread count for next downloads
* a lot of incoherences fixed
* delete DownloadManagerTest.java (too many changes to keep this file updated)
This commit is contained in:
kapodamy 2018-09-23 15:12:23 -03:00
parent 45fea983b9
commit 5825843f68
48 changed files with 4379 additions and 1119 deletions
Download patch file Download diff file Expand all files Collapse all files

638
app/src/main/java/us/shandian/giga/get/DownloadMission.java
View file

@ -1,102 +1,165 @@
package us.shandian.giga.get;
import android.os.Handler;
import android.os.Looper;
import android.os.Message;
import android.util.Log;
import java.io.File;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.lang.ref.WeakReference;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.ConnectException;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import javax.net.ssl.SSLException;
import us.shandian.giga.postprocessing.Postprocessing;
import us.shandian.giga.service.DownloadManagerService;
import us.shandian.giga.util.Utility;
import static org.schabi.newpipe.BuildConfig.DEBUG;
public class DownloadMission implements Serializable {
private static final long serialVersionUID = 0L;
public class DownloadMission extends Mission {
private static final long serialVersionUID = 3L;// last bump: 16 october 2018
private static final String TAG = DownloadMission.class.getSimpleName();
static final int BUFFER_SIZE = 64 * 1024;
final static int BLOCK_SIZE = 512 * 1024;
public interface MissionListener {
HashMap<MissionListener, Handler> handlerStore = new HashMap<>();
private static final String TAG = "DownloadMission";
void onProgressUpdate(DownloadMission downloadMission, long done, long total);
void onFinish(DownloadMission downloadMission);
void onError(DownloadMission downloadMission, int errCode);
}
public static final int ERROR_SERVER_UNSUPPORTED = 206;
public static final int ERROR_UNKNOWN = 233;
public static final int ERROR_NOTHING = -1;
public static final int ERROR_PATH_CREATION = 1000;
public static final int ERROR_FILE_CREATION = 1001;
public static final int ERROR_UNKNOWN_EXCEPTION = 1002;
public static final int ERROR_PERMISSION_DENIED = 1003;
public static final int ERROR_SSL_EXCEPTION = 1004;
public static final int ERROR_UNKNOWN_HOST = 1005;
public static final int ERROR_CONNECT_HOST = 1006;
public static final int ERROR_POSTPROCESSING_FAILED = 1007;
public static final int ERROR_HTTP_NO_CONTENT = 204;
public static final int ERROR_HTTP_UNSUPPORTED_RANGE = 206;
/**
* The filename
* The urls of the file to download
*/
public String name;
public String[] urls;
/**
* The url of the file to download
* Number of blocks the size of {@link DownloadMission#BLOCK_SIZE}
*/
public String url;
/**
* The directory to store the download
*/
public String location;
/**
* Number of blocks the size of {@link DownloadManager#BLOCK_SIZE}
*/
public long blocks;
/**
* Number of bytes
*/
public long length;
long blocks = -1;
/**
* Number of bytes downloaded
*/
public long done;
/**
* Indicates a file generated dynamically on the web server
*/
public boolean unknownLength;
/**
* offset in the file where the data should be written
*/
public long[] offsets;
/**
* The post-processing algorithm arguments
*/
public String[] postprocessingArgs;
/**
* The post-processing algorithm name
*/
public String postprocessingName;
/**
* Indicates if the post-processing algorithm is actually running, used to detect corrupt downloads
*/
public boolean postprocessingRunning;
/**
* Indicate if the post-processing algorithm works on the same file
*/
public boolean postprocessingThis;
/**
* The current resource to download {@code urls[current]}
*/
public int current;
/**
* Metadata where the mission state is saved
*/
public File metadata;
/**
* maximum attempts
*/
public int maxRetry;
public int threadCount = 3;
public int finishCount;
private final List<Long> threadPositions = new ArrayList<>();
public final Map<Long, Boolean> blockState = new HashMap<>();
public boolean running;
public boolean finished;
public boolean fallback;
public int errCode = -1;
public long timestamp;
boolean fallback;
private int finishCount;
public transient boolean running;
public transient boolean enqueued = true;
public int errCode = ERROR_NOTHING;
public transient Exception errObject = null;
public transient boolean recovered;
private transient ArrayList<WeakReference<MissionListener>> mListeners = new ArrayList<>();
public transient Handler mHandler;
private transient boolean mWritingToFile;
private static final int NO_IDENTIFIER = -1;
@SuppressWarnings("UseSparseArrays")// LongSparseArray is not serializable
private final HashMap<Long, Boolean> blockState = new HashMap<>();
final List<Long> threadBlockPositions = new ArrayList<>();
final List<Integer> threadBytePositions = new ArrayList<>();
private transient boolean deleted;
int currentThreadCount;
private transient Thread[] threads = null;
private transient Thread init = null;
protected DownloadMission() {
public DownloadMission() {
}
public DownloadMission(String name, String url, String location) {
public DownloadMission(String url, String name, String location, char kind) {
this(new String[]{url}, name, location, kind, null, null);
}
public DownloadMission(String[] urls, String name, String location, char kind, String postprocessingName, String[] postprocessingArgs) {
if (name == null) throw new NullPointerException("name is null");
if (name.isEmpty()) throw new IllegalArgumentException("name is empty");
if (url == null) throw new NullPointerException("url is null");
if (url.isEmpty()) throw new IllegalArgumentException("url is empty");
if (urls == null) throw new NullPointerException("urls is null");
if (urls.length < 1) throw new IllegalArgumentException("urls is empty");
if (location == null) throw new NullPointerException("location is null");
if (location.isEmpty()) throw new IllegalArgumentException("location is empty");
this.url = url;
this.urls = urls;
this.name = name;
this.location = location;
}
this.kind = kind;
this.offsets = new long[urls.length];
if (postprocessingName != null) {
Postprocessing algorithm = Postprocessing.getAlgorithm(postprocessingName, null);
this.postprocessingThis = algorithm.worksOnSameFile;
this.offsets[0] = algorithm.recommendedReserve;
this.postprocessingName = postprocessingName;
this.postprocessingArgs = postprocessingArgs;
} else {
if (DEBUG && urls.length > 1) {
Log.w(TAG, "mission created with multiple urls ¿missing post-processing algorithm?");
}
}
}
private void checkBlock(long block) {
if (block < 0 || block >= blocks) {
@ -110,12 +173,12 @@ public class DownloadMission implements Serializable {
* @param block the block identifier
* @return true if the block is reserved and false if otherwise
*/
public boolean isBlockPreserved(long block) {
boolean isBlockPreserved(long block) {
checkBlock(block);
return blockState.containsKey(block) ? blockState.get(block) : false;
}
public void preserveBlock(long block) {
void preserveBlock(long block) {
checkBlock(block);
synchronized (blockState) {
blockState.put(block, true);
@ -123,125 +186,192 @@ public class DownloadMission implements Serializable {
}
/**
* Set the download position of the file
* Set the block of the file
*
* @param threadId the identifier of the thread
* @param position the download position of the thread
* @param position the block of the thread
*/
public void setPosition(int threadId, long position) {
threadPositions.set(threadId, position);
void setBlockPosition(int threadId, long position) {
threadBlockPositions.set(threadId, position);
}
/**
* Get the position of a thread
* Get the block of a file
*
* @param threadId the identifier of the thread
* @return the position for the thread
* @return the block for the thread
*/
public long getPosition(int threadId) {
return threadPositions.get(threadId);
long getBlockPosition(int threadId) {
return threadBlockPositions.get(threadId);
}
public synchronized void notifyProgress(long deltaLen) {
/**
* Save the position of the desired thread
*
* @param threadId the identifier of the thread
* @param position the relative position in bytes or zero
*/
void setThreadBytePosition(int threadId, int position) {
threadBytePositions.set(threadId, position);
}
/**
* Get position inside of the block, where thread will be resumed
*
* @param threadId the identifier of the thread
* @return the relative position in bytes or zero
*/
int getBlockBytePosition(int threadId) {
return threadBytePositions.get(threadId);
}
/**
* Open connection
*
* @param threadId id of the calling thread, used only for debug
* @param rangeStart range start
* @param rangeEnd range end
* @return a {@link java.net.URLConnection URLConnection} linking to the URL.
* @throws IOException if an I/O exception occurs.
* @throws HttpError if the the http response is not satisfiable
*/
HttpURLConnection openConnection(int threadId, long rangeStart, long rangeEnd) throws IOException, HttpError {
URL url = new URL(urls[current]);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setInstanceFollowRedirects(true);
if (rangeStart >= 0) {
String req = "bytes=" + rangeStart + "-";
if (rangeEnd > 0) req += rangeEnd;
conn.setRequestProperty("Range", req);
if (DEBUG) {
Log.d(TAG, threadId + ":" + conn.getRequestProperty("Range"));
Log.d(TAG, threadId + ":Content-Length=" + conn.getContentLength() + " Code:" + conn.getResponseCode());
}
}
int statusCode = conn.getResponseCode();
switch (statusCode) {
case 204:
case 205:
case 207:
throw new HttpError(conn.getResponseCode());
default:
if (statusCode < 200 || statusCode > 299) {
throw new HttpError(statusCode);
}
}
return conn;
}
private void notify(int what) {
Message m = new Message();
m.what = what;
m.obj = this;
mHandler.sendMessage(m);
}
synchronized void notifyProgress(long deltaLen) {
if (!running) return;
if (recovered) {
recovered = false;
}
if (unknownLength) {
length += deltaLen;// Update length before proceeding
}
done += deltaLen;
if (done > length) {
done = length;
}
if (done != length) {
writeThisToFile();
if (done != length && !deleted && !mWritingToFile) {
mWritingToFile = true;
runAsync(-2, this::writeThisToFile);
}
for (WeakReference<MissionListener> ref : mListeners) {
final MissionListener listener = ref.get();
if (listener != null) {
MissionListener.handlerStore.get(listener).post(new Runnable() {
@Override
public void run() {
listener.onProgressUpdate(DownloadMission.this, done, length);
}
});
}
notify(DownloadManagerService.MESSAGE_PROGRESS);
}
synchronized void notifyError(Exception err) {
Log.e(TAG, "notifyError()", err);
if (err instanceof FileNotFoundException) {
notifyError(ERROR_FILE_CREATION, null);
} else if (err instanceof SSLException) {
notifyError(ERROR_SSL_EXCEPTION, null);
} else if (err instanceof HttpError) {
notifyError(((HttpError) err).statusCode, null);
} else if (err instanceof ConnectException) {
notifyError(ERROR_CONNECT_HOST, null);
} else if (err instanceof UnknownHostException) {
notifyError(ERROR_UNKNOWN_HOST, null);
} else {
notifyError(ERROR_UNKNOWN_EXCEPTION, err);
}
}
/**
* Called by a download thread when it finished.
*/
public synchronized void notifyFinished() {
if (errCode > 0) return;
synchronized void notifyError(int code, Exception err) {
Log.e(TAG, "notifyError() code = " + code, err);
errCode = code;
errObject = err;
pause();
notify(DownloadManagerService.MESSAGE_ERROR);
}
synchronized void notifyFinished() {
if (errCode > ERROR_NOTHING) return;
finishCount++;
if (finishCount == threadCount) {
onFinish();
if (finishCount == currentThreadCount) {
if ((current + 1) < urls.length) {
// prepare next sub-mission
long current_offset = offsets[current++];
offsets[current] = current_offset + length;
initializer();
return;
}
current++;
unknownLength = false;
if (!doPostprocessing()) return;
if (errCode > ERROR_NOTHING) return;
if (DEBUG) {
Log.d(TAG, "onFinish");
}
running = false;
deleteThisFromFile();
notify(DownloadManagerService.MESSAGE_FINISHED);
}
}
/**
* Called when all parts are downloaded
*/
private void onFinish() {
if (errCode > 0) return;
private void notifyPostProcessing(boolean processing) {
if (DEBUG) {
Log.d(TAG, "onFinish");
Log.d(TAG, (processing ? "enter" : "exit") + " postprocessing on " + location + File.separator + name);
}
running = false;
finished = true;
deleteThisFromFile();
for (WeakReference<MissionListener> ref : mListeners) {
final MissionListener listener = ref.get();
if (listener != null) {
MissionListener.handlerStore.get(listener).post(new Runnable() {
@Override
public void run() {
listener.onFinish(DownloadMission.this);
}
});
synchronized (blockState) {
if (!processing) {
postprocessingName = null;
postprocessingArgs = null;
}
}
}
public synchronized void notifyError(int err) {
errCode = err;
writeThisToFile();
for (WeakReference<MissionListener> ref : mListeners) {
final MissionListener listener = ref.get();
MissionListener.handlerStore.get(listener).post(new Runnable() {
@Override
public void run() {
listener.onError(DownloadMission.this, errCode);
}
});
}
}
public synchronized void addListener(MissionListener listener) {
Handler handler = new Handler(Looper.getMainLooper());
MissionListener.handlerStore.put(listener, handler);
mListeners.add(new WeakReference<>(listener));
}
public synchronized void removeListener(MissionListener listener) {
for (Iterator<WeakReference<MissionListener>> iterator = mListeners.iterator();
iterator.hasNext(); ) {
WeakReference<MissionListener> weakRef = iterator.next();
if (listener != null && listener == weakRef.get()) {
iterator.remove();
}
// don't return without fully write the current state
postprocessingRunning = processing;
Utility.writeToFile(metadata, DownloadMission.this);
}
}
@ -249,92 +379,206 @@ public class DownloadMission implements Serializable {
* Start downloading with multiple threads.
*/
public void start() {
if (!running && !finished) {
running = true;
if (running || current >= urls.length) return;
enqueued = false;
running = true;
errCode = ERROR_NOTHING;
if (!fallback) {
for (int i = 0; i < threadCount; i++) {
if (threadPositions.size() <= i && !recovered) {
threadPositions.add((long) i);
}
new Thread(new DownloadRunnable(this, i)).start();
}
} else {
// In fallback mode, resuming is not supported.
threadCount = 1;
if (blocks < 0) {
initializer();
return;
}
init = null;
if (threads == null) {
threads = new Thread[currentThreadCount];
}
if (fallback) {
if (unknownLength) {
done = 0;
blocks = 0;
new Thread(new DownloadRunnableFallback(this)).start();
length = 0;
}
threads[0] = runAsync(1, new DownloadRunnableFallback(this));
} else {
for (int i = 0; i < currentThreadCount; i++) {
threads[i] = runAsync(i + 1, new DownloadRunnable(this, i));
}
}
}
public void pause() {
if (running) {
running = false;
recovered = true;
/**
* Pause the mission, does not affect the blocks that are being downloaded.
*/
public synchronized void pause() {
if (!running) return;
// TODO: Notify & Write state to info file
// if (err)
running = false;
recovered = true;
enqueued = false;
if (init != null && init != Thread.currentThread() && init.isAlive()) {
init.interrupt();
try {
init.join();
} catch (InterruptedException e) {
// nothing to do
}
resetState();
return;
}
if (DEBUG && blocks < 1) {
Log.w(TAG, "pausing a download that can not be resumed.");
}
if (threads == null || Thread.interrupted()) {
writeThisToFile();
return;
}
// wait for all threads are suspended before save the state
runAsync(-1, () -> {
try {
for (Thread thread : threads) {
if (thread == Thread.currentThread()) continue;
if (thread.isAlive()) {
thread.interrupt();
thread.join();
}
}
} catch (Exception e) {
// nothing to do
} finally {
writeThisToFile();
}
});
}
/**
* Removes the file and the meta file
*/
public void delete() {
deleteThisFromFile();
new File(location, name).delete();
@Override
public boolean delete() {
deleted = true;
boolean res = deleteThisFromFile();
if (!super.delete()) res = false;
return res;
}
void resetState() {
done = 0;
blocks = -1;
errCode = ERROR_NOTHING;
fallback = false;
unknownLength = false;
finishCount = 0;
threadBlockPositions.clear();
threadBytePositions.clear();
blockState.clear();
threads = null;
Utility.writeToFile(metadata, DownloadMission.this);
}
private void initializer() {
init = runAsync(DownloadInitializer.mId, new DownloadInitializer(this));
}
/**
* Write this {@link DownloadMission} to the meta file asynchronously
* if no thread is already running.
*/
public void writeThisToFile() {
if (!mWritingToFile) {
mWritingToFile = true;
new Thread() {
@Override
public void run() {
doWriteThisToFile();
mWritingToFile = false;
}
}.start();
}
}
/**
* Write this {@link DownloadMission} to the meta file.
*/
private void doWriteThisToFile() {
private void writeThisToFile() {
synchronized (blockState) {
Utility.writeToFile(getMetaFilename(), this);
if (deleted) return;
Utility.writeToFile(metadata, DownloadMission.this);
}
mWritingToFile = false;
}
public boolean isFinished() {
return current >= urls.length && postprocessingName == null;
}
private boolean doPostprocessing() {
if (postprocessingName == null) return true;
try {
notifyPostProcessing(true);
notifyProgress(0);
Thread.currentThread().setName("[" + TAG + "] post-processing = " + postprocessingName + " filename = " + name);
Postprocessing algorithm = Postprocessing.getAlgorithm(postprocessingName, this);
algorithm.run();
} catch (Exception err) {
StringBuilder args = new StringBuilder(" ");
if (postprocessingArgs != null) {
for (String arg : postprocessingArgs) {
args.append(", ");
args.append(arg);
}
args.delete(0, 1);
}
Log.e(TAG, String.format("Post-processing failed. algorithm = %s args = [%s]", postprocessingName, args), err);
notifyError(ERROR_POSTPROCESSING_FAILED, err);
return false;
} finally {
notifyPostProcessing(false);
}
if (errCode != ERROR_NOTHING) notify(DownloadManagerService.MESSAGE_ERROR);
return errCode == ERROR_NOTHING;
}
private boolean deleteThisFromFile() {
synchronized (blockState) {
return metadata.delete();
}
}
private void readObject(ObjectInputStream inputStream)
throws java.io.IOException, ClassNotFoundException
{
inputStream.defaultReadObject();
mListeners = new ArrayList<>();
}
private void deleteThisFromFile() {
new File(getMetaFilename()).delete();
}
/**
* Get the path of the meta file
* run a method in a new thread
*
* @return the path to the meta file
* @param id id of new thread (used for debugging only)
* @param who the object whose {@code run} method is invoked when this thread is started
* @return the created thread
*/
private String getMetaFilename() {
return location + "/" + name + ".giga";
private Thread runAsync(int id, Runnable who) {
// known thread ids:
// -2: state saving by notifyProgress() method
// -1: wait for saving the state by pause() method
// 0: initializer
// >=1: any download thread
Thread thread = new Thread(who);
if (DEBUG) {
thread.setName(String.format("[%s] id = %s filename = %s", TAG, id, name));
}
thread.start();
return thread;
}
public File getDownloadedFile() {
return new File(location, name);
}
static class HttpError extends Exception {
int statusCode;
HttpError(int statusCode) {
this.statusCode = statusCode;
}
@Override
public String getMessage() {
return "Http status code" + String.valueOf(statusCode);
}
}
}