narayana/anotherim
narayana/anotherim
3
2
Fork 0
Code Issues 3 Pull requests Actions Packages Projects Releases 4 Wiki Activity

fixed concurrent modification when killing mam queries

Browse source
This commit is contained in:
Daniel Gultsch 2018-03-01 08:27:30 +01:00
parent 2f3ea872d9
commit 0deffef8da
1 changed files with 67 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
src/main/java/eu/siacs/conversations/services/MessageArchiveService.java
View file

@ -17,7 +17,6 @@ import eu.siacs.conversations.generator.AbstractGenerator;
import eu.siacs.conversations.xml.Namespace; import eu.siacs.conversations.xml.Namespace;
import eu.siacs.conversations.xml.Element; import eu.siacs.conversations.xml.Element;
import eu.siacs.conversations.xmpp.OnAdvancedStreamFeaturesLoaded; import eu.siacs.conversations.xmpp.OnAdvancedStreamFeaturesLoaded;
import eu.siacs.conversations.xmpp.OnIqPacketReceived;
import eu.siacs.conversations.xmpp.jid.Jid; import eu.siacs.conversations.xmpp.jid.Jid;
import eu.siacs.conversations.xmpp.mam.MamReference; import eu.siacs.conversations.xmpp.mam.MamReference;
import eu.siacs.conversations.xmpp.stanzas.IqPacket; import eu.siacs.conversations.xmpp.stanzas.IqPacket;
@ -29,12 +28,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
private final HashSet<Query> queries = new HashSet<>(); private final HashSet<Query> queries = new HashSet<>();
private final ArrayList<Query> pendingQueries = new ArrayList<>(); private final ArrayList<Query> pendingQueries = new ArrayList<>();
public enum PagingOrder { MessageArchiveService(final XmppConnectionService service) {
NORMAL,
REVERSE
}
public MessageArchiveService(final XmppConnectionService service) {
this.mXmppConnectionService = service; this.mXmppConnectionService = service;
} }
@ -74,7 +68,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
this.execute(query); this.execute(query);
} }
public void catchupMUC(final Conversation conversation) { void catchupMUC(final Conversation conversation) {
if (conversation.getLastMessageTransmitted().getTimestamp() < 0 && conversation.countMessages() == 0) { if (conversation.getLastMessageTransmitted().getTimestamp() < 0 && conversation.countMessages() == 0) {
query(conversation, query(conversation,
new MamReference(0), new MamReference(0),
@ -137,7 +131,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
this.queries.add(reverseCatchup); this.queries.add(reverseCatchup);
this.execute(reverseCatchup); this.execute(reverseCatchup);
} }
query = new Query(conversation, maxCatchup, end, allowCatchup); query = new Query(conversation, maxCatchup, end, true);
} else { } else {
query = new Query(conversation, startActual, end, false); query = new Query(conversation, startActual, end, false);
} }
@ -151,7 +145,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
} }
public void executePendingQueries(final Account account) { void executePendingQueries(final Account account) {
List<Query> pending = new ArrayList<>(); List<Query> pending = new ArrayList<>();
synchronized (this.pendingQueries) { synchronized (this.pendingQueries) {
for (Iterator<Query> iterator = this.pendingQueries.iterator(); iterator.hasNext(); ) { for (Iterator<Query> iterator = this.pendingQueries.iterator(); iterator.hasNext(); ) {
@ -172,26 +166,23 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
if (account.getStatus() == Account.State.ONLINE) { if (account.getStatus() == Account.State.ONLINE) {
Log.d(Config.LOGTAG, account.getJid().toBareJid().toString() + ": running mam query " + query.toString()); Log.d(Config.LOGTAG, account.getJid().toBareJid().toString() + ": running mam query " + query.toString());
IqPacket packet = this.mXmppConnectionService.getIqGenerator().queryMessageArchiveManagement(query); IqPacket packet = this.mXmppConnectionService.getIqGenerator().queryMessageArchiveManagement(query);
this.mXmppConnectionService.sendIqPacket(account, packet, new OnIqPacketReceived() { this.mXmppConnectionService.sendIqPacket(account, packet, (a, p) -> {
@Override Element fin = p.findChild("fin", Namespace.MAM);
public void onIqPacketReceived(Account account, IqPacket packet) { if (p.getType() == IqPacket.TYPE.TIMEOUT) {
Element fin = packet.findChild("fin", Namespace.MAM);
if (packet.getType() == IqPacket.TYPE.TIMEOUT) {
synchronized (MessageArchiveService.this.queries) { synchronized (MessageArchiveService.this.queries) {
MessageArchiveService.this.queries.remove(query); MessageArchiveService.this.queries.remove(query);
if (query.hasCallback()) { if (query.hasCallback()) {
query.callback(false); query.callback(false);
} }
} }
} else if (packet.getType() == IqPacket.TYPE.RESULT && fin != null ) { } else if (p.getType() == IqPacket.TYPE.RESULT && fin != null) {
processFin(query, fin); processFin(query, fin);
} else if (packet.getType() == IqPacket.TYPE.RESULT && query.isLegacy()) { } else if (p.getType() == IqPacket.TYPE.RESULT && query.isLegacy()) {
//do nothing //do nothing
} else { } else {
Log.d(Config.LOGTAG, account.getJid().toBareJid().toString() + ": error executing mam: " + packet.toString()); Log.d(Config.LOGTAG, a.getJid().toBareJid().toString() + ": error executing mam: " + p.toString());
finalizeQuery(query, true); finalizeQuery(query, true);
} }
}
}); });
} else { } else {
synchronized (this.pendingQueries) { synchronized (this.pendingQueries) {
@ -222,7 +213,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
} }
public boolean inCatchup(Account account) { boolean inCatchup(Account account) {
synchronized (this.queries) { synchronized (this.queries) {
for (Query query : queries) { for (Query query : queries) {
if (query.account == account && query.isCatchup() && query.getWith() == null) { if (query.account == account && query.isCatchup() && query.getWith() == null) {
@ -233,7 +224,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
return false; return false;
} }
public boolean queryInProgress(Conversation conversation, XmppConnectionService.OnMoreMessagesLoaded callback) { boolean queryInProgress(Conversation conversation, XmppConnectionService.OnMoreMessagesLoaded callback) {
synchronized (this.queries) { synchronized (this.queries) {
for (Query query : queries) { for (Query query : queries) {
if (query.conversation == conversation) { if (query.conversation == conversation) {
@ -307,15 +298,19 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
} }
public void kill(Conversation conversation) { void kill(Conversation conversation) {
final ArrayList<Query> toBeKilled = new ArrayList<>();
synchronized (this.queries) { synchronized (this.queries) {
for (Query q : queries) { for (Query q : queries) {
if (q.conversation == conversation) { if (q.conversation == conversation) {
toBeKilled.add(q);
}
}
}
for (Query q : toBeKilled) {
kill(q); kill(q);
} }
} }
}
}
private void kill(Query query) { private void kill(Query query) {
Log.d(Config.LOGTAG, query.getAccount().getJid().toBareJid() + ": killing mam query prematurely"); Log.d(Config.LOGTAG, query.getAccount().getJid().toBareJid() + ": killing mam query prematurely");
@ -359,7 +354,13 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
} }
public enum PagingOrder {
NORMAL,
REVERSE
}
public class Query { public class Query {
public HashSet<ReceiptRequest> pendingReceiptRequests = new HashSet<>();
private int totalCount = 0; private int totalCount = 0;
private int actualCount = 0; private int actualCount = 0;
private int actualInThisQuery = 0; private int actualInThisQuery = 0;
@ -372,17 +373,16 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
private PagingOrder pagingOrder = PagingOrder.NORMAL; private PagingOrder pagingOrder = PagingOrder.NORMAL;
private XmppConnectionService.OnMoreMessagesLoaded callback = null; private XmppConnectionService.OnMoreMessagesLoaded callback = null;
private boolean catchup = true; private boolean catchup = true;
public HashSet<ReceiptRequest> pendingReceiptRequests = new HashSet<>();
public Query(Conversation conversation, MamReference start, long end, boolean catchup) { Query(Conversation conversation, MamReference start, long end, boolean catchup) {
this(conversation.getAccount(), catchup ? start : start.timeOnly(), end); this(conversation.getAccount(), catchup ? start : start.timeOnly(), end);
this.conversation = conversation; this.conversation = conversation;
this.pagingOrder = catchup ? PagingOrder.NORMAL : PagingOrder.REVERSE; this.pagingOrder = catchup ? PagingOrder.NORMAL : PagingOrder.REVERSE;
this.catchup = catchup; this.catchup = catchup;
} }
public Query(Account account, MamReference start, long end) { Query(Account account, MamReference start, long end) {
this.account = account; this.account = account;
if (start.getReference() != null) { if (start.getReference() != null) {
this.reference = start.getReference(); this.reference = start.getReference();
@ -422,7 +422,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
return query; return query;
} }
public Query prev(String reference) { Query prev(String reference) {
Query query = page(reference); Query query = page(reference);
query.pagingOrder = PagingOrder.REVERSE; query.pagingOrder = PagingOrder.REVERSE;
return query; return query;
@ -490,11 +490,11 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
this.actualCount++; this.actualCount++;
} }
public int getTotalCount() { int getTotalCount() {
return this.totalCount; return this.totalCount;
} }
public int getActualMessageCount() { int getActualMessageCount() {
return this.actualCount; return this.actualCount;
} }
@ -530,7 +530,7 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
builder.append(", end="); builder.append(", end=");
builder.append(AbstractGenerator.getTimestamp(this.end)); builder.append(AbstractGenerator.getTimestamp(this.end));
builder.append(", order="+pagingOrder.toString()); builder.append(", order=").append(pagingOrder.toString());
if (this.reference != null) { if (this.reference != null) {
if (this.pagingOrder == PagingOrder.NORMAL) { if (this.pagingOrder == PagingOrder.NORMAL) {
builder.append(", after="); builder.append(", after=");
@ -539,11 +539,11 @@ public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
} }
builder.append(this.reference); builder.append(this.reference);
} }
builder.append(", catchup="+Boolean.toString(catchup)); builder.append(", catchup=").append(Boolean.toString(catchup));
return builder.toString(); return builder.toString();
} }
public boolean hasCallback() { boolean hasCallback() {
return this.callback != null; return this.callback != null;
} }
} }