run roster sync in replacing task manager

src/main/java/eu/siacs/conversations/services/XmppConnectionService.java

@@ -105,6 +105,7 @@ import eu.siacs.conversations.utils.OnPhoneContactsLoadedListener;
 import eu.siacs.conversations.utils.PRNGFixes;
 import eu.siacs.conversations.utils.PhoneHelper;
 import eu.siacs.conversations.utils.ReplacingSerialSingleThreadExecutor;
+import eu.siacs.conversations.utils.ReplacingTaskManager;
 import eu.siacs.conversations.utils.Resolver;
 import eu.siacs.conversations.utils.SerialSingleThreadExecutor;
 import eu.siacs.conversations.xml.Namespace;
@@ -157,6 +158,7 @@ public class XmppConnectionService extends Service {
 	private final SerialSingleThreadExecutor mDatabaseWriterExecutor = new SerialSingleThreadExecutor("DatabaseWriter");
 	private final SerialSingleThreadExecutor mDatabaseReaderExecutor = new SerialSingleThreadExecutor("DatabaseReader");
 	private final SerialSingleThreadExecutor mNotificationExecutor = new SerialSingleThreadExecutor("NotificationExecutor");
+	private final ReplacingTaskManager mRosterSyncTaskManager = new ReplacingTaskManager();
 	private final IBinder mBinder = new XmppConnectionBinder();
 	private final List<Conversation> conversations = new CopyOnWriteArrayList<>();
 	private final IqGenerator mIqGenerator = new IqGenerator(this);
@@ -1459,7 +1461,7 @@ public class XmppConnectionService extends Service {
 
 
 	public void syncRoster(final Account account) {
-		mDatabaseWriterExecutor.execute(() -> databaseBackend.writeRoster(account.getRoster()));
+		mRosterSyncTaskManager.execute(account, () -> databaseBackend.writeRoster(account.getRoster()));
 	}
 
 	public List<Conversation> getConversations() {
@@ -1855,6 +1857,7 @@ public class XmppConnectionService extends Service {
 			};
 			mDatabaseWriterExecutor.execute(runnable);
 			this.accounts.remove(account);
+			this.mRosterSyncTaskManager.clear(account);
 			updateAccountUi();
 			getNotificationService().updateErrorNotification();
 			syncEnabledAccountSetting();

src/main/java/eu/siacs/conversations/utils/ReplacingSerialSingleThreadExecutor.java

@@ -2,13 +2,13 @@ package eu.siacs.conversations.utils;
 
 public class ReplacingSerialSingleThreadExecutor extends SerialSingleThreadExecutor {
 
-    public ReplacingSerialSingleThreadExecutor(boolean prepareLooper) {
+	public ReplacingSerialSingleThreadExecutor(boolean prepareLooper) {
-        super(ReplacingSerialSingleThreadExecutor.class.getName(), prepareLooper);
+		super(ReplacingSerialSingleThreadExecutor.class.getName(), prepareLooper);
-    }
+	}
 
-    @Override
+	@Override
-    public synchronized void execute(final Runnable r) {
+	public synchronized void execute(final Runnable r) {
-        tasks.clear();
+		tasks.clear();
-        super.execute(r);
+		super.execute(r);
-    }
+	}
 }

src/main/java/eu/siacs/conversations/utils/ReplacingTaskManager.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2018, Daniel Gultsch All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without modification,
+ * are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice, this
+ * list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation and/or
+ * other materials provided with the distribution.
+ *
+ * 3. Neither the name of the copyright holder nor the names of its contributors
+ * may be used to endorse or promote products derived from this software without
+ * specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package eu.siacs.conversations.utils;
+
+import java.util.HashMap;
+
+import eu.siacs.conversations.entities.Account;
+
+public class ReplacingTaskManager {
+
+	private final HashMap<Account, ReplacingSerialSingleThreadExecutor> executors = new HashMap<>();
+
+	public void execute(final Account account, Runnable runnable) {
+		ReplacingSerialSingleThreadExecutor executor;
+		synchronized (this.executors) {
+			executor = this.executors.get(account);
+			if (executor == null) {
+				executor = new ReplacingSerialSingleThreadExecutor(false);
+				this.executors.put(account, executor);
+			}
+			executor.execute(runnable);
+		}
+	}
+
+	public void clear(Account account) {
+		synchronized (this.executors) {
+			this.executors.remove(account);
+		}
+	}
+}

src/main/java/eu/siacs/conversations/utils/SerialSingleThreadExecutor.java

@@ -13,8 +13,8 @@ import eu.siacs.conversations.services.AttachFileToConversationRunnable;
 
 public class SerialSingleThreadExecutor implements Executor {
 
-	final Executor executor = Executors.newSingleThreadExecutor();
+	private final Executor executor = Executors.newSingleThreadExecutor();
-	protected final ArrayDeque<Runnable> tasks = new ArrayDeque<>();
+	final ArrayDeque<Runnable> tasks = new ArrayDeque<>();
 	private Runnable active;
 	private final String name;
 
@@ -22,26 +22,19 @@ public class SerialSingleThreadExecutor implements Executor {
 		this(name, false);
 	}
 
-	public SerialSingleThreadExecutor(String name, boolean prepareLooper) {
+	SerialSingleThreadExecutor(String name, boolean prepareLooper) {
 		if (prepareLooper) {
-			execute(new Runnable() {
+			execute(Looper::prepare);
-				@Override
-				public void run() {
-					Looper.prepare();
-				}
-			});
 		}
 		this.name = name;
 	}
 
 	public synchronized void execute(final Runnable r) {
-		tasks.offer(new Runnable() {
+		tasks.offer(() -> {
-			public void run() {
+			try {
-				try {
+				r.run();
-					r.run();
+			} finally {
-				} finally {
+				scheduleNext();
-					scheduleNext();
-				}
 			}
 		});
 		if (active == null) {
@@ -49,7 +42,7 @@ public class SerialSingleThreadExecutor implements Executor {
 		}
 	}
 
-	protected synchronized void scheduleNext() {
+	private synchronized void scheduleNext() {
 		if ((active =  tasks.poll()) != null) {
 			executor.execute(active);
 			int remaining = tasks.size();