true from this method when the router implementation has not been designed to
+ * handle resubmitting such requests can result in a request being resubmitted perpetually.
+ *
+ * @return false by default. true when and only when the router implementation has specific logic to handle
+ * submitting requests which have received a Busy or Unable to Deliver Answer from one peer, to an alternative peer, and to avoid
+ * perpetual re-submission of such requests.
+ */
+ boolean canProcessBusyOrUnableToDeliverAnswer();
+
+
+ /**
+ * Called when a 3002 or 3004 is received for a request. This method attempts to resubmit the request to an alternative peer.
+ *
+ * @param request
+ * @param table
+ */
+ void processBusyOrUnableToDeliverAnswer(IRequest request, IPeerTable table) throws InternalException, RouteException;
+
+
/**
* Based on Redirect entries or any other factors, this method changes route information.
* @param message
@@ -128,6 +152,8 @@ public interface IRouter {
* @throws RouteException
* @throws AvpDataException
*/
+
+
boolean updateRoute(IRequest message) throws RouteException, AvpDataException;
}
diff --git a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/controller/PeerImpl.java b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/controller/PeerImpl.java
index 3415b6cec..0a3506019 100644
--- a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/controller/PeerImpl.java
+++ b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/controller/PeerImpl.java
@@ -383,6 +383,17 @@ private boolean isRedirectAnswer(Avp avpResCode, IMessage answer) {
}
}
+ private boolean isBusyOrUnableToDeliverAnswer(Avp avpResCode, IMessage answer) {
+ try {
+ // E-bit set indicating a protocol error, and Result Code one of 3002 or 3004
+ return (answer.getFlags() & 0x20) != 0 && avpResCode != null
+ && (avpResCode.getInteger32() == ResultCode.TOO_BUSY || avpResCode.getInteger32() == ResultCode.UNABLE_TO_DELIVER);
+ }
+ catch (AvpDataException e) {
+ return false;
+ }
+ }
+
@Override
public IStatistic getStatistic() {
return statistic;
@@ -480,6 +491,24 @@ private IMessage processRedirectAnswer(IMessage request, IMessage answer) {
return answer;
}
+ private IMessage processBusyOrUnableToDeliverAnswer(IMessage request, IMessage answer) {
+ if (router.canProcessBusyOrUnableToDeliverAnswer()) {
+ try {
+ logger.debug("Message with [sessionId={}] received a Busy or Unable to Deliver Answer and will be resubmitted.", request.getSessionId());
+ router.processBusyOrUnableToDeliverAnswer(request, table);
+ return null;
+ }
+ catch (Throwable exc) {
+ // Any error when attempting a resubmit to an alternative peer simply results in the original
+ // Busy or Unable to Deliver Answer being returned
+ if (logger.isErrorEnabled()) {
+ logger.error("Failed to reprocess busy or unable to deliver response - all peers exhausted?", exc);
+ }
+ }
+ }
+ return answer;
+ }
+
@Override
public void connect() throws InternalException, IOException, IllegalDiameterStateException {
if (getState(PeerState.class) != PeerState.DOWN) {
@@ -1034,6 +1063,12 @@ public boolean receiveMessage(IMessage message) {
//if return value is not null, there was some error, lets try to invoke listener if it exists...
isProcessed = message == null;
}
+ avpResCode = message.getAvps().getAvp(RESULT_CODE);
+ if (isBusyOrUnableToDeliverAnswer(avpResCode, message)) {
+ message = processBusyOrUnableToDeliverAnswer(request, message);
+ // if return value is not null, there was some error, lets try to invoke listener if it exists...
+ isProcessed = message == null;
+ }
if (message != null) {
if (request.getEventListener() != null) {
diff --git a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/RouterImpl.java b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/RouterImpl.java
index f2bbfdbce..d526dfd33 100644
--- a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/RouterImpl.java
+++ b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/RouterImpl.java
@@ -507,7 +507,7 @@ public IPeer getPeer(IMessage message, IPeerTable manager) throws RouteException
}
// Balancing
- IPeer peer = selectPeer(availablePeers);
+ IPeer peer = selectPeer(message, availablePeers);
if (peer == null) {
throw new RouteException("Unable to find valid connection to peer[" + destHost + "] in realm[" + destRealm + "]");
}
@@ -639,6 +639,33 @@ public void processRedirectAnswer(IRequest request, IAnswer answer, IPeerTable t
}
}
+ /**
+ * This method should always return false unless specifically designed to handle
+ * submitting requests which have received a Busy or Unable to Deliver Answer from one peer, to an alternative peer, and to avoid
+ * perpetual re-submission of such requests.
+ *
+ * @return false
+ */
+ @Override
+ public boolean canProcessBusyOrUnableToDeliverAnswer() {
+ return false;
+ }
+
+ public void processBusyOrUnableToDeliverAnswer(IRequest request, IPeerTable table) throws InternalException, RouteException {
+ try {
+ table.sendMessage((IMessage) request);
+ }
+ catch (AvpDataException exc) {
+ throw new InternalException(exc);
+ }
+ catch (IllegalDiameterStateException e) {
+ throw new InternalException(e);
+ }
+ catch (IOException e) {
+ throw new InternalException(e);
+ }
+ }
+
/**
*
*/
@@ -797,6 +824,10 @@ protected IPeer selectPeer(List
* {@code
* for (m = 0; m < n; m++) {
- * if (W(Sm) > 0) {
- * for (i = m+1; i < n; i++) {
- * if (C(Sm)*W(Si) > C(Si)*W(Sm))
- * m = i;
+ * if (W(Sm) > 0) {
+ * for (i = m+1; i < n; i++) {
+ * if (C(Sm)*W(Si) > C(Si)*W(Sm))
+ * m = i;
+ * }
+ * return Sm;
* }
- * return Sm;
- * }
* }
* return NULL;
* }
@@ -125,6 +126,19 @@ public WeightedLeastConnectionsRouter(IContainer container, IConcurrentFactory c
*/
@Override
public IPeer selectPeer(List availablePeers) {
+ return selectPeer(null, availablePeers);
+ }
+
+ /**
+ * Return peer with least connections
+ *
+ * @param message the message to be sent
+ * @param availablePeers list of peers that are in {@link PeerState#OKAY OKAY} state
+ * @return the selected peer according to algorithm
+ *
+ */
+ @Override
+ public IPeer selectPeer(IMessage message, List availablePeers) {
int peerSize = availablePeers != null ? availablePeers.size() : 0;
// Return none if empty, or first if only one member found
diff --git a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinResubmittingRouter.java b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinResubmittingRouter.java
new file mode 100644
index 000000000..52d2a07c3
--- /dev/null
+++ b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinResubmittingRouter.java
@@ -0,0 +1,393 @@
+/*
+ * TeleStax, Open Source Cloud Communications
+ * Copyright 2011-2016, Telestax Inc and individual contributors
+ * by the @authors tag.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation; either version 3 of
+ * the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see > attemptedPeers = new ConcurrentHashMap>();
+
+ protected WeightedRoundRobinResubmittingRouter(IRealmTable table, Configuration config) {
+ super(null, null, table, config, null);
+ }
+
+ public WeightedRoundRobinResubmittingRouter(IContainer container, IConcurrentFactory concurrentFactory, IRealmTable realmTable, Configuration config,
+ MetaData aMetaData) {
+ super(container, concurrentFactory, realmTable, config, aMetaData);
+ }
+
+ /**
+ * The WeightedRoundRobinResubmittingRouter is specifically designed to handle submitting requests
+ * which have received a Busy or Unable to Deliver Answer from one peer, to an alternative peer,
+ * and to avoid perpetual re-submission of such requests.
+ *
+ * @return true
+ */
+ @Override
+ public boolean canProcessBusyOrUnableToDeliverAnswer() {
+ return true;
+ }
+
+ /**
+ * Select peer by weighted round-robin scheduling
+ * As documented in http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
+ *
+ *
+ * The weighted round-robin scheduling is designed to better handle servers
+ * with different processing capacities. Each server can be assigned a weight,
+ * an integer value that indicates the processing capacity. Servers with higher
+ * weights receive new connections first than those with less weights, and servers
+ * with higher weights get more connections than those with less weights and servers
+ * with equal weights get equal connections. The pseudo code of weighted round-robin
+ * scheduling is as follows:
+ *
+ * Supposing that there is a server set S = {S0, S1, …, Sn-1};
+ * W(Si) indicates the weight of Si;
+ * i indicates the server selected last time, and i is initialized with -1;
+ * cw is the current weight in scheduling, and cw is initialized with zero;
+ * max(S) is the maximum weight of all the servers in S;
+ * gcd(S) is the greatest common divisor of all server weights in S;
+ *
+ *
+ *
+ * {@code
+ * while (true) {
+ * i = (i + 1) mod n;
+ * if (i == 0) {
+ * cw = cw - gcd(S);
+ * if (cw <= 0) {
+ * cw = max(S);
+ * if (cw == 0)
+ * return NULL;
+ * }
+ * }
+ * if (W(Si) >= cw)
+ * return Si;
+ * }
+ * }
+ *
+ *
+ * For example, the real servers, A, B and C, have the weights, 4, 3, 2 respectively,
+ * a scheduling sequence will be AABABCABC in a scheduling period (mod sum(Wi)).
+ *
+ * In an optimized implementation of the weighted round-robin scheduling, a scheduling sequence
+ * will be generated according to the server weights after the rules of IPVS are modified.
+ * The network connections are directed to the different real servers based on the scheduling
+ * sequence in a round-robin manner.
+ *
+ * The weighted round-robin scheduling is better than the round-robin scheduling, when the
+ * processing capacity of real servers are different. However, it may lead to dynamic load
+ * imbalance among the real servers if the load of the requests vary highly. In short, there
+ * is the possibility that a majority of requests requiring large responses may be directed
+ * to the same real server.
+ *
+ * Actually, the round-robin scheduling is a special instance of the weighted round-robin
+ * scheduling, in which all the weights are equal.
+ *
+ * This method is internally synchronized due to concurrent modifications to lastSelectedPeer and currentWeight.
+ * Please consider this when relying on heavy throughput.
+ *
+ * Please note: if the list of availablePeers changes between calls (e.g. if a peer becomes active or inactive),
+ * the balancing algorithm is disturbed and might be distributed uneven.
+ * This is likely to happen if peers are flapping.
+ *
+ * @param availablePeers
+ * list of peers that are in {@link PeerState#OKAY OKAY} state
+ * @return the selected peer according to algorithm
+ * @see
+ * http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
+ */
+ @Override
+ public IPeer selectPeer(List availablePeers) {
+ return selectPeer(null, availablePeers);
+ }
+
+ /**
+ * Select peer by weighted round-robin scheduling
+ *
+ * This method ensures that, when the message is passed, that
+ * the same peer that responded with the Busy or Unable To Deliver Answer is not selected for
+ * any subsequent submissions of the same request.
+ *
+ * @param message
+ * The message to be re-attempted due to a Busy or Unable To Deliver Answer
+ * @param availablePeers
+ * list of peers that are in {@link PeerState#OKAY OKAY} state
+ * @return the selected peer according to algorithm, ensuring that if the message is passed, that
+ * the same peer that responded with the Busy or Unable To Deliver Answer is not selected a second time
+ *
+ */
+ @Override
+ public IPeer selectPeer(IMessage message, List availablePeers) {
+ IPeer selectedPeer = null;
+ int peerSize = availablePeers != null ? availablePeers.size() : 0;
+
+ // Return none if empty, or first if only one member found
+ if (peerSize <= 0) {
+ return null;
+ }
+
+ if (message != null && message.getPeer() != null) {
+ addAttemptedPeer(message, message.getPeer());
+
+ long numberOfAttempts = 0;
+ MessageKey messageKey = getMessageKey(message);
+ Set peerSet = attemptedPeers.get(messageKey);
+ if (peerSet != null) {
+ numberOfAttempts = peerSet.size();
+ }
+ if (logger.isTraceEnabled()) {
+ logger.trace("Selecting subsequent peer for {} [numberOfAttempts={}]", messageKey, numberOfAttempts);
+ }
+
+ if (numberOfAttempts == peerSize) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("All peers exhausted for {}, giving up...", messageKey);
+ }
+ removeAttemptedPeers(messageKey);
+ return null;
+ }
+ }
+
+ if (peerSize == 1) {
+ return availablePeers.iterator().next();
+ }
+
+ // Find maximum weight and greatest common divisor of weight across all peers
+ int maxWeight = 0;
+ Integer gcd = null;
+ for (IPeer peer : availablePeers) {
+ maxWeight = Math.max(maxWeight, peer.getRating());
+ gcd = (gcd == null) ? peer.getRating() : gcd(gcd, peer.getRating());
+ }
+
+ // Find best matching candidate. Synchronized here due to consistent changes on member variables
+ synchronized (this) {
+ for (;;) {
+ lastSelectedPeer = (lastSelectedPeer + 1) % peerSize;
+ if (lastSelectedPeer == 0) {
+ currentWeight = currentWeight - gcd;
+ if (currentWeight <= 0) {
+ currentWeight = maxWeight;
+ }
+ }
+ if (peerSize <= lastSelectedPeer) {
+ lastSelectedPeer = -1; // safety first, restart if peer size has accidentally changed.
+ continue;
+ }
+ IPeer candidate = availablePeers.get(lastSelectedPeer);
+ if (candidate.getRating() >= currentWeight) {
+ if (message != null && message.getPeer() != null) {
+ if (isPeerPreviouslyAttempted(lastSelectedPeer, availablePeers, message)) {
+ continue;
+ }
+ }
+
+ selectedPeer = availablePeers.get(lastSelectedPeer);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Selected Peer [uri={}, realmName={}, rating={}]", selectedPeer.getUri(), selectedPeer.getRealmName(), selectedPeer.getRating());
+ }
+ return selectedPeer;
+ }
+ }
+ }
+ }
+
+ private MessageKey getMessageKey(final IMessage message) {
+ return new MessageKey(message);
+ }
+
+ private boolean isPeerPreviouslyAttempted(int selectedPeerIndex, List availablePeers, IMessage message) {
+ boolean isPeerPreviouslyAttempted = false;
+ final MessageKey messageKey = getMessageKey(message);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Checking whether selected Peer [id={}] has already had {} sent to it", selectedPeerIndex, messageKey);
+ }
+
+ if (attemptedPeers.containsKey(messageKey)) {
+ IPeer candidate = availablePeers.get(selectedPeerIndex);
+ if (attemptedPeers.get(messageKey).contains(candidate)) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Peer [uri={}, realmName={}, rating={}] has been tried before, try next peer", candidate.getUri(), candidate.getRealmName(),
+ candidate.getRating());
+ }
+ isPeerPreviouslyAttempted = true;
+ }
+ }
+
+ return isPeerPreviouslyAttempted;
+ }
+
+ private synchronized void addAttemptedPeer(final IMessage message, IPeer peer) {
+ final MessageKey messageKey = getMessageKey(message);
+ if (attemptedPeers.containsKey(messageKey)) {
+ attemptedPeers.get(messageKey).add(peer);
+ }
+ else {
+ Set peerSet = new HashSet();
+ peerSet.add(peer);
+ attemptedPeers.put(messageKey, peerSet);
+
+ new Timer().schedule(new TimerTask() {
+ @Override
+ public void run() {
+ removeAttemptedPeers(messageKey);
+ }
+ }, ATTEMPTED_PEER_RETENTION_PERIOD_MS);
+ }
+ }
+
+ private void removeAttemptedPeers(MessageKey messageKey) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Removing attemptedPeers for {} (currently [attemptedPeers.size()={}]) ", messageKey, attemptedPeers.size());
+ }
+ Set peerSet = attemptedPeers.remove(messageKey);
+ if (peerSet != null) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("peerSet with [size={}] has been removed for {}", peerSet.size(), messageKey);
+ }
+ }
+ else {
+ if (logger.isWarnEnabled()) {
+ logger.warn("No peers removed from attemptedPeers for {}!", messageKey);
+ }
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("Done removing attemptedPeers for {} (now [attemptedPeers.size()={}]) ", messageKey, attemptedPeers.size());
+ }
+ }
+
+ /**
+ * Return greatest common divisor for two integers
+ * https://en.wikipedia.org/wiki/Greatest_common_divisor#Using_Euclid.27s_algorithm
+ *
+ * @param a
+ * @param b
+ * @return greatest common divisor
+ */
+ protected int gcd(int a, int b) {
+ return (b == 0) ? a : gcd(b, a % b);
+ }
+
+ /**
+ * Defines a class which can be used to uniquely define any single message within any given session.
+ *
+ */
+ private class MessageKey {
+ private String sessionId;
+ private long endToEndId;
+ private int flags;
+
+ MessageKey(IMessage message) {
+ super();
+ this.sessionId = message.getSessionId();
+ this.endToEndId = message.getEndToEndIdentifier();
+ this.flags = message.getFlags();
+ }
+
+ @Override
+ public String toString() {
+ return "MessageKey [sessionId=" + sessionId + ", endToEndId=" + endToEndId + ", flags=" + flags + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + getOuterType().hashCode();
+ result = prime * result + (int) (endToEndId ^ (endToEndId >>> 32));
+ result = prime * result + flags;
+ result = prime * result + ((sessionId == null) ? 0 : sessionId.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ MessageKey other = (MessageKey) obj;
+ if (!getOuterType().equals(other.getOuterType())) {
+ return false;
+ }
+ if (endToEndId != other.endToEndId) {
+ return false;
+ }
+ if (flags != other.flags) {
+ return false;
+ }
+ if (sessionId == null) {
+ if (other.sessionId != null) {
+ return false;
+ }
+ }
+ else if (!sessionId.equals(other.sessionId)) {
+ return false;
+ }
+ return true;
+ }
+
+ private WeightedRoundRobinResubmittingRouter getOuterType() {
+ return WeightedRoundRobinResubmittingRouter.this;
+ }
+ }
+}
diff --git a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinRouter.java b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinRouter.java
index f72954ed4..604154447 100644
--- a/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinRouter.java
+++ b/core/jdiameter/impl/src/main/java/org/jdiameter/client/impl/router/WeightedRoundRobinRouter.java
@@ -19,25 +19,26 @@
package org.jdiameter.client.impl.router;
+import java.util.List;
+
import org.jdiameter.api.Configuration;
import org.jdiameter.api.MetaData;
import org.jdiameter.api.PeerState;
import org.jdiameter.client.api.IContainer;
+import org.jdiameter.client.api.IMessage;
import org.jdiameter.client.api.controller.IPeer;
import org.jdiameter.client.api.controller.IRealmTable;
import org.jdiameter.common.api.concurrent.IConcurrentFactory;
-
import org.jdiameter.server.api.IRouter;
-import java.util.List;
-
/**
* Weighted round-robin router implementation
*
- * @see http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
* @author Nils Sowen
+ * @see
+ * http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
*/
-public class WeightedRoundRobinRouter extends RouterImpl implements IRouter{
+public class WeightedRoundRobinRouter extends RouterImpl implements IRouter {
private int lastSelectedPeer = -1;
private int currentWeight = 0;
@@ -74,17 +75,17 @@ public WeightedRoundRobinRouter(IContainer container, IConcurrentFactory concurr
*
* {@code
* while (true) {
- * i = (i + 1) mod n;
- * if (i == 0) {
- * cw = cw - gcd(S);
- * if (cw <= 0) {
- * cw = max(S);
- * if (cw == 0)
- * return NULL;
+ * i = (i + 1) mod n;
+ * if (i == 0) {
+ * cw = cw - gcd(S);
+ * if (cw <= 0) {
+ * cw = max(S);
+ * if (cw == 0)
+ * return NULL;
+ * }
* }
- * }
- * if (W(Si) >= cw)
- * return Si;
+ * if (W(Si) >= cw)
+ * return Si;
* }
* }
*
@@ -108,18 +109,31 @@ public WeightedRoundRobinRouter(IContainer container, IConcurrentFactory concurr
*
* This method is internally synchronized due to concurrent modifications to lastSelectedPeer and currentWeight.
* Please consider this when relying on heavy throughput.
- *
+ *
* Please note: if the list of availablePeers changes between calls (e.g. if a peer becomes active or inactive),
* the balancing algorithm is disturbed and might be distributed uneven.
* This is likely to happen if peers are flapping.
*
* @param availablePeers list of peers that are in {@link PeerState#OKAY OKAY} state
* @return the selected peer according to algorithm
- * @see http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
+ * @see
+ * http://kb.linuxvirtualserver.org/wiki/Weighted_Round-Robin_Scheduling
*/
@Override
public IPeer selectPeer(List availablePeers) {
+ return selectPeer(null, availablePeers);
+ }
+ /**
+ * Select peer by weighted round-robin scheduling
+ *
+ * @param message the message to be sent
+ * @param availablePeers list of peers that are in {@link PeerState#OKAY OKAY} state
+ * @return the selected peer according to algorithm
+ *
+ */
+ @Override
+ public IPeer selectPeer(IMessage message, List availablePeers) {
int peerSize = availablePeers != null ? availablePeers.size() : 0;
// Return none if empty, or first if only one member found
@@ -140,7 +154,7 @@ public IPeer selectPeer(List availablePeers) {
// Find best matching candidate. Synchronized here due to consistent changes on member variables
synchronized (this) {
- for ( ;; ) {
+ for (; ; ) {
lastSelectedPeer = (lastSelectedPeer + 1) % peerSize;
if (lastSelectedPeer == 0) {
currentWeight = currentWeight - gcd;
diff --git a/core/jdiameter/impl/src/test/java/org/jdiameter/client/impl/router/TestRouter.java b/core/jdiameter/impl/src/test/java/org/jdiameter/client/impl/router/TestRouter.java
index 15be520ac..e325e210f 100644
--- a/core/jdiameter/impl/src/test/java/org/jdiameter/client/impl/router/TestRouter.java
+++ b/core/jdiameter/impl/src/test/java/org/jdiameter/client/impl/router/TestRouter.java
@@ -19,7 +19,26 @@
package org.jdiameter.client.impl.router;
-import org.jdiameter.api.*;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.URISyntaxException;
+import java.net.UnknownServiceException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+import org.jdiameter.api.ApplicationId;
+import org.jdiameter.api.Avp;
+import org.jdiameter.api.Configuration;
+import org.jdiameter.api.IllegalDiameterStateException;
+import org.jdiameter.api.InternalException;
+import org.jdiameter.api.LocalAction;
+import org.jdiameter.api.OverloadException;
+import org.jdiameter.api.PeerStateListener;
+import org.jdiameter.api.Realm;
+import org.jdiameter.api.Statistic;
+import org.jdiameter.api.URI;
import org.jdiameter.api.app.StateChangeListener;
import org.jdiameter.client.api.IAnswer;
import org.jdiameter.client.api.IMessage;
@@ -29,426 +48,688 @@
import org.jdiameter.client.api.fsm.EventTypes;
import org.jdiameter.client.api.io.IConnectionListener;
import org.jdiameter.client.api.io.TransportException;
+import org.jdiameter.client.impl.helpers.UIDGenerator;
import org.jdiameter.client.impl.helpers.XMLConfiguration;
+import org.jdiameter.client.impl.parser.MessageParser;
import org.jdiameter.common.api.statistic.IStatistic;
import org.jdiameter.common.api.statistic.IStatisticManager;
import org.jdiameter.common.api.statistic.IStatisticRecord;
import org.jdiameter.common.impl.controller.AbstractPeer;
import org.jdiameter.common.impl.statistic.StatisticManagerImpl;
import org.jdiameter.server.api.agent.IAgentConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.URISyntaxException;
-import java.net.UnknownServiceException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Set;
-
-import static org.testng.AssertJUnit.assertEquals;
+import junit.framework.TestCase;
/**
* Various testcases for Router implementations
*
* @author Nils Sowen
*/
-public class TestRouter {
-
- @Test
- public void testWeightedRoundRobin() throws Exception {
-
- Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobin-config.xml");
- WeightedRoundRobinRouter router = new WeightedRoundRobinRouter(new RealmTableTest(), config);
-
- IStatisticManager manager = new StatisticManagerImpl(config);
- PeerTest p1 = new PeerTest(1, 1, true, manager);
- PeerTest p2 = new PeerTest(2, 1, true, manager);
- PeerTest p3 = new PeerTest(3, 1, true, manager);
- PeerTest p4 = new PeerTest(4, 1, true, manager);
-
- List peers = new ArrayList(3);
- peers.add(p1);
- peers.add(p2);
- peers.add(p3);
-
- // Test simple round robin (all weight = 1)
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
-
- // Test weighted round robin (p1=2, p2=1, p3=1)
- p1.setRating(2);
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
-
- // Test weighted round robin (p1=2, p2=2, p3=1)
- p2.setRating(2);
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
-
- // Test equally weighted round robin (p1=2, p2=2, p3=2)
- p3.setRating(2);
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
-
- // Add Peer-4 with weight 1 to list
- peers.add(p4);
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- // expected glitch here: due to the sudden availibity of Peer-4, the algorithm is disturbed
- assertEquals(p4.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p4.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
- assertEquals(p4.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p3.toString(), router.selectPeer(peers).toString());
-
- // Next cycle would produce Peer-4, but reduce peer list now
- peers = peers.subList(0,2); // now: Peer-1, Peer-2
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- }
-
- @Test
- public void testWeightedLeastConnections() throws Exception {
-
- Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedleastconnections-config.xml");
- WeightedLeastConnectionsRouter router = new WeightedLeastConnectionsRouter(new RealmTableTest(), config);
-
- IStatisticManager manager = new StatisticManagerImpl(config);
- PeerTest p1 = new PeerTest(1, 1, true, manager);
- PeerTest p2 = new PeerTest(2, 1, true, manager);
- PeerTest p3 = new PeerTest(3, 1, true, manager);
-
- List peers = new ArrayList(2);
- peers.add(p1);
- peers.add(p2);
-
- // Test simple round robin (all weight = 1)
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- // increase p1 requests/s by 1
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
-
- // increase p2 requests/s by 1
- p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name()+'.'+p2.getUri()).inc();
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- // decrease p1 requests/s by 1
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenResponsePerSecond.name()+'.'+p1.getUri()).dec();
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- // increase p1 requests/s by 3
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
-
- // increase p2 requests/s by 1
- p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name()+'.'+p2.getUri()).inc();
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
-
- // increase weight of p1
- p1.setRating(2);
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- // decrease p1 requests/s by 1
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).dec();
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- // increase p1 requests/s by 2
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name()+'.'+p1.getUri()).inc();
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
-
- // increase weight and requests/s of p2
- p2.setRating(2);
- p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name()+'.'+p2.getUri()).inc();
- p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name()+'.'+p2.getUri()).inc();
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
- assertEquals(p2.toString(), router.selectPeer(peers).toString());
-
- // increase p2 requests/s by 1
- p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name()+'.'+p2.getUri()).inc();
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
- assertEquals(p1.toString(), router.selectPeer(peers).toString());
-
- }
-
- private static class RealmTableTest implements IRealmTable {
-
- public Realm matchRealm(IRequest request) {
- return null;
- }
+public class TestRouter extends TestCase {
+ private static final Logger logger = LoggerFactory.getLogger(TestRouter.class);
+ private static UIDGenerator uid = new UIDGenerator();
+
+ @Test
+ public void testWeightedRoundRobin() throws Exception {
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobin-config.xml");
+ WeightedRoundRobinRouter router = new WeightedRoundRobinRouter(new RealmTableTest(), config);
+
+ assertFalse(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 1, true, manager);
+ PeerTest p3 = new PeerTest(3, 1, true, manager);
+ PeerTest p4 = new PeerTest(4, 1, true, manager);
+
+ List peers = new ArrayList(3);
+ peers.add(p1);
+ peers.add(p2);
+ peers.add(p3);
+
+ // Test simple round robin (all weight = 1)
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test weighted round robin (p1=2, p2=1, p3=1)
+ p1.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test weighted round robin (p1=2, p2=2, p3=1)
+ p2.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test equally weighted round robin (p1=2, p2=2, p3=2)
+ p3.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Add Peer-4 with weight 1 to list
+ peers.add(p4);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ // expected glitch here: due to the sudden availibity of Peer-4, the algorithm is disturbed
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Next cycle would produce Peer-4, but reduce peer list now
+ peers = peers.subList(0, 2); // now: Peer-1, Peer-2
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ }
+
+ /*
+ * WeightedRoundRobinResubmittingRouter should behave exactly as the WeightedRoundRobinRouter
+ * when no Busy or Unable to Deliver Answers are received and hence no re-submissions to alternative peers
+ * are required.
+ */
+ @Test
+ public void testWeightedRoundRobinResubmittingNoResubmissions() throws Exception {
+ logger.debug("*** Executing testWeightedRoundRobinResubmittingNoResubmissions ***");
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml");
+ WeightedRoundRobinResubmittingRouter router = new WeightedRoundRobinResubmittingRouter(new RealmTableTest(), config);
+
+ assertTrue(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 1, true, manager);
+ PeerTest p3 = new PeerTest(3, 1, true, manager);
+ PeerTest p4 = new PeerTest(4, 1, true, manager);
+
+ List peers = new ArrayList(3);
+ peers.add(p1);
+ peers.add(p2);
+ peers.add(p3);
+
+ // Test simple round robin (all weight = 1)
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test weighted round robin (p1=2, p2=1, p3=1)
+ p1.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test weighted round robin (p1=2, p2=2, p3=1)
+ p2.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Test equally weighted round robin (p1=2, p2=2, p3=2)
+ p3.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Add Peer-4 with weight 1 to list
+ peers.add(p4);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ // expected glitch here: due to the sudden availibity of Peer-4, the algorithm is disturbed
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+
+ // Next cycle would produce Peer-4, but reduce peer list now
+ peers = peers.subList(0, 2); // now: Peer-1, Peer-2
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ logger.debug("*** Execution of testWeightedRoundRobinResubmittingNoResubmissions completed ***");
+ }
+
+ /*
+ * Validates that, when a peer responds with a Busy or Unable to Deliver Answer, an alternative peer is
+ * selected to resubmit the Request to, based on the existing Round Robin weighting algorithm.
+ */
+ @Test
+ public void testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswer() throws Exception {
+ logger.debug("*** Executing testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswer ***");
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml");
+ WeightedRoundRobinResubmittingRouter router = new WeightedRoundRobinResubmittingRouter(new RealmTableTest(), config);
+
+ assertTrue(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 2, true, manager);
+ PeerTest p3 = new PeerTest(3, 3, true, manager);
+ PeerTest p4 = new PeerTest(4, 4, true, manager);
+
+ List peers = new ArrayList(3);
+ peers.add(p1);
+ peers.add(p2);
+ peers.add(p3);
+ peers.add(p4);
+
+ // Create any message
+ MessageParser messageParser = new MessageParser();
+ IMessage request = messageParser.createEmptyMessage(123, 3l);
+ request.setRequest(true);
+ request.getAvps().addAvp(Avp.SESSION_ID, getSessionId(), true, false, false);
+
+ // Test weighted round robin on a single, resubmitted request
+ assertEquals(p4.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p4);
+ assertEquals(p3.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p3);
+ assertEquals(p2.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p2);
+ assertEquals(p1.toString(), router.selectPeer(request, peers).toString());
+
+ logger.debug("*** Execution of testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswer completed ***");
+ }
+
+ /*
+ * Validates that, when all peers have been tried and each has responded with with a Busy or Unable to Deliver Answer,
+ * the router gives up trying to assign a peer.
+ */
+ @Test
+ public void testWeightedRoundRobinResubmittingPeersExhaused() throws Exception {
+ logger.debug("*** Executing testWeightedRoundRobinResubmittingPeersExhaused ***");
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml");
+ WeightedRoundRobinResubmittingRouter router = new WeightedRoundRobinResubmittingRouter(new RealmTableTest(), config);
+
+ assertTrue(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 2, true, manager);
+ PeerTest p3 = new PeerTest(3, 3, true, manager);
+ PeerTest p4 = new PeerTest(4, 4, true, manager);
+
+ List peers = new ArrayList(3);
+ peers.add(p1);
+ peers.add(p2);
+ peers.add(p3);
+ peers.add(p4);
+
+ // Create any message
+ MessageParser messageParser = new MessageParser();
+ IMessage request = messageParser.createEmptyMessage(123, 3l);
+ request.setRequest(true);
+ request.getAvps().addAvp(Avp.SESSION_ID, getSessionId(), true, false, false);
+
+ // Test weighted round robin on a single, resubmitted request
+ assertEquals(p4.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p4);
+ assertEquals(p3.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p3);
+ assertEquals(p2.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p2);
+ assertEquals(p1.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p1);
+ assertNull(router.selectPeer(request, peers));
+
+ logger.debug("*** Execution of testWeightedRoundRobinResubmittingPeersExhaused completed ***");
+ }
+
+ /*
+ * Validates that, when a peer responds with a Busy or Unable to Deliver Answer, an alternative peer is
+ * selected to resubmit the Request to, based on the existing Round Robin weighting algorithm, also after
+ * previous unrelated requests have moved the state of the round robin algorithm onwards from the initial state
+ * prior to submitting the (to be resubmitted) request.
+ */
+ @Test
+ public void testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswerAfterPreviousMessages() throws Exception {
+ logger.debug("*** Executing testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswerAfterPreviousMessages ***");
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml");
+ WeightedRoundRobinResubmittingRouter router = new WeightedRoundRobinResubmittingRouter(new RealmTableTest(), config);
+
+ assertTrue(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 2, true, manager);
+ PeerTest p3 = new PeerTest(3, 3, true, manager);
+ PeerTest p4 = new PeerTest(4, 4, true, manager);
+
+ List peers = new ArrayList(3);
+ peers.add(p1);
+ peers.add(p2);
+ peers.add(p3);
+ peers.add(p4);
+
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p3.toString(), router.selectPeer(peers).toString());
+ assertEquals(p4.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // Create any message
+ MessageParser messageParser = new MessageParser();
+ IMessage request = messageParser.createEmptyMessage(123, 3l);
+ request.setRequest(true);
+ request.getAvps().addAvp(Avp.SESSION_ID, getSessionId(), true, false, false);
+
+ // Test weighted round robin on a single, resubmitted request
+ assertEquals(p2.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p2);
+ assertEquals(p3.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p3);
+ assertEquals(p4.toString(), router.selectPeer(request, peers).toString());
+ request.setPeer(p4);
+ assertEquals(p1.toString(), router.selectPeer(request, peers).toString());
+
+ logger.debug("*** Execution of testWeightedRoundRobinResubmittingOnBusyOrUnableToDeliverAnswerAfterPreviousMessages completed ***");
+ }
+
+ private String getSessionId() {
+ long id = uid.nextLong();
+ long high32 = (id & 0xffffffff00000000L) >> 32;
+ long low32 = (id & 0xffffffffL);
+ StringBuilder sb = new StringBuilder();
+ sb.append("localhost").append(";").append(high32).append(";").append(low32);
+ return sb.toString();
+ }
+
+ @Test
+ public void testWeightedLeastConnections() throws Exception {
+
+ Configuration config = new XMLConfiguration("src/test/resources/jdiameter-weightedleastconnections-config.xml");
+ WeightedLeastConnectionsRouter router = new WeightedLeastConnectionsRouter(new RealmTableTest(), config);
+
+ assertFalse(router.canProcessBusyOrUnableToDeliverAnswer());
+
+ IStatisticManager manager = new StatisticManagerImpl(config);
+ PeerTest p1 = new PeerTest(1, 1, true, manager);
+ PeerTest p2 = new PeerTest(2, 1, true, manager);
+ PeerTest p3 = new PeerTest(3, 1, true, manager);
+
+ List peers = new ArrayList(2);
+ peers.add(p1);
+ peers.add(p2);
+
+ // Test simple round robin (all weight = 1)
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // increase p1 requests/s by 1
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+
+ // increase p2 requests/s by 1
+ p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name() + '.' + p2.getUri()).inc();
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // decrease p1 requests/s by 1
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenResponsePerSecond.name() + '.' + p1.getUri()).dec();
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // increase p1 requests/s by 3
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+
+ // increase p2 requests/s by 1
+ p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name() + '.' + p2.getUri()).inc();
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+
+ // increase weight of p1
+ p1.setRating(2);
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // decrease p1 requests/s by 1
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).dec();
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ // increase p1 requests/s by 2
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ p1.getStatistic().getRecordByName(IStatisticRecord.Counters.AppGenRequestPerSecond.name() + '.' + p1.getUri()).inc();
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+
+ // increase weight and requests/s of p2
+ p2.setRating(2);
+ p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name() + '.' + p2.getUri()).inc();
+ p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name() + '.' + p2.getUri()).inc();
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+ assertEquals(p2.toString(), router.selectPeer(peers).toString());
+
+ // increase p2 requests/s by 1
+ p2.getStatistic().getRecordByName(IStatisticRecord.Counters.NetGenRequestPerSecond.name() + '.' + p2.getUri()).inc();
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+ assertEquals(p1.toString(), router.selectPeer(peers).toString());
+
+ }
+
+ private static class RealmTableTest implements IRealmTable {
+
+ public Realm matchRealm(IRequest request) {
+ return null;
+ }
- public Realm matchRealm(IAnswer message, String destRealm) {
- return null;
- }
+ public Realm matchRealm(IAnswer message, String destRealm) {
+ return null;
+ }
- public Realm getRealm(String realmName, ApplicationId applicationId) {
- return null;
- }
+ public Realm getRealm(String realmName, ApplicationId applicationId) {
+ return null;
+ }
- public Realm removeRealmApplicationId(String realmName, ApplicationId appId) {
- return null;
- }
+ public Realm removeRealmApplicationId(String realmName, ApplicationId appId) {
+ return null;
+ }
- public Collection removeRealm(String realmName) {
- return null;
- }
+ public Collection removeRealm(String realmName) {
+ return null;
+ }
- public Collection getRealms(String realm) {
- return null;
- }
+ public Collection getRealms(String realm) {
+ return null;
+ }
- public Collection getRealms() {
- return null;
- }
+ public Collection getRealms() {
+ return null;
+ }
- public String getRealmForPeer(String fqdn) {
- return null;
- }
+ public String getRealmForPeer(String fqdn) {
+ return null;
+ }
- public void addLocalApplicationId(ApplicationId ap) {
+ public void addLocalApplicationId(ApplicationId ap) {
- }
+ }
- public void removeLocalApplicationId(ApplicationId a) {
+ public void removeLocalApplicationId(ApplicationId a) {
- }
+ }
- public void addLocalRealm(String localRealm, String fqdn) {
+ public void addLocalRealm(String localRealm, String fqdn) {
- }
+ }
- public Realm addRealm(String name, ApplicationId appId, LocalAction locAction, IAgentConfiguration agentConfImpl, boolean isDynamic, long expirationTime, String[] hosts) throws InternalException {
- return null;
- }
+ public Realm addRealm(String name, ApplicationId appId, LocalAction locAction, IAgentConfiguration agentConfImpl, boolean isDynamic, long expirationTime,
+ String[] hosts) throws InternalException {
+ return null;
+ }
- public Statistic getStatistic(String realmName) {
- return null;
- }
+ public Statistic getStatistic(String realmName) {
+ return null;
+ }
- public Realm addRealm(String realmName, ApplicationId applicationId, LocalAction action, String agentConfiguration, boolean dynamic, long expirationTime, String[] hosts) throws InternalException {
- return null;
- }
+ public Realm addRealm(String realmName, ApplicationId applicationId, LocalAction action, String agentConfiguration, boolean dynamic, long expirationTime,
+ String[] hosts) throws InternalException {
+ return null;
+ }
- public boolean realmExists(String realmName) {
- return false;
- }
+ public boolean realmExists(String realmName) {
+ return false;
+ }
- public boolean isWrapperFor(Class iface) throws InternalException {
- return false;
- }
+ public boolean isWrapperFor(Class iface) throws InternalException {
+ return false;
+ }
+
+ public T unwrap(Class iface) throws InternalException {
+ return null;
+ }
- public T unwrap(Class iface) throws InternalException {
- return null;
- }
-
- public List getAllRealmSet(){
- return null;
- }
+ public List getAllRealmSet() {
+ return null;
}
+ }
- private static class PeerTest extends AbstractPeer implements IPeer {
+ private static class PeerTest extends AbstractPeer implements IPeer {
- private int id;
- private int rating;
- private boolean connected;
+ private int id;
+ private int rating;
+ private boolean connected;
- public PeerTest(int id, int rating, boolean connected, IStatisticManager manager) throws URISyntaxException, UnknownServiceException {
- super(new URI("aaa://"+id), manager);
- this.id = id;
- this.rating = rating;
- this.connected = connected;
- createPeerStatistics();
- }
+ public PeerTest(int id, int rating, boolean connected, IStatisticManager manager) throws URISyntaxException, UnknownServiceException {
+ super(new URI("aaa://" + id), manager);
+ this.id = id;
+ this.rating = rating;
+ this.connected = connected;
+ createPeerStatistics();
+ }
- public void setRating(int rating) {
- this.rating = rating;
- }
+ public void setRating(int rating) {
+ this.rating = rating;
+ }
- public int getRating() {
- return rating;
- }
+ public int getRating() {
+ return rating;
+ }
- public long getHopByHopIdentifier() {
- return 0;
- }
+ public long getHopByHopIdentifier() {
+ return 0;
+ }
- public void addMessage(IMessage message) {
+ public void addMessage(IMessage message) {
- }
+ }
- public void remMessage(IMessage message) {
+ public void remMessage(IMessage message) {
- }
+ }
- public IMessage[] remAllMessage() {
- return new IMessage[0];
- }
+ public IMessage[] remAllMessage() {
+ return new IMessage[0];
+ }
- public boolean handleMessage(EventTypes type, IMessage message, String key) throws TransportException, OverloadException, InternalException {
- return false;
- }
+ public boolean handleMessage(EventTypes type, IMessage message, String key) throws TransportException, OverloadException, InternalException {
+ return false;
+ }
- public boolean sendMessage(IMessage message) throws TransportException, OverloadException, InternalException {
- return false;
- }
+ public boolean sendMessage(IMessage message) throws TransportException, OverloadException, InternalException {
+ return false;
+ }
- public boolean hasValidConnection() {
- return connected;
- }
+ public boolean hasValidConnection() {
+ return connected;
+ }
- public void setRealm(String realm) {
+ public void setRealm(String realm) {
- }
+ }
- public void addStateChangeListener(StateChangeListener listener) {
+ public void addStateChangeListener(StateChangeListener listener) {
- }
+ }
- public void remStateChangeListener(StateChangeListener listener) {
+ public void remStateChangeListener(StateChangeListener listener) {
- }
+ }
- public void addConnectionListener(IConnectionListener listener) {
+ public void addConnectionListener(IConnectionListener listener) {
- }
+ }
- public void remConnectionListener(IConnectionListener listener) {
+ public void remConnectionListener(IConnectionListener listener) {
- }
+ }
- public IStatistic getStatistic() {
- return statistic;
- }
+ public IStatistic getStatistic() {
+ return statistic;
+ }
- public boolean isConnected() {
- return connected;
- }
+ public boolean isConnected() {
+ return connected;
+ }
- public void connect() throws InternalException, IOException, IllegalDiameterStateException {
+ public void connect() throws InternalException, IOException, IllegalDiameterStateException {
- }
+ }
- @Override
- public void disconnect(int disconnectCause) throws InternalException, IllegalDiameterStateException {
+ @Override
+ public void disconnect(int disconnectCause) throws InternalException, IllegalDiameterStateException {
- }
+ }
- public E getState(Class enumc) {
- return null;
- }
+ public E getState(Class enumc) {
+ return null;
+ }
- public URI getUri() {
- return uri;
- }
+ public URI getUri() {
+ return uri;
+ }
- public InetAddress[] getIPAddresses() {
- return new InetAddress[0];
- }
+ public InetAddress[] getIPAddresses() {
+ return new InetAddress[0];
+ }
- public String getRealmName() {
- return null;
- }
+ public String getRealmName() {
+ return null;
+ }
- public long getVendorId() {
- return 0;
- }
+ public long getVendorId() {
+ return 0;
+ }
- public String getProductName() {
- return null;
- }
+ public String getProductName() {
+ return null;
+ }
- public long getFirmware() {
- return 0;
- }
+ public long getFirmware() {
+ return 0;
+ }
- public Set getCommonApplications() {
- return null;
- }
+ public Set getCommonApplications() {
+ return null;
+ }
- public void addPeerStateListener(PeerStateListener listener) {
+ public void addPeerStateListener(PeerStateListener listener) {
- }
+ }
- public void removePeerStateListener(PeerStateListener listener) {
+ public void removePeerStateListener(PeerStateListener listener) {
- }
+ }
- @Override
- public String toString() {
- return "Peer-"+id;
- }
+ @Override
+ public String toString() {
+ return "Peer-" + id;
}
+ }
}
\ No newline at end of file
diff --git a/core/jdiameter/impl/src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml b/core/jdiameter/impl/src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml
new file mode 100644
index 000000000..c908eb331
--- /dev/null
+++ b/core/jdiameter/impl/src/test/resources/jdiameter-weightedroundrobinresubmitting-config.xml
@@ -0,0 +1,160 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/core/jdiameter/impl/src/test/resources/log4j.properties b/core/jdiameter/impl/src/test/resources/log4j.properties
new file mode 100644
index 000000000..24d12894e
--- /dev/null
+++ b/core/jdiameter/impl/src/test/resources/log4j.properties
@@ -0,0 +1,9 @@
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=TRACE, A1
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+# A1 uses PatternLayout.
+log4j.appender.A1.layout=org.apache.log4j.PatternLayout
+log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n