Pull request for issue #10: Support to one-to-many type SCTP sockets

.gitignore

@@ -3,3 +3,8 @@ target
 /.project
 /ManagementTest_sctp.xml
 /sctptest.log
+
+# Ignore AppNGIN-generated build.xml and ivy.xml
+build.xml
+ivy.xml
+

sctp-api/src/main/java/org/mobicents/protocols/api/Management.java

@@ -274,6 +274,32 @@ public Server addServer(String serverName, String hostAddress, int port, IpChann
 	public Association addServerAssociation(String peerAddress, int peerPort, String serverName, String assocName, IpChannelType ipChannelType)
 			throws Exception;
 
+	/**
+	 * Add server Association. IP channel type is SCTP.
+	 * 
+	 * @param peerAddress
+	 *            the peer IP address that this association will accept
+	 *            connection from
+	 * @param peerPort
+	 *            the peer port that this association will accept connection
+	 *            from
+	 * @param serverName
+	 *            the Server that this association belongs to
+	 * @param assocName
+	 *            unique name of Association
+	 * @param ipChannelType
+	 *            IP channel type: SCTP or TCP
+	 *            
+	 * @param extraHostAddresses
+	 *            When SCTP multi-homing configuration extra IP addresses can be put here
+	 *            If multi-homing absence this parameter can be null 	 
+	 * * @return
+	 * @throws Exception
+	 */
+	public Association addServerAssociation(String peerAddress, int peerPort, String serverName, String assocName, IpChannelType ipChannelType, String[] extraHostAddresses)
+			throws Exception;
+
+
 	/**
 	 * Add Association. IP channel type is SCTP.
 	 * 

sctp-impl/src/main/java/org/mobicents/protocols/sctp/AssociationImpl.java

@@ -182,6 +182,30 @@ public AssociationImpl(String peerAddress, int peerPort, String serverName, Stri
 		this.type = AssociationType.SERVER;
 
 	}
+
+	/**
+	 * Creating a SERVER Association
+	 * 
+	 * @param peerAddress
+	 * @param peerPort
+	 * @param serverName
+	 * @param assocName
+	 * @param ipChannelType
+	 * @param extraHostAddresses
+	 */
+	public AssociationImpl(String peerAddress, int peerPort, String serverName, String assocName,
+			IpChannelType ipChannelType,  String[] extraHostAddresses) {
+		this();
+		this.peerAddress = peerAddress;
+		this.peerPort = peerPort;
+		this.serverName = serverName;
+		this.name = assocName;
+		this.ipChannelType = ipChannelType;
+		this.extraHostAddresses = extraHostAddresses;
+
+		this.type = AssociationType.SERVER;
+
+	}
 
 	/**
 	 * Creating an ANONYMOUS_SERVER Association

sctp-impl/src/main/java/org/mobicents/protocols/sctp/ManagementImpl.java

@@ -59,6 +59,7 @@ public class ManagementImpl implements Management {
 
 	private static final Logger logger = Logger.getLogger(ManagementImpl.class);
 
+	private static final String DISABLE_CONFIG_PERSISTANCE_KEY = "ss7.disableDefaultConfigPersistance";
 	private static final String SCTP_PERSIST_DIR_KEY = "sctp.persist.dir";
 	private static final String USER_DIR_KEY = "user.dir";
 	private static final String PERSIST_FILE_NAME = "sctp.xml";
@@ -393,8 +394,16 @@ public boolean isStarted(){
 		return this.started;
 	}
 
+	private boolean isConfigPersistanceDisabled() {
+		String disableConfigPersistanceString = System.getProperty(DISABLE_CONFIG_PERSISTANCE_KEY, "false");
+		return Boolean.valueOf(disableConfigPersistanceString);
+	}
+
 	@SuppressWarnings("unchecked")
 	public void load() throws FileNotFoundException {
+		if (isConfigPersistanceDisabled()) {
+			return;
+		}
 		XMLObjectReader reader = null;
 		try {
 			reader = XMLObjectReader.newInstance(new FileInputStream(persistFile.toString()));
@@ -438,6 +447,9 @@ public void load() throws FileNotFoundException {
 	}
 
 	public void store() {
+		if (isConfigPersistanceDisabled()) {
+			return;
+		}
 		try {
 			XMLObjectWriter writer = XMLObjectWriter.newInstance(new FileOutputStream(persistFile.toString()));
 			writer.setBinding(binding);
@@ -683,6 +695,8 @@ public AssociationImpl addServerAssociation(String peerAddress, int peerPort, St
 
 	public AssociationImpl addServerAssociation(String peerAddress, int peerPort, String serverName, String assocName, IpChannelType ipChannelType)
 			throws Exception {
+		return (AssociationImpl)addServerAssociation(peerAddress, peerPort, serverName, assocName, ipChannelType, null);
+		/*
 
 		if (!this.started) {
 			throw new Exception(String.format("Management=%s not started", this.name));
@@ -763,6 +777,95 @@ public AssociationImpl addServerAssociation(String peerAddress, int peerPort, St
 				logger.info(String.format("Added Associoation=%s of type=%s", association.getName(), association.getAssociationType()));
 			}
 
+			return association;
+		}
+		*/
+	}
+
+	@Override
+	public Association addServerAssociation(String peerAddress, int peerPort, String serverName, String assocName,
+			IpChannelType ipChannelType, String[] extraHostAddresses) throws Exception {
+		// MADE BY KB
+
+		if (!this.started) {
+			throw new Exception(String.format("Management=%s not started", this.name));
+		}
+
+		if (peerAddress == null) {
+			throw new Exception("Peer address cannot be null");
+		}
+
+		if (peerPort < 1) {
+			throw new Exception("Peer port cannot be less than 1");
+		}
+
+		if (serverName == null) {
+			throw new Exception("Server name cannot be null");
+		}
+
+		if (assocName == null) {
+			throw new Exception("Association name cannot be null");
+		}
+
+		synchronized (this) {
+			if (this.associations.get(assocName) != null) {
+				throw new Exception(String.format("Already has association=%s", assocName));
+			}
+
+			Server server = null;
+
+			for (FastList.Node<Server> n = this.servers.head(), end = this.servers.tail(); (n = n.getNext()) != end;) {
+				Server serverTemp = n.getValue();
+				if (serverTemp.getName().equals(serverName)) {
+					server = serverTemp;
+				}
+			}
+
+			if (server == null) {
+				throw new Exception(String.format("No Server found for name=%s", serverName));
+			}
+
+			for (FastMap.Entry<String, Association> n = this.associations.head(), end = this.associations.tail(); (n = n.getNext()) != end;) {
+				Association associationTemp = n.getValue();
+
+				if (peerAddress.equals(associationTemp.getPeerAddress()) && associationTemp.getPeerPort() == peerPort) {
+					throw new Exception(String.format("Already has association=%s with same peer address=%s and port=%d", associationTemp.getName(),
+							peerAddress, peerPort));
+				}
+			}
+
+			if (server.getIpChannelType() != ipChannelType)
+				throw new Exception(String.format("Server and Accociation has different IP channel type"));
+
+			AssociationImpl association = new AssociationImpl(peerAddress, peerPort, serverName, assocName, ipChannelType, extraHostAddresses);
+			association.setManagement(this);
+
+			AssociationMap<String, Association> newAssociations = new AssociationMap<String, Association>();
+			newAssociations.putAll(this.associations);
+			newAssociations.put(assocName, association);
+			this.associations = newAssociations;
+			// this.associations.put(assocName, association);
+
+			FastList<String> newAssociations2 = new FastList<String>();
+			newAssociations2.addAll(((ServerImpl) server).associations);
+			newAssociations2.add(assocName);
+			((ServerImpl) server).associations = newAssociations2;
+			// ((ServerImpl) server).associations.add(assocName);
+
+			this.store();
+
+			for (ManagementEventListener lstr : managementEventListeners) {
+				try {
+					lstr.onAssociationAdded(association);
+				} catch (Throwable ee) {
+					logger.error("Exception while invoking onAssociationAdded", ee);
+				}
+			}
+
+			if (logger.isInfoEnabled()) {
+				logger.info(String.format("Added Associoation=%s of type=%s", association.getName(), association.getAssociationType()));
+			}
+
 			return association;
 		}
 	}
@@ -845,6 +948,26 @@ public AssociationImpl addAssociation(String hostAddress, int hostPort, String p
 		}
 	}
 
+	public Association getAssociation(String peerAddress, int peerPort) throws Exception {
+		if (peerAddress == null || peerPort == 0) {
+			throw new Exception("PeerIp is null or peerPort is zero!");
+		}
+		Association associationToFind = null;
+		for (FastMap.Entry<String, Association> n = associations.head(), end = associations.tail(); (n = n.getNext()) != end;) {
+			Association associationTemp = n.getValue();
+			if (associationTemp.getPeerAddress().equals(peerAddress) && associationTemp.getPeerPort() == peerPort) {
+				associationToFind = associationTemp;
+			}
+		}
+
+
+		if (associationToFind == null) {
+			throw new Exception(String.format("No Association found based on PeerAddress=%s and PeerPort=%2", peerAddress, peerPort));
+		}
+		return associationToFind;
+	}
+
+
 	public Association getAssociation(String assocName) throws Exception {
 		if (assocName == null) {
 			throw new Exception("Association name cannot be null");

sctp-impl/src/main/java/org/mobicents/protocols/sctp/SelectorThread.java

@@ -222,8 +222,10 @@ private void doAccept(AbstractSelectableChannel serverSocketChannel, AbstractSel
 		// server selection
 		for (Server srv : this.management.servers) {
 			ServerImpl srvv = (ServerImpl) srv;
+			logger.debug(String.format("*** Server: %s", srvv));
 			if (srvv.getIpChannel() == serverSocketChannel) { // we have found a server
 				for (SocketAddress sockAdd : peerAddresses) {
+					logger.debug(String.format("****** Socket Address: %s", sockAdd));
 
 					inetAddress = ((InetSocketAddress) sockAdd).getAddress();
 					port = ((InetSocketAddress) sockAdd).getPort();
@@ -239,7 +241,7 @@ private void doAccept(AbstractSelectableChannel serverSocketChannel, AbstractSel
 
 					for (FastMap.Entry<String, Association> n = associations.head(), end = associations.tail(); (n = n.getNext()) != end && !provisioned;) {
 						AssociationImpl association = (AssociationImpl)n.getValue();
-
+						logger.debug(String.format("********* Association: %s", association));
 						// check if an association binds to the found server
 						if (srv.getName().equals(association.getServerName())) {
 
@@ -318,6 +320,28 @@ private void doAccept(AbstractSelectableChannel serverSocketChannel, AbstractSel
 						}
 						return;
 					}
+					// changes begin
+					// get anonymAssociation created
+					try {
+						AssociationImpl tmpAssociation = null;
+						String assocByName = anonymAssociation.getPeerAddress()+":"+anonymAssociation.getPeerPort();
+						try {
+							tmpAssociation = (AssociationImpl)this.management.getAssociation(assocByName);
+						} catch (Exception ex) {
+							logger.debug(String.format("Could not find association based on name: %s. We will try to find based on address and port...", assocByName));
+						}
+						if (tmpAssociation == null) {
+							tmpAssociation = (AssociationImpl)this.management.getAssociation(anonymAssociation.getPeerAddress(), anonymAssociation.getPeerPort());
+						}
+                        if (tmpAssociation != null) {
+                            tmpAssociation.setSocketChannel(socketChannel);
+                            tmpAssociation.setManagement(this.management);
+                            anonymAssociation = tmpAssociation;
+                        }
+                    } catch (Exception e) {
+                        logger.error(String.format("Rejected anonymous %s", anonymAssociation), e);
+                    }
+					// changes - end
 
 					if (!anonymAssociation.isStarted()) {
 						// connection is rejected
@@ -338,7 +362,8 @@ private void doAccept(AbstractSelectableChannel serverSocketChannel, AbstractSel
 					if (logger.isInfoEnabled()) {
 						logger.info(String.format("Accepted anonymous %s", anonymAssociation));
 					}
-
+
+
 					if (anonymAssociation.getIpChannelType() == IpChannelType.TCP) {
 						AssocChangeEvent ace = AssocChangeEvent.COMM_UP;
 						AssociationChangeNotification2 acn = new AssociationChangeNotification2(ace);

sctp-impl/src/main/java/org/mobicents/protocols/sctp/ServerImpl.java

@@ -40,6 +40,7 @@
 import org.mobicents.protocols.api.Server;
 
 import com.sun.nio.sctp.SctpServerChannel;
+import com.sun.nio.sctp.SctpStandardSocketOptions;
 
 /**
  * @author amit bhayani
@@ -177,6 +178,8 @@ private void doInitSocketSctp() throws IOException {
 		// Create a new non-blocking server socket channel
 		this.serverChannelSctp = SctpServerChannel.open();
 		this.serverChannelSctp.configureBlocking(false);
+		//KB:
+		this.serverChannelSctp.setOption(SctpStandardSocketOptions.SCTP_INIT_MAXSTREAMS, SctpStandardSocketOptions.InitMaxStreams.create(17, 17));
 
 		// Bind the server socket to the specified address and port
 		InetSocketAddress isa = new InetSocketAddress(this.hostAddress, this.hostport);