This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/main by this push:
new a45fa4c73c Minor fixes from code review
a45fa4c73c is described below
commit a45fa4c73c03d7925371e80f8fa1826470a2289a
Author: remm <[email protected]>
AuthorDate: Thu Jul 9 20:42:02 2026 +0200
Minor fixes from code review
Direct access to entries in AbstractReplicatedMap (it seems the double
lookup is obvious now and can be removed, but looking at the history
things were a lot more convoluted before).
Make ExceptionUtils constructor private, hopefully no regression on this
one ...
NioSender.setMessage looked really weird with offset > 0 (which does not
happen).
PooledSender called a second time keepalive.
ReplicatedMap.publishEntryInfo should return an empty array instead of
null or an exception.
Update javadoc for AbstractReplicatedMap. The javadoc generated (using
OpenCode) looks good.
---
.../tribes/membership/LocalStrings.properties | 1 +
.../catalina/tribes/membership/StaticMember.java | 2 +-
.../membership/StaticMembershipProvider.java | 7 +-
.../tribes/tipis/AbstractReplicatedMap.java | 114 +++++++++++++--------
.../catalina/tribes/tipis/ReplicatedMap.java | 8 +-
.../catalina/tribes/transport/AbstractSender.java | 2 +-
.../catalina/tribes/transport/PooledSender.java | 14 ++-
.../catalina/tribes/transport/ReceiverBase.java | 31 ++----
.../tribes/transport/ReplicationTransmitter.java | 2 +-
.../catalina/tribes/transport/RxTaskPool.java | 5 +-
.../tribes/transport/nio/LocalStrings.properties | 1 +
.../catalina/tribes/transport/nio/NioReceiver.java | 10 +-
.../tribes/transport/nio/NioReplicationTask.java | 8 +-
.../catalina/tribes/transport/nio/NioSender.java | 8 +-
.../tribes/transport/nio/PooledParallelSender.java | 7 +-
java/org/apache/catalina/tribes/util/Arrays.java | 16 ++-
.../catalina/tribes/util/ExceptionUtils.java | 2 +-
.../apache/catalina/tribes/util/UUIDGenerator.java | 23 +++--
18 files changed, 141 insertions(+), 120 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
index 687814df7c..6ab3bf6ff5 100644
--- a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
@@ -59,6 +59,7 @@ memberImpl.large.payload=Payload is too large for tribes to
handle.
memberImpl.notEnough.bytes=Not enough bytes in member package.
memberImpl.package.small=Member package too small to validate.
+staticMember.invalid.host=Invalid host [{0}]
staticMember.invalid.uuidLength=UUID must be exactly 16 bytes, not:[{0}]
staticMembershipProvider.heartbeat.failed=Unable to send
StaticMembershipProvider.ping message.
diff --git a/java/org/apache/catalina/tribes/membership/StaticMember.java
b/java/org/apache/catalina/tribes/membership/StaticMember.java
index 7df9845e02..4baf279aed 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMember.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMember.java
@@ -69,7 +69,7 @@ public class StaticMember extends MemberImpl {
try {
setHostname(host);
} catch (IOException ioe) {
- throw new RuntimeException(ioe);
+ throw new
IllegalArgumentException(sm.getString("staticMember.invalid.host", host), ioe);
}
}
diff --git
a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
index 489c976a65..3e94478057 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
@@ -119,6 +119,7 @@ public class StaticMembershipProvider extends
MembershipProviderBase
@Override
public void init(Properties properties) throws Exception {
+ // All the default properties have been set by
StaticMembershipService.setDefaults, so no need to check
String expirationTimeStr = properties.getProperty("expirationTime");
this.expirationTime = Long.parseLong(expirationTimeStr);
String connectTimeoutStr = properties.getProperty("connectTimeout");
@@ -396,7 +397,7 @@ public class StaticMembershipProvider extends
MembershipProviderBase
protected void checkExpired() {
Member[] expired = membership.expire(expirationTime);
for (Member member : expired) {
- membershipListener.memberDisappeared(member);
+ memberDisappeared(member);
}
}
@@ -506,8 +507,8 @@ public class StaticMembershipProvider extends
MembershipProviderBase
@Override
public String toString() {
- return "MemberMessage[" + "name=" + new String(membershipId) + ";
type=" + getTypeDesc() + "; member=" +
- member + ']';
+ return "MemberMessage[" + "name=" + new String(membershipId,
StandardCharsets.ISO_8859_1) + "; type=" +
+ getTypeDesc() + "; member=" + member + ']';
}
/**
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index c2f527d630..ad005129bf 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -183,17 +183,28 @@ public abstract class AbstractReplicatedMap<K, V>
//
------------------------------------------------------------------------------
/**
- * Creates a new map.
- *
- * @param owner The map owner
- * @param channel The channel to use for communication
- * @param timeout long - timeout for RPC messages
- * @param mapContextName String - unique name for this map, to allow
multiple maps per channel
- * @param initialCapacity int - the size of this map, see HashMap
- * @param loadFactor float - load factor, see HashMap
- * @param channelSendOptions Send options
- * @param cls - a list of classloaders to be used for
deserialization of objects.
- * @param terminate - Flag for whether to terminate this map that
failed to start.
+ * Creates a new replicated map with the specified configuration.
+ *
+ * Initializes the internal {@link ConcurrentHashMap} with the given
capacity and load factor,
+ * then delegates to {@link #init(MapOwner, Channel, String, long, int,
ClassLoader[], boolean)}
+ * to configure the map, register listeners, and perform state transfer
from existing cluster members.
+ *
+ * @param owner the object that owns this map and receives
notifications when entries
+ * become primary on this node
+ * @param channel the channel used for cluster communication;
must not be {@code null}
+ * @param timeout the timeout in milliseconds for RPC messages
+ * @param mapContextName a unique name for this map that distinguishes
it from other maps
+ * sharing the same channel
+ * @param initialCapacity the initial capacity of the internal map, as
defined by
+ * {@link
ConcurrentHashMap#ConcurrentHashMap(int, float, int)}
+ * @param loadFactor the load factor of the internal map, as
defined by
+ * {@link
ConcurrentHashMap#ConcurrentHashMap(int, float, int)}
+ * @param channelSendOptions the send options to use when transmitting
messages through the channel
+ * @param cls an array of class loaders used during
deserialization of received objects,
+ * or {@code null} if no external loaders are
needed
+ * @param terminate if {@code true}, the map will terminate
itself and throw a {@link RuntimeException}
+ * when initialization fails; if {@code false},
the failure is logged and the map
+ * continues in a degraded state
*/
public AbstractReplicatedMap(MapOwner owner, Channel channel, long
timeout, String mapContextName,
int initialCapacity, float loadFactor, int channelSendOptions,
ClassLoader[] cls, boolean terminate) {
@@ -218,16 +229,36 @@ public abstract class AbstractReplicatedMap<K, V>
}
/**
- * Initializes the map by creating the RPC channel, registering itself as
a channel listener This method is also
- * responsible for initiating the state transfer
+ * Initializes the replicated map by configuring its internal state,
registering with the channel,
+ * and performing state transfer from existing cluster members.
*
- * @param owner Object
- * @param channel Channel
- * @param mapContextName String
- * @param timeout long
- * @param channelSendOptions int
- * @param cls ClassLoader[]
- * @param terminate - Flag for whether to terminate this map that
failed to start.
+ * This method performs the following steps:
+ * <ol>
+ * <li>Stores the provided configuration parameters</li>
+ * <li>Creates an {@link RpcChannel} for sending RPC messages</li>
+ * <li>Registers this map as a {@link ChannelListener} and {@link
MembershipListener} on the channel</li>
+ * <li>Broadcasts a {@code MSG_INIT} message to announce this map to
existing cluster members</li>
+ * <li>Invokes {@link #transferState()} to receive the current map state
from an existing member</li>
+ * <li>Broadcasts a {@code MSG_START} message to signal that this map is
ready for normal operation</li>
+ * </ol>
+ *
+ * If a {@link ChannelException} occurs during the broadcast or state
transfer steps, the behavior
+ * depends on the {@code terminate} flag: when {@code true}, the map is
torn down via {@link #breakdown()}
+ * and a {@link RuntimeException} is thrown; when {@code false}, the error
is logged and the map
+ * continues in a partially initialized state.
+ *
+ * @param owner the object that owns this map and receives
notifications when entries
+ * become primary on this node
+ * @param channel the channel used for cluster communication;
must not be {@code null}
+ * @param mapContextName a unique name for this map that distinguishes
it from other maps
+ * sharing the same channel
+ * @param timeout the timeout in milliseconds for RPC messages
+ * @param channelSendOptions the send options to use when transmitting
messages through the channel
+ * @param cls an array of class loaders used during
deserialization of received objects,
+ * or {@code null} if no external loaders are
needed
+ * @param terminate if {@code true}, the map will terminate
itself and throw a {@link RuntimeException}
+ * when initialization fails; if {@code false},
the failure is logged and the map
+ * continues in a degraded state
*/
protected void init(MapOwner owner, Channel channel, String
mapContextName, long timeout, int channelSendOptions,
ClassLoader[] cls, boolean terminate) {
@@ -1323,12 +1354,8 @@ public abstract class AbstractReplicatedMap<K, V>
entry.setCopy(false);
entry.setPrimary(channel.getLocalMember(false));
- V old = null;
-
- // make sure that any old values get removed
- if (containsKey(key)) {
- old = remove(key);
- }
+ // Make sure that any old values get removed, even if ours was only a
proxy
+ V old = remove(key, notify);
try {
if (notify) {
Member[] backup = publishEntryInfo(key, value);
@@ -1390,7 +1417,7 @@ public abstract class AbstractReplicatedMap<K, V>
}
/**
- * Returns the entire contents of the map Map.Entry.getValue() will return
a LazyReplicatedMap.MapEntry object
+ * Returns the entire contents of the map Map.Entry.getValue() will return
a AbstractReplicatedMap.MapEntry object
* containing all the information about the object.
*
* @return Set
@@ -1426,7 +1453,7 @@ public abstract class AbstractReplicatedMap<K, V>
public Set<Map.Entry<K,V>> entrySet() {
LinkedHashSet<Map.Entry<K,V>> set = new
LinkedHashSet<>(innerMap.size());
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
- MapEntry<K,V> entry = innerMap.get(e.getKey());
+ MapEntry<K,V> entry = e.getValue();
if (entry != null && entry.isActive()) {
set.add(entry);
}
@@ -1446,7 +1473,7 @@ public abstract class AbstractReplicatedMap<K, V>
LinkedHashSet<K> set = new LinkedHashSet<>(innerMap.size());
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
K key = e.getKey();
- MapEntry<K,V> entry = innerMap.get(key);
+ MapEntry<K,V> entry = e.getValue();
if (entry != null && entry.isActive()) {
set.add(key);
}
@@ -1463,15 +1490,12 @@ public abstract class AbstractReplicatedMap<K, V>
*/
@Override
public int size() {
- // todo, implement a counter variable instead
// only count active members in this node
int counter = 0;
- for (Entry<K,?> e : innerMap.entrySet()) {
- if (e != null) {
- MapEntry<K,V> entry = innerMap.get(e.getKey());
- if (entry != null && entry.isActive() && entry.getValue() !=
null) {
- counter++;
- }
+ for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
+ MapEntry<K,V> entry = e.getValue();
+ if (entry != null && entry.isActive() && entry.getValue() != null)
{
+ counter++;
}
}
return counter;
@@ -1496,7 +1520,7 @@ public abstract class AbstractReplicatedMap<K, V>
public Collection<V> values() {
List<V> values = new ArrayList<>();
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
- MapEntry<K,V> entry = innerMap.get(e.getKey());
+ MapEntry<K,V> entry = e.getValue();
if (entry != null && entry.isActive() && entry.getValue() != null)
{
values.add(entry.getValue());
}
@@ -1735,12 +1759,20 @@ public abstract class AbstractReplicatedMap<K, V>
@Override
public int hashCode() {
- return key.hashCode();
+ return key == null ? 0 : key.hashCode();
}
@Override
public boolean equals(Object o) {
- return key.equals(o);
+ if (!(o instanceof MapEntry)) {
+ return false;
+ }
+ @SuppressWarnings("rawtypes")
+ MapEntry other = (MapEntry) o;
+ if (key == null) {
+ return other.key == null;
+ }
+ return key.equals(other.key);
}
/**
@@ -1851,8 +1883,8 @@ public abstract class AbstractReplicatedMap<K, V>
*/
@Override
public String toString() {
- return "MapMessage[context=" + new String(mapId) + "; type=" +
getTypeDesc() + "; key=" + key + "; value=" +
- value + ']';
+ return "MapMessage[context=" + new String(mapId,
StandardCharsets.ISO_8859_1) + "; type=" + getTypeDesc() +
+ "; key=" + key + "; value=" + value + ']';
}
/**
diff --git a/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
b/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
index c436c0fb8a..527a8d9ee1 100644
--- a/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
@@ -149,8 +149,8 @@ public class ReplicatedMap<K, V> extends
AbstractReplicatedMap<K,V> {
// select a backup node
Member[] backup = getMapMembers();
- if (backup == null || backup.length == 0) {
- return null;
+ if (backup.length == 0) {
+ return backup;
}
try {
@@ -174,7 +174,7 @@ public class ReplicatedMap<K, V> extends
AbstractReplicatedMap<K,V> {
if (realFaultyMembers.length != 0) {
backup = excludeFromSet(realFaultyMembers, backup);
if (backup.length == 0) {
- throw e;
+ return backup;
} else {
if (getLog().isWarnEnabled()) {
getLog().warn(sm.getString("replicatedMap.unableReplicate.completely", key,
@@ -248,7 +248,7 @@ public class ReplicatedMap<K, V> extends
AbstractReplicatedMap<K,V> {
}
}
- } // while
+ }
long complete = System.currentTimeMillis() - start;
if (log.isInfoEnabled()) {
log.info(sm.getString("replicatedMap.relocate.complete",
Long.toString(complete)));
diff --git a/java/org/apache/catalina/tribes/transport/AbstractSender.java
b/java/org/apache/catalina/tribes/transport/AbstractSender.java
index eb6d5ed677..49ff819ca9 100644
--- a/java/org/apache/catalina/tribes/transport/AbstractSender.java
+++ b/java/org/apache/catalina/tribes/transport/AbstractSender.java
@@ -118,7 +118,7 @@ public abstract class AbstractSender implements DataSender {
public boolean keepalive() {
boolean disconnect = false;
if (isUdpBased()) {
- disconnect = true; // always disconnect UDP, TODO optimize the
keepalive handling
+ disconnect = true; // always disconnect UDP
} else if (keepAliveCount >= 0 && requestCount > keepAliveCount) {
disconnect = true;
} else if (keepAliveTime >= 0 && (System.currentTimeMillis() -
connectTime) > keepAliveTime) {
diff --git a/java/org/apache/catalina/tribes/transport/PooledSender.java
b/java/org/apache/catalina/tribes/transport/PooledSender.java
index 1f25b99083..35a267c372 100644
--- a/java/org/apache/catalina/tribes/transport/PooledSender.java
+++ b/java/org/apache/catalina/tribes/transport/PooledSender.java
@@ -64,7 +64,6 @@ public abstract class PooledSender extends AbstractSender
implements MultiPointS
* @param sender the sender to return
*/
public void returnSender(DataSender sender) {
- sender.keepalive();
queue.returnSender(sender);
}
@@ -221,12 +220,12 @@ public abstract class PooledSender extends AbstractSender
implements MultiPointS
long delta = System.currentTimeMillis() - start;
if (delta > timeout && timeout > 0) {
return null;
- } else {
- try {
- wait(Math.max(timeout - delta, 1));
- } catch (InterruptedException x) {
- // Ignore
- }
+ }
+ long remaining = timeout > 0 ? Math.max(timeout - delta, 1) :
1;
+ try {
+ wait(remaining);
+ } catch (InterruptedException x) {
+ // Ignore
}
}
}
@@ -236,7 +235,6 @@ public abstract class PooledSender extends AbstractSender
implements MultiPointS
sender.disconnect();
return;
}
- // to do
inuse.remove(sender);
// just in case the limit has changed
if (notinuse.size() < this.getLimit()) {
diff --git a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
index 1351386dff..0ca6f9700e 100644
--- a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
+++ b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
@@ -145,21 +145,11 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
channel = null;
}
- /**
- * Returns the message listener for this receiver.
- *
- * @return the message listener
- */
@Override
public MessageListener getMessageListener() {
return listener;
}
- /**
- * Returns the port on which this receiver is listening.
- *
- * @return the port number
- */
@Override
public int getPort() {
return port;
@@ -183,11 +173,6 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
return txBufSize;
}
- /**
- * Sets the message listener for this receiver.
- *
- * @param listener the message listener to set
- */
@Override
public void setMessageListener(MessageListener listener) {
this.listener = listener;
@@ -280,11 +265,12 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
*/
protected int bindUdp(DatagramSocket socket, int portstart, int retries)
throws IOException {
InetSocketAddress addr = null;
+ int port = portstart;
while (retries > 0) {
try {
- addr = new InetSocketAddress(getBind(), portstart);
+ addr = new InetSocketAddress(getBind(), port);
socket.bind(addr);
- setUdpPort(portstart);
+ setUdpPort(port);
log.info(sm.getString("receiverBase.udp.bind", addr));
return 0;
} catch (IOException ioe) {
@@ -293,13 +279,12 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
log.info(sm.getString("receiverBase.unable.bind.udp",
addr));
throw ioe;
}
- portstart++;
+ port++;
try {
Thread.sleep(25);
} catch (InterruptedException ti) {
Thread.currentThread().interrupt();
}
- retries = bindUdp(socket, portstart, retries);
}
}
return retries;
@@ -373,11 +358,6 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
return this.host;
}
- /**
- * Returns the host address for this receiver.
- *
- * @return the host address
- */
@Override
public String getHost() {
return getAddress();
@@ -397,6 +377,7 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
*
* @return true if the receiver is listening
*/
+ @Deprecated
public boolean doListen() {
return listen;
}
@@ -406,6 +387,7 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
*
* @return the message listener
*/
+ @Deprecated
public MessageListener getListener() {
return listener;
}
@@ -616,6 +598,7 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
*
* @param listener the message listener to set
*/
+ @Deprecated
public void setListener(MessageListener listener) {
this.listener = listener;
}
diff --git
a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
index 13a3db44b9..dfc9b5a83a 100644
--- a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
+++ b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
@@ -108,7 +108,7 @@ public class ReplicationTransmitter implements
ChannelSender {
/**
* Call transmitter to check for sender socket status
*
- * @see org.apache.catalina.ha.tcp.SimpleTcpCluster#backgroundProcess()
+ * @see org.apache.catalina.tribes.ChannelSender#heartbeat()
*/
@Override
public void heartbeat() {
diff --git a/java/org/apache/catalina/tribes/transport/RxTaskPool.java
b/java/org/apache/catalina/tribes/transport/RxTaskPool.java
index 6120bd379e..bf75c9aa05 100644
--- a/java/org/apache/catalina/tribes/transport/RxTaskPool.java
+++ b/java/org/apache/catalina/tribes/transport/RxTaskPool.java
@@ -21,8 +21,7 @@ import java.util.Iterator;
import java.util.List;
/**
- * A very simple thread pool class. The pool size is set at construction time
and remains fixed. Threads are cycled
- * through a FIFO idle queue.
+ * A very simple thread pool class. Threads are cycled through a FIFO idle
queue.
*/
public class RxTaskPool {
@@ -30,7 +29,7 @@ public class RxTaskPool {
final List<AbstractRxTask> used = new ArrayList<>();
final Object mutex = new Object();
- boolean running = true;
+ volatile boolean running = true;
private int maxTasks;
private int minTasks;
diff --git
a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
index 441b425d12..7a4ad9a184 100644
--- a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
@@ -56,6 +56,7 @@ parallelNioSender.sender.disconnected.notRetry=Not retrying
send for:[{0}]; Send
parallelNioSender.sender.disconnected.sendFailed=Send failed, and sender is
disconnected. Not retrying.
parallelNioSender.unable.setup.NioSender=Unable to setup NioSender.
+pooledParallelSender.disconnectOnError=Disconnect following error
pooledParallelSender.sender.disconnected=Sender not connected.
pooledParallelSender.unable.open=Unable to open NIO selector.
pooledParallelSender.unable.retrieveSender=Unable to retrieve a sender from
the sender pool
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index 065b7d0ade..8602710505 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -170,7 +170,11 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
log.trace("Adding event to selector:" + event);
}
if (isListening()) {
- selector.wakeup();
+ try {
+ selector.wakeup();
+ } catch (ClosedSelectorException ignore) {
+ // Selector already closed during shutdown
+ }
}
}
}
@@ -298,7 +302,7 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
* @throws IOException IO error
*/
protected void listen() throws Exception {
- if (doListen()) {
+ if (isListening()) {
log.warn(sm.getString("nioReceiver.alreadyStarted"));
return;
}
@@ -313,7 +317,7 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
registerChannel(selector, datagramChannel, SelectionKey.OP_READ,
oreader);
}
- while (doListen() && selector != null) {
+ while (isListening() && selector != null) {
// this may block for a long time, upon return the
// selected set contains keys of the ready channels
try {
diff --git
a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
index 7e4567ee49..fa283949c7 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
@@ -42,11 +42,9 @@ import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
- * A worker thread class which can drain channels and echo-back the input.
Each instance is constructed with a reference
- * to the owning thread pool object. When started, the thread loops forever
waiting to be awakened to service the
- * channel associated with a SelectionKey object. The worker is tasked by
calling its serviceChannel() method with a
- * SelectionKey object. The serviceChannel() method stores the key reference
in the thread object then calls notify() to
- * wake it up. When the channel has been drained, the worker thread returns
itself to its parent pool.
+ * A worker thread class which drains a channel and processes received
messages.
+ * Each instance services a single SelectionKey per invocation. After
completion,
+ * the worker returns itself to its parent pool for reuse.
*/
public class NioReplicationTask extends AbstractRxTask {
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioSender.java
b/java/org/apache/catalina/tribes/transport/nio/NioSender.java
index 3fdbea3aeb..de0e497cf5 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioSender.java
@@ -426,7 +426,11 @@ public class NioSender extends AbstractSender {
*/
public void setMessage(byte[] data, int offset, int length) throws
IOException {
if (data != null) {
- current = data;
+ if (offset == 0 && length == data.length) {
+ current = data;
+ } else {
+ current = Arrays.copyOfRange(data, offset, offset + length);
+ }
remaining = length;
ackbuf.clear();
if (writebuf != null) {
@@ -437,8 +441,6 @@ public class NioSender extends AbstractSender {
if (writebuf.capacity() < length) {
writebuf = getBuffer(length);
}
-
- // TODO use ByteBuffer.wrap to avoid copying the data.
writebuf.put(data, offset, length);
writebuf.flip();
if (isConnected()) {
diff --git
a/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
b/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
index d41b739bb9..36f26b75f6 100644
--- a/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
@@ -24,12 +24,14 @@ import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.transport.DataSender;
import org.apache.catalina.tribes.transport.PooledSender;
import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
/**
* A pooled sender that uses {@link ParallelNioSender} instances for parallel
message delivery.
*/
public class PooledParallelSender extends PooledSender implements
PooledParallelSenderMBean {
- /** StringManager for internationalized log messages. */
+ private static final Log log =
LogFactory.getLog(PooledParallelSender.class);
protected static final StringManager sm =
StringManager.getManager(PooledParallelSender.class);
/**
@@ -61,6 +63,9 @@ public class PooledParallelSender extends PooledSender
implements PooledParallel
sender.sendMessage(destination, message);
sender.keepalive();
} catch (ChannelException x) {
+ if (log.isDebugEnabled()) {
+
log.debug(sm.getString("pooledParallelSender.disconnectOnError"), x);
+ }
sender.disconnect();
throw x;
} finally {
diff --git a/java/org/apache/catalina/tribes/util/Arrays.java
b/java/org/apache/catalina/tribes/util/Arrays.java
index 8d986d0968..e4c4cdf0a8 100644
--- a/java/org/apache/catalina/tribes/util/Arrays.java
+++ b/java/org/apache/catalina/tribes/util/Arrays.java
@@ -54,7 +54,7 @@ public class Arrays {
if (keyoffset < 0 || keyoffset >= key.length) {
throw new
ArrayIndexOutOfBoundsException(sm.getString("arrays.keyoffset.outOfBounds"));
}
- if (length > (key.length - keyoffset)) {
+ if (length < 0 || length > (key.length - keyoffset)) {
throw new
ArrayIndexOutOfBoundsException(sm.getString("arrays.length.outOfBounds"));
}
// we don't have enough data to validate it
@@ -235,17 +235,12 @@ public class Arrays {
* @return {@code true} if the arrays are equal
*/
public static boolean equals(Object[] o1, Object[] o2) {
- boolean result = o1.length == o2.length;
- if (result) {
- for (int i = 0; i < o1.length && result; i++) {
- result = o1[i].equals(o2[i]);
- }
- }
- return result;
+ return java.util.Arrays.equals(o1, o2);
}
/**
- * Checks if two member arrays contain the same members.
+ * Checks if two member arrays contain the same members. Both arrays will
+ * be mutated to be sorted.
*
* @param m1 the first member array
* @param m2 the second member array
@@ -258,7 +253,8 @@ public class Arrays {
}
/**
- * Merges two member arrays, removing duplicates.
+ * Merges two member arrays, removing duplicates. Both arrays will
+ * be mutated to be sorted.
*
* @param m1 the first member array
* @param m2 the second member array
diff --git a/java/org/apache/catalina/tribes/util/ExceptionUtils.java
b/java/org/apache/catalina/tribes/util/ExceptionUtils.java
index a3908b9cba..90346f26ef 100644
--- a/java/org/apache/catalina/tribes/util/ExceptionUtils.java
+++ b/java/org/apache/catalina/tribes/util/ExceptionUtils.java
@@ -24,7 +24,7 @@ public class ExceptionUtils {
/**
* Default constructor. This class is not intended to be instantiated.
*/
- public ExceptionUtils() {
+ private ExceptionUtils() {
// NO-OP
}
diff --git a/java/org/apache/catalina/tribes/util/UUIDGenerator.java
b/java/org/apache/catalina/tribes/util/UUIDGenerator.java
index 95e1019fd2..6f78b57117 100644
--- a/java/org/apache/catalina/tribes/util/UUIDGenerator.java
+++ b/java/org/apache/catalina/tribes/util/UUIDGenerator.java
@@ -91,26 +91,27 @@ public class UUIDGenerator {
* Generates a random UUID and stores it in the provided byte array.
*
* @param secure Whether to use a secure random generator
- * @param into The byte array to store the UUID
+ * @param dest The byte array to store the UUID
* @param offset The offset in the byte array
*
* @return The byte array containing the UUID
*
- * @throws ArrayIndexOutOfBoundsException If the byte array is too small
+ * @throws ArrayIndexOutOfBoundsException If the byte array is too small
or null
*/
- public static byte[] randomUUID(boolean secure, byte[] into, int offset) {
- if ((offset + UUID_LENGTH) > into.length) {
+ public static byte[] randomUUID(boolean secure, byte[] dest, int offset) {
+ int destLength = (dest == null) ? 0 : dest.length;
+ if ((offset + UUID_LENGTH) > destLength) {
throw new ArrayIndexOutOfBoundsException(
sm.getString("uuidGenerator.unable.fit",
Integer.toString(UUID_LENGTH),
- Integer.toString(into.length),
Integer.toString(offset + UUID_LENGTH)));
+ Integer.toString(destLength),
Integer.toString(offset + UUID_LENGTH)));
}
Random r = (secure && (secrand != null)) ? secrand : rand;
- nextBytes(into, offset, UUID_LENGTH, r);
- into[6 + offset] &= 0x0F;
- into[6 + offset] |= (UUID_VERSION << 4);
- into[8 + offset] &= 0x3F; // 0011 1111
- into[8 + offset] |= (byte) 0x80; // 1000 0000
- return into;
+ nextBytes(dest, offset, UUID_LENGTH, r);
+ dest[6 + offset] &= 0x0F;
+ dest[6 + offset] |= (UUID_VERSION << 4);
+ dest[8 + offset] &= 0x3F; // 0011 1111
+ dest[8 + offset] |= (byte) 0x80; // 1000 0000
+ return dest;
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
