diff --git a/AndroidManifest.xml b/AndroidManifest.xml --- a/AndroidManifest.xml +++ b/AndroidManifest.xml @@ -224,6 +224,12 @@ android:value="org.kde.kdeconnect.Plugins.SharePlugin.ShareChooserTargetService" /> + + + + + + { private final ShareNotification shareNotification; private NetworkPacket currentNetworkPacket; private String currentFileName; private int currentFileNum; private long totalReceived; private long lastProgressTimeMillis; private long prevProgressPercentage; - private final CallBack callBack; - private final Handler handler; - private final Object lock; //Use to protect concurrent access to the variables below private final List networkPacketList; private int totalNumFiles; private long totalPayloadSize; private boolean isRunning; - CompositeReceiveFileRunnable(Device device, CallBack callBack) { - this.device = device; - this.callBack = callBack; + CompositeReceiveFileJob(Device device, BackgroundJob.Callback callBack) { + super(device, callBack); lock = new Object(); networkPacketList = new ArrayList<>(); shareNotification = new ShareNotification(device); + shareNotification.setJobId(getId()); currentFileNum = 0; totalNumFiles = 0; totalPayloadSize = 0; totalReceived = 0; lastProgressTimeMillis = 0; prevProgressPercentage = 0; - handler = new Handler(Looper.getMainLooper()); + } + + private Device getDevice() { + return requestInfo; } boolean isRunning() { return isRunning; } @@ -93,8 +86,8 @@ this.totalNumFiles = numberOfFiles; this.totalPayloadSize = totalPayloadSize; - shareNotification.setTitle(device.getContext().getResources() - .getQuantityString(R.plurals.incoming_file_title, totalNumFiles, totalNumFiles, device.getName())); + shareNotification.setTitle(getDevice().getContext().getResources() + .getQuantityString(R.plurals.incoming_file_title, totalNumFiles, totalNumFiles, getDevice().getName())); } } @@ -106,8 +99,8 @@ totalNumFiles = networkPacket.getInt(SharePlugin.KEY_NUMBER_OF_FILES, 1); totalPayloadSize = networkPacket.getLong(SharePlugin.KEY_TOTAL_PAYLOAD_SIZE); - shareNotification.setTitle(device.getContext().getResources() - .getQuantityString(R.plurals.incoming_file_title, totalNumFiles, totalNumFiles, device.getName())); + shareNotification.setTitle(getDevice().getContext().getResources() + .getQuantityString(R.plurals.incoming_file_title, totalNumFiles, totalNumFiles, getDevice().getName())); } } } @@ -126,7 +119,7 @@ isRunning = true; - while (!done) { + while (!done && !canceled) { synchronized (lock) { currentNetworkPacket = networkPacketList.get(0); } @@ -138,16 +131,19 @@ fileDocument = getDocumentFileFor(currentFileName, currentNetworkPacket.getBoolean("open")); if (currentNetworkPacket.hasPayload()) { - outputStream = new BufferedOutputStream(device.getContext().getContentResolver().openOutputStream(fileDocument.getUri())); + outputStream = new BufferedOutputStream(getDevice().getContext().getContentResolver().openOutputStream(fileDocument.getUri())); InputStream inputStream = currentNetworkPacket.getPayload().getInputStream(); long received = receiveFile(inputStream, outputStream); currentNetworkPacket.getPayload().close(); if ( received != currentNetworkPacket.getPayloadSize()) { fileDocument.delete(); - throw new RuntimeException("Failed to receive: " + currentFileName + " received:" + received + " bytes, expected: " + currentNetworkPacket.getPayloadSize() + " bytes"); + + if (!canceled) { + throw new RuntimeException("Failed to receive: " + currentFileName + " received:" + received + " bytes, expected: " + currentNetworkPacket.getPayloadSize() + " bytes"); + } } else { publishFile(fileDocument, received); } @@ -163,7 +159,7 @@ listIsEmpty = networkPacketList.isEmpty(); } - if (listIsEmpty) { + if (listIsEmpty && !canceled) { try { Thread.sleep(1000); } catch (InterruptedException ignored) {} @@ -182,6 +178,12 @@ isRunning = false; + if (canceled) { + Log.e("ERIK", "I've been cancelled"); + shareNotification.cancel(); + return; + } + int numFiles; synchronized (lock) { numFiles = totalNumFiles; @@ -192,26 +194,28 @@ openFile(fileDocument); } else { //Update the notification and allow to open the file from it - shareNotification.setFinished(device.getContext().getResources().getQuantityString(R.plurals.received_files_title, numFiles, device.getName(), numFiles)); + shareNotification.setFinished(getDevice().getContext().getResources().getQuantityString(R.plurals.received_files_title, numFiles, getDevice().getName(), numFiles)); if (totalNumFiles == 1 && fileDocument != null) { shareNotification.setURI(fileDocument.getUri(), fileDocument.getType(), fileDocument.getName()); } shareNotification.show(); } - handler.post(() -> callBack.onSuccess(this)); + reportResult(null); } catch (Exception e) { isRunning = false; int failedFiles; synchronized (lock) { failedFiles = (totalNumFiles - currentFileNum + 1); } - shareNotification.setFinished(device.getContext().getResources().getQuantityString(R.plurals.received_files_fail_title, failedFiles, device.getName(), failedFiles, totalNumFiles)); + + shareNotification.setFinished(getDevice().getContext().getResources().getQuantityString(R.plurals.received_files_fail_title, failedFiles, getDevice().getName(), failedFiles, totalNumFiles)); shareNotification.show(); - handler.post(() -> callBack.onError(this, e)); + reportError(e); } finally { + Log.e("ERIK", "Closing all input and output streams"); closeAllInputStreams(); networkPacketList.clear(); if (outputStream != null) { @@ -230,12 +234,12 @@ //We need to check for already existing files only when storing in the default path. //User-defined paths use the new Storage Access Framework that already handles this. //If the file should be opened immediately store it in the standard location to avoid the FileProvider trouble (See ShareNotification::setURI) - if (open || !ShareSettingsFragment.isCustomDestinationEnabled(device.getContext())) { + if (open || !ShareSettingsFragment.isCustomDestinationEnabled(getDevice().getContext())) { final String defaultPath = ShareSettingsFragment.getDefaultDestinationDirectory().getAbsolutePath(); filenameToUse = FilesHelper.findNonExistingNameForNewFile(defaultPath, filenameToUse); destinationFolderDocument = DocumentFile.fromFile(new File(defaultPath)); } else { - destinationFolderDocument = ShareSettingsFragment.getDestinationDirectory(device.getContext()); + destinationFolderDocument = ShareSettingsFragment.getDestinationDirectory(getDevice().getContext()); } String displayName = FilesHelper.getFileNameWithoutExt(filenameToUse); String mimeType = FilesHelper.getMimeTypeFromFile(filenameToUse); @@ -247,7 +251,7 @@ DocumentFile fileDocument = destinationFolderDocument.createFile(mimeType, displayName); if (fileDocument == null) { - throw new RuntimeException(device.getContext().getString(R.string.cannot_create_file, filenameToUse)); + throw new RuntimeException(getDevice().getContext().getString(R.string.cannot_create_file, filenameToUse)); } return fileDocument; @@ -258,7 +262,7 @@ int count; long received = 0; - while ((count = input.read(data)) >= 0) { + while ((count = input.read(data)) >= 0 && !canceled) { received += count; totalReceived += count; @@ -285,27 +289,28 @@ private void closeAllInputStreams() { for (NetworkPacket np : networkPacketList) { + Log.e("ERIK", "closing a networkPackets payload"); np.getPayload().close(); } } private void setProgress(int progress) { synchronized (lock) { - shareNotification.setProgress(progress, device.getContext().getResources() + shareNotification.setProgress(progress, getDevice().getContext().getResources() .getQuantityString(R.plurals.incoming_files_text, totalNumFiles, currentFileName, currentFileNum, totalNumFiles)); } shareNotification.show(); } private void publishFile(DocumentFile fileDocument, long size) { - if (!ShareSettingsFragment.isCustomDestinationEnabled(device.getContext())) { + if (!ShareSettingsFragment.isCustomDestinationEnabled(getDevice().getContext())) { Log.i("SharePlugin", "Adding to downloads"); - DownloadManager manager = (DownloadManager) device.getContext().getSystemService(Context.DOWNLOAD_SERVICE); - manager.addCompletedDownload(fileDocument.getUri().getLastPathSegment(), device.getName(), true, fileDocument.getType(), fileDocument.getUri().getPath(), size, false); + DownloadManager manager = (DownloadManager) getDevice().getContext().getSystemService(Context.DOWNLOAD_SERVICE); + manager.addCompletedDownload(fileDocument.getUri().getLastPathSegment(), getDevice().getName(), true, fileDocument.getType(), fileDocument.getUri().getPath(), size, false); } else { //Make sure it is added to the Android Gallery anyway Log.i("SharePlugin", "Adding to gallery"); - MediaStoreHelper.indexFile(device.getContext(), fileDocument.getUri()); + MediaStoreHelper.indexFile(getDevice().getContext(), fileDocument.getUri()); } } @@ -315,13 +320,13 @@ if (Build.VERSION.SDK_INT >= 24) { //Nougat and later require "content://" uris instead of "file://" uris File file = new File(fileDocument.getUri().getPath()); - Uri contentUri = FileProvider.getUriForFile(device.getContext(), "org.kde.kdeconnect_tp.fileprovider", file); + Uri contentUri = FileProvider.getUriForFile(getDevice().getContext(), "org.kde.kdeconnect_tp.fileprovider", file); intent.setDataAndType(contentUri, mimeType); intent.addFlags(Intent.FLAG_GRANT_READ_URI_PERMISSION); } else { intent.setDataAndType(fileDocument.getUri(), mimeType); } - device.getContext().startActivity(intent); + getDevice().getContext().startActivity(intent); } } diff --git a/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareActivity.java b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareActivity.java --- a/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareActivity.java +++ b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareActivity.java @@ -147,7 +147,6 @@ final String deviceId = intent.getStringExtra("deviceId"); if (deviceId != null) { - BackgroundService.runWithPlugin(this, deviceId, SharePlugin.class, plugin -> { plugin.share(intent); finish(); diff --git a/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareBroadcastReceiver.java b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareBroadcastReceiver.java new file mode 100644 --- /dev/null +++ b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareBroadcastReceiver.java @@ -0,0 +1,57 @@ +/* + * Copyright 2018 Erik Duisters + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation; either version 2 of + * the License or (at your option) version 3 or any later version + * accepted by the membership of KDE e.V. (or its successor approved + * by the membership of KDE e.V.), which shall act as a proxy + * defined in Section 14 of version 3 of the license. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.kde.kdeconnect.Plugins.SharePlugin; + +import android.content.BroadcastReceiver; +import android.content.Context; +import android.content.Intent; +import android.util.Log; + +import org.kde.kdeconnect.BackgroundService; + +public class ShareBroadcastReceiver extends BroadcastReceiver { + @Override + public void onReceive(Context context, Intent intent) { + switch (intent.getAction()) { + case SharePlugin.ACTION_CANCEL_SHARE: + cancelShare(context, intent); + break; + default: + Log.d("ShareBroadcastReceiver", "Unhandled Action received: " + intent.getAction()); + } + } + + private void cancelShare(Context context, Intent intent) { + if (!intent.hasExtra(SharePlugin.CANCEL_SHARE_BACKGROUND_JOB_ID_EXTRA) || + !intent.hasExtra(SharePlugin.CANCEL_SHARE_DEVICE_ID_EXTRA)) { + Log.e("ShareBroadcastReceiver", "cancelShare() - not all expected extra's are present. Ignoring this cancel intent"); + return; + } + + long jobId = intent.getLongExtra(SharePlugin.CANCEL_SHARE_BACKGROUND_JOB_ID_EXTRA, -1); + String deviceId = intent.getStringExtra(SharePlugin.CANCEL_SHARE_DEVICE_ID_EXTRA); + + BackgroundService.RunCommand(context, service -> { + SharePlugin plugin = service.getDevice(deviceId).getPlugin(SharePlugin.class); + plugin.cancelJob(jobId); + }); + } +} diff --git a/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareNotification.java b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareNotification.java --- a/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareNotification.java +++ b/src/org/kde/kdeconnect/Plugins/SharePlugin/ShareNotification.java @@ -48,6 +48,7 @@ private final int notificationId; private NotificationCompat.Builder builder; private final Device device; + private long currentJobId; //https://documentation.onesignal.com/docs/android-customizations#section-big-picture private static final int bigImageWidth = 1440; @@ -73,7 +74,23 @@ notificationManager.cancel(notificationId); } - public int getId() { + public void setJobId(long jobId) { + builder.mActions.clear(); + + currentJobId = jobId; + Intent cancelIntent = new Intent(device.getContext(), ShareBroadcastReceiver.class); + cancelIntent.addFlags(Intent.FLAG_RECEIVER_FOREGROUND); + cancelIntent.setAction(SharePlugin.ACTION_CANCEL_SHARE); + cancelIntent.putExtra(SharePlugin.CANCEL_SHARE_BACKGROUND_JOB_ID_EXTRA, jobId); + cancelIntent.putExtra(SharePlugin.CANCEL_SHARE_DEVICE_ID_EXTRA, device.getDeviceId()); + PendingIntent cancelPendingIntent = PendingIntent.getBroadcast(device.getContext(), 0, cancelIntent, PendingIntent.FLAG_UPDATE_CURRENT); + + builder.addAction(R.drawable.ic_reject_pairing, device.getContext().getString(R.string.cancel), cancelPendingIntent); + } + + public long getCurrentJobId() { return currentJobId; } + + public int getNotificationId() { return notificationId; } diff --git a/src/org/kde/kdeconnect/Plugins/SharePlugin/SharePlugin.java b/src/org/kde/kdeconnect/Plugins/SharePlugin/SharePlugin.java --- a/src/org/kde/kdeconnect/Plugins/SharePlugin/SharePlugin.java +++ b/src/org/kde/kdeconnect/Plugins/SharePlugin/SharePlugin.java @@ -43,35 +43,41 @@ import org.kde.kdeconnect.Plugins.Plugin; import org.kde.kdeconnect.Plugins.PluginFactory; import org.kde.kdeconnect.UserInterface.PluginSettingsFragment; +import org.kde.kdeconnect.async.BackgroundJob; +import org.kde.kdeconnect.async.BackgroundJobHandler; import org.kde.kdeconnect_tp.R; import java.net.URL; import java.util.ArrayList; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import androidx.annotation.NonNull; import androidx.annotation.WorkerThread; import androidx.core.app.NotificationCompat; import androidx.core.content.ContextCompat; @PluginFactory.LoadablePlugin public class SharePlugin extends Plugin { + final static String ACTION_CANCEL_SHARE = "org.kde.kdeconnect.Plugins.SharePlugin.CancelShare"; + final static String CANCEL_SHARE_DEVICE_ID_EXTRA = "deviceId"; + final static String CANCEL_SHARE_BACKGROUND_JOB_ID_EXTRA = "backgroundJobId"; + private final static String PACKET_TYPE_SHARE_REQUEST = "kdeconnect.share.request"; private final static String PACKET_TYPE_SHARE_REQUEST_UPDATE = "kdeconnect.share.request.update"; final static String KEY_NUMBER_OF_FILES = "numberOfFiles"; final static String KEY_TOTAL_PAYLOAD_SIZE = "totalPayloadSize"; private final static boolean openUrlsDirectly = true; - private ExecutorService executorService; + private BackgroundJobHandler backgroundJobHandler; private final Handler handler; - private CompositeReceiveFileRunnable receiveFileRunnable; - private final Callback receiveFileRunnableCallback; + + private CompositeReceiveFileJob receiveFileJob; + private final Callback receiveFileJobCallback; public SharePlugin() { - executorService = Executors.newFixedThreadPool(5); + backgroundJobHandler = BackgroundJobHandler.newFixedThreadPoolBackgroundJobHander(5); handler = new Handler(Looper.getMainLooper()); - receiveFileRunnableCallback = new Callback(); + receiveFileJobCallback = new Callback(); } @Override @@ -122,8 +128,8 @@ public boolean onPacketReceived(NetworkPacket np) { try { if (np.getType().equals(PACKET_TYPE_SHARE_REQUEST_UPDATE)) { - if (receiveFileRunnable != null && receiveFileRunnable.isRunning()) { - receiveFileRunnable.updateTotals(np.getInt(KEY_NUMBER_OF_FILES), np.getLong(KEY_TOTAL_PAYLOAD_SIZE)); + if (receiveFileJob != null && receiveFileJob.isRunning()) { + receiveFileJob.updateTotals(np.getInt(KEY_NUMBER_OF_FILES), np.getLong(KEY_TOTAL_PAYLOAD_SIZE)); } else { Log.d("SharePlugin", "Received update packet but CompositeUploadJob is null or not running"); } @@ -200,29 +206,29 @@ @WorkerThread private void receiveFile(NetworkPacket np) { - CompositeReceiveFileRunnable runnable; + CompositeReceiveFileJob job; boolean hasNumberOfFiles = np.has(KEY_NUMBER_OF_FILES); boolean hasOpen = np.has("open"); - if (hasNumberOfFiles && !hasOpen && receiveFileRunnable != null) { - runnable = receiveFileRunnable; + if (hasNumberOfFiles && !hasOpen && receiveFileJob != null) { + job = receiveFileJob; } else { - runnable = new CompositeReceiveFileRunnable(device, receiveFileRunnableCallback); + job = new CompositeReceiveFileJob(device, receiveFileJobCallback); } if (!hasNumberOfFiles) { np.set(KEY_NUMBER_OF_FILES, 1); np.set(KEY_TOTAL_PAYLOAD_SIZE, np.getPayloadSize()); } - runnable.addNetworkPacket(np); + job.addNetworkPacket(np); - if (runnable != receiveFileRunnable) { + if (job != receiveFileJob) { if (hasNumberOfFiles && !hasOpen) { - receiveFileRunnable = runnable; + receiveFileJob = job; } - executorService.execute(runnable); + backgroundJobHandler.runJob(job); } } @@ -232,7 +238,6 @@ } void queuedSendUriList(final ArrayList uriList) { - //Read all the data early, as we only have permissions to do it while the activity is alive final ArrayList toSend = new ArrayList<>(); for (Uri uri : uriList) { @@ -281,7 +286,6 @@ } queuedSendUriList(uriList); - } catch (Exception e) { Log.e("ShareActivity", "Exception"); e.printStackTrace(); @@ -315,7 +319,6 @@ device.sendPacket(np); } } - } @Override @@ -333,19 +336,32 @@ return new String[]{Manifest.permission.WRITE_EXTERNAL_STORAGE}; } - private class Callback implements CompositeReceiveFileRunnable.CallBack { + private class Callback implements CompositeReceiveFileJob.Callback { @Override - public void onSuccess(CompositeReceiveFileRunnable runnable) { - if (runnable == receiveFileRunnable) { - receiveFileRunnable = null; + public void onResult(@NonNull BackgroundJob job, Void result) { + if (job == receiveFileJob) { + receiveFileJob = null; } } @Override - public void onError(CompositeReceiveFileRunnable runnable, Throwable error) { - Log.e("SharePlugin", "onError() - " + error.getMessage()); - if (runnable == receiveFileRunnable) { - receiveFileRunnable = null; + public void onError(@NonNull BackgroundJob job, @NonNull Throwable error) { + if (job == receiveFileJob) { + receiveFileJob = null; + } + } + } + + void cancelJob(long jobId) { + if (backgroundJobHandler.isRunning(jobId)) { + BackgroundJob job = backgroundJobHandler.getJob(jobId); + + if (job != null) { + job.cancel(); + + if (job == receiveFileJob) { + receiveFileJob = null; + } } } } diff --git a/src/org/kde/kdeconnect/async/BackgroundJob.java b/src/org/kde/kdeconnect/async/BackgroundJob.java new file mode 100644 --- /dev/null +++ b/src/org/kde/kdeconnect/async/BackgroundJob.java @@ -0,0 +1,76 @@ +/* + * Copyright 2018 Erik Duisters + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation; either version 2 of + * the License or (at your option) version 3 or any later version + * accepted by the membership of KDE e.V. (or its successor approved + * by the membership of KDE e.V.), which shall act as a proxy + * defined in Section 14 of version 3 of the license. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.kde.kdeconnect.async; + +import java.util.concurrent.atomic.AtomicLong; + +import androidx.annotation.NonNull; + +public abstract class BackgroundJob implements Runnable { + private static AtomicLong atomicLong = new AtomicLong(0); + protected volatile boolean canceled; + private BackgroundJobHandler backgroundJobHandler; + private long id; + + protected I requestInfo; + private Callback callback; + + public BackgroundJob(I requestInfo, Callback callback) { + this.id = atomicLong.incrementAndGet(); + this.requestInfo = requestInfo; + this.callback = callback; + } + + void setBackgroundJobHandler(BackgroundJobHandler handler) { + this.backgroundJobHandler = handler; + } + + public long getId() { return id; } + public I getRequestInfo() { return requestInfo; } + + public void cancel() { + canceled = true; + backgroundJobHandler.cancelJob(this); + } + + public boolean isCancelled() { + return canceled; + } + + public interface Callback { + void onResult(@NonNull BackgroundJob job, R result); + void onError(@NonNull BackgroundJob job, @NonNull Throwable error); + } + + protected void reportResult(R result) { + backgroundJobHandler.runOnUiThread(() -> { + callback.onResult(this, result); + backgroundJobHandler.onFinished(this); + }); + } + + protected void reportError(@NonNull Throwable error) { + backgroundJobHandler.runOnUiThread(() -> { + callback.onError(this, error); + backgroundJobHandler.onFinished(this); + }); + } +} diff --git a/src/org/kde/kdeconnect/async/BackgroundJobHandler.java b/src/org/kde/kdeconnect/async/BackgroundJobHandler.java new file mode 100644 --- /dev/null +++ b/src/org/kde/kdeconnect/async/BackgroundJobHandler.java @@ -0,0 +1,170 @@ +/* + * Copyright 2018 Erik Duisters + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation; either version 2 of + * the License or (at your option) version 3 or any later version + * accepted by the membership of KDE e.V. (or its successor approved + * by the membership of KDE e.V.), which shall act as a proxy + * defined in Section 14 of version 3 of the license. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.kde.kdeconnect.async; + +import android.os.Handler; +import android.os.Looper; +import android.util.Log; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import androidx.annotation.Nullable; + +public class BackgroundJobHandler { + private static final String TAG = BackgroundJobHandler.class.getSimpleName(); + + private final Map> jobMap = new HashMap<>(); + private final Object jobMapLock = new Object(); + + private class MyThreadPoolExecutor extends ThreadPoolExecutor { + MyThreadPoolExecutor(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { + super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue); + } + + @Override + protected void afterExecute(Runnable r, Throwable t) { + super.afterExecute(r, t); + + if (!(r instanceof Future)) { + return; + } + + Future future = (Future) r; + + if (t == null) { + try { + future.get(); + } catch (CancellationException ce) { + Log.d(TAG,"afterExecute got a CancellationException"); + } catch (ExecutionException ee) { + t = ee; + } catch (InterruptedException ie) { + Log.d(TAG, "afterExecute got an InterruptedException"); + Thread.currentThread().interrupt(); // ignore/reset + } + } + + if (t != null) { + BackgroundJobHandler.this.handleUncaughtException(future, t); + } + } + } + + private final ThreadPoolExecutor threadPoolExecutor; + private Handler handler; + + private BackgroundJobHandler(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { + this.handler = new Handler(Looper.getMainLooper()); + this.threadPoolExecutor = new MyThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue); + } + + public void runJob(BackgroundJob bgJob) { + Future f; + + bgJob.setBackgroundJobHandler(this); + + try { + synchronized (jobMapLock) { + f = threadPoolExecutor.submit(bgJob); + jobMap.put(bgJob, f); + } + } catch (RejectedExecutionException e) { + Log.d(TAG,"threadPoolExecutor.submit rejected a background job: " + e.getMessage()); + + bgJob.reportError(e); + } + } + + public boolean isRunning(long jobId) { + synchronized (jobMapLock) { + for (BackgroundJob job : jobMap.keySet()) { + if (job.getId() == jobId) { + return true; + } + } + } + + return false; + } + + @Nullable + public BackgroundJob getJob(long jobId) { + synchronized (jobMapLock) { + for (BackgroundJob job : jobMap.keySet()) { + if (job.getId() == jobId) { + return job; + } + } + } + + return null; + } + + void cancelJob(BackgroundJob job) { + synchronized (jobMapLock) { + if (jobMap.containsKey(job)) { + Future f = jobMap.get(job); + + if (f.cancel(true)) { + threadPoolExecutor.purge(); + } + + jobMap.remove(job); + } + } + } + + private void handleUncaughtException(Future ft, Throwable t) { + synchronized (jobMapLock) { + for (Map.Entry> pairs : jobMap.entrySet()) { + Future future = pairs.getValue(); + + if (future == ft) { + pairs.getKey().reportError(t); + break; + } + } + } + } + + void onFinished(BackgroundJob job) { + synchronized (jobMapLock) { + jobMap.remove(job); + } + } + + void runOnUiThread(Runnable runnable) { + handler.post(runnable); + } + + public static BackgroundJobHandler newFixedThreadPoolBackgroundJobHander(int numThreads) { + return new BackgroundJobHandler(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); + } +}