+ * If data is due to be sent at some time in the future, the worker + * schedules a wakeup for that time and also listens for events indicating + * that new data may be ready to send. + *
+ * If there's no data to send, the worker listens for events indicating
+ * that new data may be ready to send.
+ */
+ private enum State {
+ CREATED,
+ CHECKING_FOR_DATA,
+ WAITING_FOR_DATA,
+ CONNECTIVITY_CHECK,
+ WRITING_UPLOADING,
+ DESTROYED
+ }
+
+ private static final Logger LOG =
+ getLogger(MailboxUploadWorker.class.getName());
+
+ /**
+ * When we're waiting for data to send and an event indicates that new data
+ * may have become available, wait this long before checking the DB. This
+ * should help to avoid creating lots of small files when several acks or
+ * messages become available to send in a short period (eg when reading a
+ * file downloaded from a mailbox).
+ */
+ private static final long CHECK_DELAY_MS = 5_000;
+
+ /**
+ * How long to wait before retrying when an exception occurs while writing
+ * a file.
+ */
+ private static final long RETRY_DELAY_MS = MINUTES.toMillis(1);
+
+ private final Executor ioExecutor;
+ private final DatabaseComponent db;
+ private final Clock clock;
+ private final TaskScheduler taskScheduler;
+ private final EventBus eventBus;
+ private final ConnectivityChecker connectivityChecker;
+ private final MailboxApiCaller mailboxApiCaller;
+ private final MailboxApi mailboxApi;
+ private final MailboxFileManager mailboxFileManager;
+ private final MailboxProperties mailboxProperties;
+ private final MailboxFolderId folderId;
+ private final ContactId contactId;
+
+ private final Object lock = new Object();
+
+ @GuardedBy("lock")
+ private State state = State.CREATED;
+
+ @GuardedBy("lock")
+ @Nullable
+ private Cancellable wakeupTask = null, checkTask = null, apiCall = null;
+
+ @GuardedBy("lock")
+ @Nullable
+ private File file = null;
+
+ MailboxUploadWorker(@IoExecutor Executor ioExecutor,
+ DatabaseComponent db,
+ Clock clock,
+ TaskScheduler taskScheduler,
+ EventBus eventBus,
+ ConnectivityChecker connectivityChecker,
+ MailboxApiCaller mailboxApiCaller,
+ MailboxApi mailboxApi,
+ MailboxFileManager mailboxFileManager,
+ MailboxProperties mailboxProperties,
+ MailboxFolderId folderId,
+ ContactId contactId) {
+ this.ioExecutor = ioExecutor;
+ this.db = db;
+ this.clock = clock;
+ this.taskScheduler = taskScheduler;
+ this.eventBus = eventBus;
+ this.connectivityChecker = connectivityChecker;
+ this.mailboxApiCaller = mailboxApiCaller;
+ this.mailboxApi = mailboxApi;
+ this.mailboxFileManager = mailboxFileManager;
+ this.mailboxProperties = mailboxProperties;
+ this.folderId = folderId;
+ this.contactId = contactId;
+ }
+
+ @Override
+ public void start() {
+ LOG.info("Started");
+ synchronized (lock) {
+ // Don't allow the worker to be reused
+ if (state != State.CREATED) throw new IllegalStateException();
+ state = State.CHECKING_FOR_DATA;
+ }
+ ioExecutor.execute(this::checkForDataToSend);
+ }
+
+ @Override
+ public void destroy() {
+ LOG.info("Destroyed");
+ Cancellable wakeupTask, checkTask, apiCall;
+ File file;
+ synchronized (lock) {
+ state = State.DESTROYED;
+ wakeupTask = this.wakeupTask;
+ this.wakeupTask = null;
+ checkTask = this.checkTask;
+ this.checkTask = null;
+ apiCall = this.apiCall;
+ this.apiCall = null;
+ file = this.file;
+ this.file = null;
+ }
+ if (wakeupTask != null) wakeupTask.cancel();
+ if (checkTask != null) checkTask.cancel();
+ if (apiCall != null) apiCall.cancel();
+ if (file != null) delete(file);
+ connectivityChecker.removeObserver(this);
+ eventBus.removeListener(this);
+ }
+
+ @IoExecutor
+ private void checkForDataToSend() {
+ synchronized (lock) {
+ checkTask = null;
+ if (state != State.CHECKING_FOR_DATA) return;
+ }
+ LOG.info("Checking for data to send");
+ try {
+ db.transaction(true, txn -> {
+ long nextSendTime;
+ if (db.containsAcksToSend(txn, contactId)) {
+ nextSendTime = 0L;
+ } else {
+ nextSendTime = db.getNextSendTime(txn, contactId,
+ MAX_LATENCY);
+ }
+ // Handle the result on the event executor to avoid races with
+ // incoming events
+ txn.attach(() -> handleNextSendTime(nextSendTime));
+ });
+ } catch (DbException e) {
+ logException(LOG, WARNING, e);
+ }
+ }
+
+ @EventExecutor
+ private void handleNextSendTime(long nextSendTime) {
+ if (nextSendTime == Long.MAX_VALUE) {
+ // Nothing is sendable now or due to be sent in the future. Wait
+ // for an event indicating that new data may be ready to send
+ waitForDataToSend();
+ } else {
+ // Work out the delay until data's ready to send (may be negative)
+ long delay = nextSendTime - clock.currentTimeMillis();
+ if (delay > 0) {
+ // Schedule a wakeup when data will be ready to send. If an
+ // event is received in the meantime indicating that new data
+ // may be ready to send, we'll cancel the wakeup
+ scheduleWakeup(delay);
+ } else {
+ // Data is ready to send now
+ checkConnectivity();
+ }
+ }
+ }
+
+ @EventExecutor
+ private void waitForDataToSend() {
+ synchronized (lock) {
+ if (state != State.CHECKING_FOR_DATA) return;
+ state = State.WAITING_FOR_DATA;
+ LOG.info("Waiting for data to send");
+ }
+ }
+
+ @EventExecutor
+ private void scheduleWakeup(long delay) {
+ synchronized (lock) {
+ if (state != State.CHECKING_FOR_DATA) return;
+ state = State.WAITING_FOR_DATA;
+ if (LOG.isLoggable(INFO)) {
+ LOG.info("Scheduling wakeup in " + delay + " ms");
+ }
+ wakeupTask = taskScheduler.schedule(this::wakeUp, ioExecutor,
+ delay, MILLISECONDS);
+ }
+ }
+
+ @IoExecutor
+ private void wakeUp() {
+ LOG.info("Woke up");
+ synchronized (lock) {
+ wakeupTask = null;
+ if (state != State.WAITING_FOR_DATA) return;
+ state = State.CHECKING_FOR_DATA;
+ }
+ checkForDataToSend();
+ }
+
+ @EventExecutor
+ private void checkConnectivity() {
+ synchronized (lock) {
+ if (state != State.CHECKING_FOR_DATA) return;
+ state = State.CONNECTIVITY_CHECK;
+ }
+ LOG.info("Checking connectivity");
+ // Avoid leaking observer in case destroy() is called concurrently
+ // before observer is added
+ connectivityChecker.checkConnectivity(mailboxProperties, this);
+ boolean destroyed;
+ synchronized (lock) {
+ destroyed = state == State.DESTROYED;
+ }
+ if (destroyed) connectivityChecker.removeObserver(this);
+ }
+
+ @Override
+ public void onConnectivityCheckSucceeded() {
+ LOG.info("Connectivity check succeeded");
+ synchronized (lock) {
+ if (state != State.CONNECTIVITY_CHECK) return;
+ state = State.WRITING_UPLOADING;
+ }
+ ioExecutor.execute(this::writeAndUploadFile);
+ }
+
+ @IoExecutor
+ private void writeAndUploadFile() {
+ synchronized (lock) {
+ if (state != State.WRITING_UPLOADING) return;
+ }
+ OutgoingSessionRecord sessionRecord = new OutgoingSessionRecord();
+ File file;
+ try {
+ file = mailboxFileManager.createAndWriteTempFileForUpload(
+ contactId, sessionRecord);
+ } catch (IOException e) {
+ logException(LOG, WARNING, e);
+ // Try again after a delay
+ synchronized (lock) {
+ if (state != State.WRITING_UPLOADING) return;
+ state = State.CHECKING_FOR_DATA;
+ checkTask = taskScheduler.schedule(this::checkForDataToSend,
+ ioExecutor, RETRY_DELAY_MS, MILLISECONDS);
+ }
+ return;
+ }
+ boolean deleteFile = false;
+ synchronized (lock) {
+ if (state == State.WRITING_UPLOADING) {
+ this.file = file;
+ apiCall = mailboxApiCaller.retryWithBackoff(
+ new SimpleApiCall(() -> apiCallUploadFile(file,
+ sessionRecord)));
+ } else {
+ deleteFile = true;
+ }
+ }
+ if (deleteFile) delete(file);
+ }
+
+ @IoExecutor
+ private void apiCallUploadFile(File file,
+ OutgoingSessionRecord sessionRecord)
+ throws IOException, ApiException {
+ synchronized (lock) {
+ if (state != State.WRITING_UPLOADING) return;
+ }
+ LOG.info("Uploading file");
+ mailboxApi.addFile(mailboxProperties, folderId, file);
+ markMessagesSentOrAcked(sessionRecord);
+ synchronized (lock) {
+ if (state != State.WRITING_UPLOADING) return;
+ state = State.CHECKING_FOR_DATA;
+ apiCall = null;
+ this.file = null;
+ }
+ delete(file);
+ checkForDataToSend();
+ }
+
+ private void markMessagesSentOrAcked(OutgoingSessionRecord sessionRecord) {
+ Collection