mirror of
https://github.com/vanitasvitae/Smack.git
synced 2024-11-22 03:52:06 +01:00
Smack 4.1.5
-----BEGIN PGP SIGNATURE----- Version: GnuPG v2 iQF8BAABCgBmBQJWUjTuXxSAAAAAAC4AKGlzc3Vlci1mcHJAbm90YXRpb25zLm9w ZW5wZ3AuZmlmdGhob3JzZW1hbi5uZXQxMzU3QjAxODY1QjI1MDNDMTg0NTNEMjA4 Q0FDMkE5Njc4NTQ4RTM1AAoJEIysKpZ4VI41zp8H/iRB0CtFvxVBFwRpP2HB0Iyc VzdxJTVXgJ7s8V3Us2Gmm0chUvs1wybshIdZ090NX7huZ4F1rWrOP7qt9S4soKgX LNdKTKTTNJ61jktLkG1HjA8yJr1Zu+XFK10Ne8997N9TnhbcVYAhsuTb8SRE8dlE G1Qihe2guJCCnWJhz9bsCXyAFt/fM5rSb/sgs/kXkE1W5bYeMHpFMvqMfab8NPBP fkt/NSxfqLfrRJneynnSwGSvRitDxpfRCcxLktTdDO31eryayIyiQIhaUhFYIGJQ tKx0/7HdCkgdUueBdtoQ/O02rjJ+kYqN2F3CErcIbB//3R6WKNF61XzOY81egYc= =dklO -----END PGP SIGNATURE----- Merge tag '4.1.5' Smack 4.1.5
This commit is contained in:
commit
8dcb3c11ff
4 changed files with 89 additions and 13 deletions
|
@ -141,6 +141,18 @@ hr {
|
||||||
|
|
||||||
<div id="pageBody">
|
<div id="pageBody">
|
||||||
|
|
||||||
|
<h2>4.1.5 -- <span style="font-weight: normal;">2015-11-22</span></h2>
|
||||||
|
|
||||||
|
<h2> Bug
|
||||||
|
</h2>
|
||||||
|
<ul>
|
||||||
|
<li>[<a href='https://igniterealtime.org/issues/browse/SMACK-698'>SMACK-698</a>] - Time creates invalid XML
|
||||||
|
</li>
|
||||||
|
<li>[<a href='https://igniterealtime.org/issues/browse/SMACK-700'>SMACK-700</a>] - Duplicate stanzas in unacknowledgedStanzas queue when stream is resumed
|
||||||
|
</li>
|
||||||
|
<li>[<a href='https://igniterealtime.org/issues/browse/SMACK-702'>SMACK-702</a>] - RejectedExecutionException in AbstractXMPPConnection.processPacket() causes connection Termination
|
||||||
|
</li>
|
||||||
|
</ul>
|
||||||
|
|
||||||
<h2>4.1.4 -- <span style="font-weight: normal;">2015-09-14</span></h2>
|
<h2>4.1.4 -- <span style="font-weight: normal;">2015-09-14</span></h2>
|
||||||
|
|
||||||
|
|
|
@ -27,14 +27,12 @@ import java.util.LinkedList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.concurrent.ArrayBlockingQueue;
|
|
||||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||||
import java.util.concurrent.CopyOnWriteArraySet;
|
import java.util.concurrent.CopyOnWriteArraySet;
|
||||||
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.ExecutorService;
|
||||||
import java.util.concurrent.Executors;
|
import java.util.concurrent.Executors;
|
||||||
import java.util.concurrent.ScheduledExecutorService;
|
import java.util.concurrent.ScheduledExecutorService;
|
||||||
import java.util.concurrent.ScheduledFuture;
|
import java.util.concurrent.ScheduledFuture;
|
||||||
import java.util.concurrent.ThreadPoolExecutor;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
import java.util.concurrent.locks.Lock;
|
import java.util.concurrent.locks.Lock;
|
||||||
|
@ -75,6 +73,7 @@ import org.jivesoftware.smack.parsing.ParsingExceptionCallback;
|
||||||
import org.jivesoftware.smack.provider.ExtensionElementProvider;
|
import org.jivesoftware.smack.provider.ExtensionElementProvider;
|
||||||
import org.jivesoftware.smack.provider.ProviderManager;
|
import org.jivesoftware.smack.provider.ProviderManager;
|
||||||
import org.jivesoftware.smack.sasl.core.SASLAnonymous;
|
import org.jivesoftware.smack.sasl.core.SASLAnonymous;
|
||||||
|
import org.jivesoftware.smack.util.BoundedThreadPoolExecutor;
|
||||||
import org.jivesoftware.smack.util.DNSUtil;
|
import org.jivesoftware.smack.util.DNSUtil;
|
||||||
import org.jivesoftware.smack.util.Objects;
|
import org.jivesoftware.smack.util.Objects;
|
||||||
import org.jivesoftware.smack.util.PacketParserUtils;
|
import org.jivesoftware.smack.util.PacketParserUtils;
|
||||||
|
@ -233,8 +232,8 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
|
||||||
* important that we use a <b>single threaded ExecutorService</b> in order to guarantee that the
|
* important that we use a <b>single threaded ExecutorService</b> in order to guarantee that the
|
||||||
* PacketListeners are invoked in the same order the stanzas arrived.
|
* PacketListeners are invoked in the same order the stanzas arrived.
|
||||||
*/
|
*/
|
||||||
private final ThreadPoolExecutor executorService = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
|
private final BoundedThreadPoolExecutor executorService = new BoundedThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
|
||||||
new ArrayBlockingQueue<Runnable>(100), new SmackExecutorThreadFactory(this, "Incoming Processor"));
|
100, new SmackExecutorThreadFactory(this, "Incoming Processor"));
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This scheduled thread pool executor is used to remove pending callbacks.
|
* This scheduled thread pool executor is used to remove pending callbacks.
|
||||||
|
@ -1002,12 +1001,13 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
|
||||||
* they are a match with the filter.
|
* they are a match with the filter.
|
||||||
*
|
*
|
||||||
* @param stanza the stanza to process.
|
* @param stanza the stanza to process.
|
||||||
|
* @throws InterruptedException
|
||||||
*/
|
*/
|
||||||
protected void processStanza(final Stanza stanza) {
|
protected void processStanza(final Stanza stanza) throws InterruptedException {
|
||||||
assert(stanza != null);
|
assert(stanza != null);
|
||||||
lastStanzaReceived = System.currentTimeMillis();
|
lastStanzaReceived = System.currentTimeMillis();
|
||||||
// Deliver the incoming packet to listeners.
|
// Deliver the incoming packet to listeners.
|
||||||
executorService.submit(new Runnable() {
|
executorService.executeBlocking(new Runnable() {
|
||||||
@Override
|
@Override
|
||||||
public void run() {
|
public void run() {
|
||||||
invokePacketCollectorsAndNotifyRecvListeners(stanza);
|
invokePacketCollectorsAndNotifyRecvListeners(stanza);
|
||||||
|
|
|
@ -0,0 +1,63 @@
|
||||||
|
/**
|
||||||
|
*
|
||||||
|
* Copyright 2015 Florian Schmaus
|
||||||
|
*
|
||||||
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
* you may not use this file except in compliance with the License.
|
||||||
|
* You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing, software
|
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
* See the License for the specific language governing permissions and
|
||||||
|
* limitations under the License.
|
||||||
|
*/
|
||||||
|
package org.jivesoftware.smack.util;
|
||||||
|
|
||||||
|
import java.util.concurrent.ArrayBlockingQueue;
|
||||||
|
import java.util.concurrent.RejectedExecutionException;
|
||||||
|
import java.util.concurrent.Semaphore;
|
||||||
|
import java.util.concurrent.ThreadFactory;
|
||||||
|
import java.util.concurrent.ThreadPoolExecutor;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
public class BoundedThreadPoolExecutor extends ThreadPoolExecutor {
|
||||||
|
|
||||||
|
private final Semaphore semaphore;
|
||||||
|
|
||||||
|
public BoundedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
|
||||||
|
TimeUnit unit, int bound, ThreadFactory threadFactory) {
|
||||||
|
// One could think that the array blocking queue bound should be "bound - 1" because the bound protected by the
|
||||||
|
// Semaphore also includes the "slot" in the worker thread executing the Runnable. But using that as bound could
|
||||||
|
// actually cause a RejectedExecutionException as the queue could fill up while the worker thread remains
|
||||||
|
// unscheduled and is thus not removing any entries.
|
||||||
|
super(corePoolSize, maximumPoolSize, keepAliveTime,
|
||||||
|
unit, new ArrayBlockingQueue<Runnable>(bound), threadFactory);
|
||||||
|
semaphore = new Semaphore(bound);
|
||||||
|
}
|
||||||
|
|
||||||
|
public void executeBlocking(final Runnable command) throws InterruptedException {
|
||||||
|
semaphore.acquire();
|
||||||
|
try {
|
||||||
|
execute(new Runnable() {
|
||||||
|
@Override
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
command.run();
|
||||||
|
} finally {
|
||||||
|
semaphore.release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
} catch (Exception e) {
|
||||||
|
semaphore.release();
|
||||||
|
if (e instanceof RejectedExecutionException) {
|
||||||
|
throw (RejectedExecutionException) e;
|
||||||
|
} else {
|
||||||
|
throw new RuntimeException(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -1086,16 +1086,17 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
||||||
if (!smSessionId.equals(resumed.getPrevId())) {
|
if (!smSessionId.equals(resumed.getPrevId())) {
|
||||||
throw new StreamIdDoesNotMatchException(smSessionId, resumed.getPrevId());
|
throw new StreamIdDoesNotMatchException(smSessionId, resumed.getPrevId());
|
||||||
}
|
}
|
||||||
|
// Mark SM as enabled and resumption as successful.
|
||||||
|
smResumedSyncPoint.reportSuccess();
|
||||||
|
smEnabledSyncPoint.reportSuccess();
|
||||||
// First, drop the stanzas already handled by the server
|
// First, drop the stanzas already handled by the server
|
||||||
processHandledCount(resumed.getHandledCount());
|
processHandledCount(resumed.getHandledCount());
|
||||||
// Then re-send what is left in the unacknowledged queue
|
// Then re-send what is left in the unacknowledged queue
|
||||||
List<Stanza> stanzasToResend = new LinkedList<Stanza>();
|
List<Stanza> stanzasToResend = new ArrayList<>(unacknowledgedStanzas.size());
|
||||||
stanzasToResend.addAll(unacknowledgedStanzas);
|
unacknowledgedStanzas.drainTo(stanzasToResend);
|
||||||
for (Stanza stanza : stanzasToResend) {
|
for (Stanza stanza : stanzasToResend) {
|
||||||
packetWriter.sendStreamElement(stanza);
|
sendStanzaInternal(stanza);
|
||||||
}
|
}
|
||||||
smResumedSyncPoint.reportSuccess();
|
|
||||||
smEnabledSyncPoint.reportSuccess();
|
|
||||||
// If there where stanzas resent, then request a SM ack for them.
|
// If there where stanzas resent, then request a SM ack for them.
|
||||||
// Writer's sendStreamElement() won't do it automatically based on
|
// Writer's sendStreamElement() won't do it automatically based on
|
||||||
// predicates.
|
// predicates.
|
||||||
|
@ -1469,7 +1470,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
@Deprecated
|
||||||
public static void setUseStreamManagementResumptiodDefault(boolean useSmResumptionDefault) {
|
public static void setUseStreamManagementResumptiodDefault(boolean useSmResumptionDefault) {
|
||||||
setUseStreamManagementDefault(useSmResumptionDefault);
|
setUseStreamManagementResumptionDefault(useSmResumptionDefault);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -1765,7 +1766,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
||||||
private void processHandledCount(long handledCount) throws StreamManagementCounterError {
|
private void processHandledCount(long handledCount) throws StreamManagementCounterError {
|
||||||
long ackedStanzasCount = SMUtils.calculateDelta(handledCount, serverHandledStanzasCount);
|
long ackedStanzasCount = SMUtils.calculateDelta(handledCount, serverHandledStanzasCount);
|
||||||
final List<Stanza> ackedStanzas = new ArrayList<Stanza>(
|
final List<Stanza> ackedStanzas = new ArrayList<Stanza>(
|
||||||
handledCount <= Integer.MAX_VALUE ? (int) handledCount
|
ackedStanzasCount <= Integer.MAX_VALUE ? (int) ackedStanzasCount
|
||||||
: Integer.MAX_VALUE);
|
: Integer.MAX_VALUE);
|
||||||
for (long i = 0; i < ackedStanzasCount; i++) {
|
for (long i = 0; i < ackedStanzasCount; i++) {
|
||||||
Stanza ackedStanza = unacknowledgedStanzas.poll();
|
Stanza ackedStanza = unacknowledgedStanzas.poll();
|
||||||
|
|
Loading…
Reference in a new issue