/*
* Conditions Of Use
*
* This software was developed by employees of the National Institute of
* Standards and Technology (NIST), an agency of the Federal Government.
* Pursuant to title 15 Untied States Code Section 105, works of NIST
* employees are not subject to copyright protection in the United States
* and are considered to be in the public domain. As a result, a formal
* license is not needed to use the software.
*
* This software is provided by NIST as a service and is expressly
* provided "AS IS." NIST MAKES NO WARRANTY OF ANY KIND, EXPRESS, IMPLIED
* OR STATUTORY, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTY OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT
* AND DATA ACCURACY. NIST does not warrant or make any representations
* regarding the use of the software or the results thereof, including but
* not limited to the correctness, accuracy, reliability or usefulness of
* the software.
*
* Permission to use this software is contingent upon your acceptance
* of the terms of this agreement
*
* .
*
*/
/******************************************************************************
* Product of NIST/ITL Advanced Networking Technologies Division (ANTD). *
******************************************************************************/
package gov.nist.javax.sip.stack;
import java.net.Socket;
import java.net.ServerSocket;
import java.io.IOException;
import java.net.SocketException;
import gov.nist.core.*;
import java.net.*;
import java.util.*;
/*
* Acknowledgement: Jeff Keyser suggested that a Stop mechanism be added to this. Niklas Uhrberg
* suggested that a means to limit the number of simultaneous active connections should be added.
* Mike Andrews suggested that the thread be accessible so as to implement clean stop using
* Thread.join(). Roger M. Persson contributed a bug fix for cleanup on stop().
*
*/
/**
* Sit in a loop waiting for incoming tcp connections and start a new thread to handle each new
* connection. This is the active object that creates new TCP MessageChannels (one for each new
* accept socket).
*
* @version 1.2 $Revision: 1.31 $ $Date: 2009/08/31 16:18:00 $
*
* @author M. Ranganathan <br/>
*
*
*/
public class TCPMessageProcessor extends MessageProcessor {
protected int nConnections;
private boolean isRunning;
private Hashtable tcpMessageChannels;
private ArrayList<TCPMessageChannel> incomingTcpMessageChannels;
private ServerSocket sock;
protected int useCount;
/**
* Constructor.
*
* @param sipStack SIPStack structure.
* @param port port where this message processor listens.
*/
protected TCPMessageProcessor(InetAddress ipAddress, SIPTransactionStack sipStack, int port) {
super(ipAddress, port, "tcp",sipStack);
this.sipStack = sipStack;
this.tcpMessageChannels = new Hashtable();
this.incomingTcpMessageChannels = new ArrayList<TCPMessageChannel>();
}
/**
* Start the processor.
*/
public void start() throws IOException {
Thread thread = new Thread(this);
thread.setName("TCPMessageProcessorThread");
thread.setPriority(Thread.MAX_PRIORITY);
thread.setDaemon(true);
this.sock = sipStack.getNetworkLayer().createServerSocket(getPort(), 0, getIpAddress());
if (getIpAddress().getHostAddress().equals(IN_ADDR_ANY)
|| getIpAddress().getHostAddress().equals(IN6_ADDR_ANY)) {
// Store the address to which we are actually bound
super.setIpAddress(sock.getInetAddress());
}
this.isRunning = true;
thread.start();
}
/**
* Run method for the thread that gets created for each accept socket.
*/
public void run() {
// Accept new connectins on our socket.
while (this.isRunning) {
try {
synchronized (this) {
// sipStack.maxConnections == -1 means we are
// willing to handle an "infinite" number of
// simultaneous connections (no resource limitation).
// This is the default behavior.
while (sipStack.maxConnections != -1
&& this.nConnections >= sipStack.maxConnections) {
try {
this.wait();
if (!this.isRunning)
return;
} catch (InterruptedException ex) {
break;
}
}
this.nConnections++;
}
Socket newsock = sock.accept();
if (sipStack.isLoggingEnabled()) {
getSIPStack().getStackLogger().logDebug("Accepting new connection!");
}
// Note that for an incoming message channel, the
// thread is already running
incomingTcpMessageChannels.add(new TCPMessageChannel(newsock, sipStack, this));
} catch (SocketException ex) {
this.isRunning = false;
} catch (IOException ex) {
// Problem accepting connection.
if (sipStack.isLoggingEnabled())
getSIPStack().getStackLogger().logException(ex);
continue;
} catch (Exception ex) {
InternalErrorHandler.handleException(ex);
}
}
}
/**
* Return the transport string.
*
* @return the transport string
*/
public String getTransport() {
return "tcp";
}
/**
* Returns the stack.
*
* @return my sip stack.
*/
public SIPTransactionStack getSIPStack() {
return sipStack;
}
/**
* Stop the message processor. Feature suggested by Jeff Keyser.
*/
public synchronized void stop() {
isRunning = false;
// this.listeningPoint = null;
try {
sock.close();
} catch (IOException e) {
e.printStackTrace();
}
Collection en = tcpMessageChannels.values();
for (Iterator it = en.iterator(); it.hasNext();) {
TCPMessageChannel next = (TCPMessageChannel) it.next();
next.close();
}
// RRPN: fix
for (Iterator incomingMCIterator = incomingTcpMessageChannels.iterator(); incomingMCIterator
.hasNext();) {
TCPMessageChannel next = (TCPMessageChannel) incomingMCIterator.next();
next.close();
}
this.notify();
}
protected synchronized void remove(TCPMessageChannel tcpMessageChannel) {
String key = tcpMessageChannel.getKey();
if (sipStack.isLoggingEnabled()) {
sipStack.getStackLogger().logDebug(Thread.currentThread() + " removing " + key);
}
/** May have been removed already */
if (tcpMessageChannels.get(key) == tcpMessageChannel) {
this.tcpMessageChannels.remove(key);
}
incomingTcpMessageChannels.remove(tcpMessageChannel);
}
public synchronized MessageChannel createMessageChannel(HostPort targetHostPort)
throws IOException {
String key = MessageChannel.getKey(targetHostPort, "TCP");
if (tcpMessageChannels.get(key) != null) {
return (TCPMessageChannel) this.tcpMessageChannels.get(key);
} else {
TCPMessageChannel retval = new TCPMessageChannel(targetHostPort.getInetAddress(),
targetHostPort.getPort(), sipStack, this);
this.tcpMessageChannels.put(key, retval);
retval.isCached = true;
if (sipStack.isLoggingEnabled()) {
sipStack.getStackLogger().logDebug("key " + key);
sipStack.getStackLogger().logDebug("Creating " + retval);
}
return retval;
}
}
protected synchronized void cacheMessageChannel(TCPMessageChannel messageChannel) {
String key = messageChannel.getKey();
TCPMessageChannel currentChannel = (TCPMessageChannel) tcpMessageChannels.get(key);
if (currentChannel != null) {
if (sipStack.isLoggingEnabled())
sipStack.getStackLogger().logDebug("Closing " + key);
currentChannel.close();
}
if (sipStack.isLoggingEnabled())
sipStack.getStackLogger().logDebug("Caching " + key);
this.tcpMessageChannels.put(key, messageChannel);
}
public synchronized MessageChannel createMessageChannel(InetAddress host, int port)
throws IOException {
try {
String key = MessageChannel.getKey(host, port, "TCP");
if (tcpMessageChannels.get(key) != null) {
return (TCPMessageChannel) this.tcpMessageChannels.get(key);
} else {
TCPMessageChannel retval = new TCPMessageChannel(host, port, sipStack, this);
this.tcpMessageChannels.put(key, retval);
retval.isCached = true;
if (sipStack.isLoggingEnabled()) {
sipStack.getStackLogger().logDebug("key " + key);
sipStack.getStackLogger().logDebug("Creating " + retval);
}
return retval;
}
} catch (UnknownHostException ex) {
throw new IOException(ex.getMessage());
}
}
/**
* TCP can handle an unlimited number of bytes.
*/
public int getMaximumMessageSize() {
return Integer.MAX_VALUE;
}
public boolean inUse() {
return this.useCount != 0;
}
/**
* Default target port for TCP
*/
public int getDefaultTargetPort() {
return 5060;
}
/**
* TCP is not a secure protocol.
*/
public boolean isSecure() {
return false;
}
}