Commit 44af3b5b authored by Gaston Dombiak's avatar Gaston Dombiak Committed by gato

Modified to support non-blocking connections. JM-687

git-svn-id: http://svn.igniterealtime.org/svn/repos/wildfire/trunk@3904 b35dd754-fafc-0310-a699-88a17e54d16e
parent 1ff31a5a
...@@ -9,18 +9,14 @@ ...@@ -9,18 +9,14 @@
package org.dom4j.io; package org.dom4j.io;
import java.io.*; import org.dom4j.*;
import java.net.URL; import org.jivesoftware.wildfire.net.MXParser;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentFactory;
import org.dom4j.Element;
import org.dom4j.ElementHandler;
import org.dom4j.QName;
import org.xmlpull.v1.XmlPullParser; import org.xmlpull.v1.XmlPullParser;
import org.xmlpull.v1.XmlPullParserException; import org.xmlpull.v1.XmlPullParserException;
import org.xmlpull.v1.XmlPullParserFactory; import org.xmlpull.v1.XmlPullParserFactory;
import org.jivesoftware.wildfire.net.MXParser;
import java.io.*;
import java.net.URL;
/** /**
* <p><code>XMPPPacketReader</code> is a Reader of DOM4J documents that * <p><code>XMPPPacketReader</code> is a Reader of DOM4J documents that
...@@ -277,6 +273,23 @@ public class XMPPPacketReader { ...@@ -277,6 +273,23 @@ public class XMPPPacketReader {
return lastActive > lastHeartbeat ? lastActive : lastHeartbeat; return lastActive > lastHeartbeat ? lastActive : lastHeartbeat;
} }
/*
* DANIELE: Add parse document by string
*/
public Document parseDocument(String xml) throws DocumentException {
/*
// Long way with reuse of DocumentFactory.
DocumentFactory df = getDocumentFactory();
SAXReader reader = new SAXReader( df );
Document document = reader.read( new StringReader( xml );*/
// Simple way
// TODO Optimize. Do not create a sax reader for each parsing
Document document = DocumentHelper.parseText(xml);
return document;
}
// Implementation methods // Implementation methods
//------------------------------------------------------------------------- //-------------------------------------------------------------------------
public Document parseDocument() throws DocumentException, IOException, XmlPullParserException { public Document parseDocument() throws DocumentException, IOException, XmlPullParserException {
......
...@@ -11,9 +11,11 @@ ...@@ -11,9 +11,11 @@
package org.jivesoftware.wildfire; package org.jivesoftware.wildfire;
import org.jivesoftware.wildfire.net.SocketReader;
import java.io.IOException;
import java.net.Socket; import java.net.Socket;
import java.util.Iterator; import java.util.Iterator;
import org.xmlpull.v1.XmlPullParserException;
/** /**
* Coordinates connections (accept, read, termination) on the server. * Coordinates connections (accept, read, termination) on the server.
...@@ -31,15 +33,17 @@ public interface ConnectionManager { ...@@ -31,15 +33,17 @@ public interface ConnectionManager {
public Iterator<ServerPort> getPorts(); public Iterator<ServerPort> getPorts();
/** /**
* Adds a socket to be managed by the connection manager. * Creates a new socket reader for the new accepted socket to be managed
* by the connection manager.
* *
* @param socket the socket to add to this manager for management. * @param socket the new accepted socket by this manager.
* @param isSecure true if the connection is secure. * @param isSecure true if the connection is secure.
* @param serverPort holds information about the port on which the server is listening for * @param serverPort holds information about the port on which the server is listening for
* connections. * connections.
* @param useBlockingMode true means that the server will use a thread per connection.
*/ */
public void addSocket(Socket socket, boolean isSecure, ServerPort serverPort) public SocketReader createSocketReader(Socket socket, boolean isSecure, ServerPort serverPort,
throws XmlPullParserException; boolean useBlockingMode) throws IOException;
/** /**
* Sets if the port listener for unsecured clients will be available or not. When disabled * Sets if the port listener for unsecured clients will be available or not. When disabled
......
...@@ -12,10 +12,10 @@ ...@@ -12,10 +12,10 @@
package org.jivesoftware.wildfire.net; package org.jivesoftware.wildfire.net;
import org.dom4j.Element; import org.dom4j.Element;
import org.jivesoftware.util.JiveGlobals;
import org.jivesoftware.wildfire.ClientSession; import org.jivesoftware.wildfire.ClientSession;
import org.jivesoftware.wildfire.PacketRouter; import org.jivesoftware.wildfire.PacketRouter;
import org.jivesoftware.wildfire.auth.UnauthorizedException; import org.jivesoftware.wildfire.auth.UnauthorizedException;
import org.jivesoftware.util.JiveGlobals;
import org.xmlpull.v1.XmlPullParserException; import org.xmlpull.v1.XmlPullParserException;
import org.xmpp.packet.IQ; import org.xmpp.packet.IQ;
import org.xmpp.packet.Message; import org.xmpp.packet.Message;
...@@ -40,8 +40,8 @@ import java.net.Socket; ...@@ -40,8 +40,8 @@ import java.net.Socket;
public class ClientSocketReader extends SocketReader { public class ClientSocketReader extends SocketReader {
public ClientSocketReader(PacketRouter router, String serverName, Socket socket, public ClientSocketReader(PacketRouter router, String serverName, Socket socket,
SocketConnection connection) { SocketConnection connection, boolean useBlockingMode) {
super(router, serverName, socket, connection); super(router, serverName, socket, connection, useBlockingMode);
} }
protected void processIQ(IQ packet) throws UnauthorizedException { protected void processIQ(IQ packet) throws UnauthorizedException {
...@@ -87,6 +87,10 @@ public class ClientSocketReader extends SocketReader { ...@@ -87,6 +87,10 @@ public class ClientSocketReader extends SocketReader {
return "jabber:client"; return "jabber:client";
} }
String getName() {
return "Client SR - " + hashCode();
}
boolean validateHost() { boolean validateHost() {
return JiveGlobals.getBooleanProperty("xmpp.client.validate.host",false); return JiveGlobals.getBooleanProperty("xmpp.client.validate.host",false);
} }
......
...@@ -33,8 +33,8 @@ import java.net.Socket; ...@@ -33,8 +33,8 @@ import java.net.Socket;
public class ComponentSocketReader extends SocketReader { public class ComponentSocketReader extends SocketReader {
public ComponentSocketReader(PacketRouter router, String serverName, Socket socket, public ComponentSocketReader(PacketRouter router, String serverName, Socket socket,
SocketConnection connection) { SocketConnection connection, boolean useBlockingMode) {
super(router, serverName, socket, connection); super(router, serverName, socket, connection, useBlockingMode);
} }
/** /**
...@@ -122,6 +122,10 @@ public class ComponentSocketReader extends SocketReader { ...@@ -122,6 +122,10 @@ public class ComponentSocketReader extends SocketReader {
return "jabber:component:accept"; return "jabber:component:accept";
} }
String getName() {
return "Component SR - " + hashCode();
}
boolean validateHost() { boolean validateHost() {
return false; return false;
} }
......
...@@ -69,8 +69,8 @@ public class ConnectionMultiplexerSocketReader extends SocketReader { ...@@ -69,8 +69,8 @@ public class ConnectionMultiplexerSocketReader extends SocketReader {
private MultiplexerPacketHandler packetHandler; private MultiplexerPacketHandler packetHandler;
public ConnectionMultiplexerSocketReader(PacketRouter router, String serverName, Socket socket, public ConnectionMultiplexerSocketReader(PacketRouter router, String serverName, Socket socket,
SocketConnection connection) { SocketConnection connection, boolean useBlockingMode) {
super(router, serverName, socket, connection); super(router, serverName, socket, connection, useBlockingMode);
// Create a pool of threads that will process received packets. If more threads are // Create a pool of threads that will process received packets. If more threads are
// required then the command will be executed on the SocketReader process // required then the command will be executed on the SocketReader process
int coreThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.core.threads", 10); int coreThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.core.threads", 10);
...@@ -205,6 +205,10 @@ public class ConnectionMultiplexerSocketReader extends SocketReader { ...@@ -205,6 +205,10 @@ public class ConnectionMultiplexerSocketReader extends SocketReader {
} }
} }
String getName() {
return "ConnectionMultiplexer SR - " + hashCode();
}
boolean validateHost() { boolean validateHost() {
return false; return false;
} }
......
...@@ -26,7 +26,9 @@ import java.net.UnknownHostException; ...@@ -26,7 +26,9 @@ import java.net.UnknownHostException;
/** /**
* Implements a network front end with a dedicated thread reading * Implements a network front end with a dedicated thread reading
* each incoming socket. * each incoming socket. The old SSL method always uses a blocking model.
*
* @author Gaston Dombiak
*/ */
public class SSLSocketAcceptThread extends Thread { public class SSLSocketAcceptThread extends Thread {
...@@ -139,7 +141,12 @@ public class SSLSocketAcceptThread extends Thread { ...@@ -139,7 +141,12 @@ public class SSLSocketAcceptThread extends Thread {
try { try {
Socket sock = serverSocket.accept(); Socket sock = serverSocket.accept();
Log.debug("SSL Connect " + sock.toString()); Log.debug("SSL Connect " + sock.toString());
connManager.addSocket(sock, true, serverPort); SocketReader reader = connManager.createSocketReader(sock, true, serverPort, true);
// Create a new reading thread for each new connected client
Thread thread = new Thread(reader, reader.getName());
thread.setDaemon(true);
thread.setPriority(Thread.NORM_PRIORITY);
thread.start();
} }
catch (SSLException se) { catch (SSLException se) {
long exceptionTime = System.currentTimeMillis(); long exceptionTime = System.currentTimeMillis();
......
...@@ -50,8 +50,8 @@ public class ServerSocketReader extends SocketReader { ...@@ -50,8 +50,8 @@ public class ServerSocketReader extends SocketReader {
private ThreadPoolExecutor threadPool; private ThreadPoolExecutor threadPool;
public ServerSocketReader(PacketRouter router, String serverName, Socket socket, public ServerSocketReader(PacketRouter router, String serverName, Socket socket,
SocketConnection connection) { SocketConnection connection, boolean useBlockingMode) {
super(router, serverName, socket, connection); super(router, serverName, socket, connection, useBlockingMode);
// Create a pool of threads that will process received packets. If more threads are // Create a pool of threads that will process received packets. If more threads are
// required then the command will be executed on the SocketReader process // required then the command will be executed on the SocketReader process
int coreThreads = JiveGlobals.getIntProperty("xmpp.server.processing.core.threads", 2); int coreThreads = JiveGlobals.getIntProperty("xmpp.server.processing.core.threads", 2);
...@@ -217,6 +217,10 @@ public class ServerSocketReader extends SocketReader { ...@@ -217,6 +217,10 @@ public class ServerSocketReader extends SocketReader {
return "jabber:server"; return "jabber:server";
} }
String getName() {
return "Server SR - " + hashCode();
}
boolean validateHost() { boolean validateHost() {
return true; return true;
} }
......
...@@ -12,21 +12,20 @@ ...@@ -12,21 +12,20 @@
package org.jivesoftware.wildfire.net; package org.jivesoftware.wildfire.net;
import org.jivesoftware.util.JiveGlobals; import org.jivesoftware.util.JiveGlobals;
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.Log;
import org.jivesoftware.wildfire.ConnectionManager; import org.jivesoftware.wildfire.ConnectionManager;
import org.jivesoftware.wildfire.ServerPort; import org.jivesoftware.wildfire.ServerPort;
import java.io.IOException; import java.io.IOException;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
/** /**
* Implements a network front end with a dedicated thread reading * Implements a network front end with a dedicated thread reading
* each incoming socket. * each incoming socket. Blocking and non-blocking modes are supported.
* By default blocking mode is used. Use the <i>xmpp.socket.blocking</i>
* system property to change the blocking mode. Restart the server after making
* changes to the system property.
* *
* @author Iain Shigeoka * @author Gaston Dombiak
*/ */
public class SocketAcceptThread extends Thread { public class SocketAcceptThread extends Thread {
...@@ -55,22 +54,11 @@ public class SocketAcceptThread extends Thread { ...@@ -55,22 +54,11 @@ public class SocketAcceptThread extends Thread {
*/ */
private ServerPort serverPort; private ServerPort serverPort;
/** private SocketAcceptingMode acceptingMode;
* True while this thread should continue running.
*/
private boolean notTerminated = true;
/**
* socket that listens for connections.
*/
ServerSocket serverSocket;
private ConnectionManager connManager;
public SocketAcceptThread(ConnectionManager connManager, ServerPort serverPort) public SocketAcceptThread(ConnectionManager connManager, ServerPort serverPort)
throws IOException { throws IOException {
super("Socket Listener at port " + serverPort.getPort()); super("Socket Listener at port " + serverPort.getPort());
this.connManager = connManager;
this.serverPort = serverPort; this.serverPort = serverPort;
// Listen on a specific network interface if it has been set. // Listen on a specific network interface if it has been set.
String interfaceName = JiveGlobals.getXMLProperty("network.interface"); String interfaceName = JiveGlobals.getXMLProperty("network.interface");
...@@ -80,7 +68,14 @@ public class SocketAcceptThread extends Thread { ...@@ -80,7 +68,14 @@ public class SocketAcceptThread extends Thread {
bindInterface = InetAddress.getByName(interfaceName); bindInterface = InetAddress.getByName(interfaceName);
} }
} }
serverSocket = new ServerSocket(serverPort.getPort(), -1, bindInterface); // Set the blocking reading mode to use
boolean useBlockingMode = JiveGlobals.getBooleanProperty("xmpp.socket.blocking", true);
if (useBlockingMode) {
acceptingMode = new BlockingAcceptingMode(connManager, serverPort, bindInterface);
}
else {
acceptingMode = new NonBlockingAcceptingMode(connManager, serverPort, bindInterface);
}
} }
/** /**
...@@ -105,19 +100,7 @@ public class SocketAcceptThread extends Thread { ...@@ -105,19 +100,7 @@ public class SocketAcceptThread extends Thread {
* Unblock the thread and force it to terminate. * Unblock the thread and force it to terminate.
*/ */
public void shutdown() { public void shutdown() {
notTerminated = false; acceptingMode.shutdown();
try {
ServerSocket sSock = serverSocket;
serverSocket = null;
if (sSock != null) {
sSock.close();
}
}
catch (IOException e) {
// we don't care, no matter what, the socket should be dead
}
} }
/** /**
...@@ -125,34 +108,8 @@ public class SocketAcceptThread extends Thread { ...@@ -125,34 +108,8 @@ public class SocketAcceptThread extends Thread {
* call getting sockets and handing them to the SocketManager. * call getting sockets and handing them to the SocketManager.
*/ */
public void run() { public void run() {
while (notTerminated) { acceptingMode.run();
try { // We stopped accepting new connections so close the listener
Socket sock = serverSocket.accept(); shutdown();
if (sock != null) {
Log.debug("Connect " + sock.toString());
connManager.addSocket(sock, false, serverPort);
}
}
catch (IOException ie) {
if (notTerminated) {
Log.error(LocaleUtils.getLocalizedString("admin.error.accept"),
ie);
}
}
catch (Throwable e) {
Log.error(LocaleUtils.getLocalizedString("admin.error.accept"), e);
}
}
try {
ServerSocket sSock = serverSocket;
serverSocket = null;
if (sSock != null) {
sSock.close();
}
}
catch (IOException e) {
// we don't care, no matter what, the socket should be dead
}
} }
} }
...@@ -30,6 +30,7 @@ import java.io.OutputStreamWriter; ...@@ -30,6 +30,7 @@ import java.io.OutputStreamWriter;
import java.io.Writer; import java.io.Writer;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.Socket; import java.net.Socket;
import java.nio.channels.Channels;
import java.util.Collection; import java.util.Collection;
import java.util.Date; import java.util.Date;
import java.util.HashMap; import java.util.HashMap;
...@@ -122,7 +123,13 @@ public class SocketConnection implements Connection { ...@@ -122,7 +123,13 @@ public class SocketConnection implements Connection {
this.secure = isSecure; this.secure = isSecure;
this.socket = socket; this.socket = socket;
writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), CHARSET)); // DANIELE: Modify socket to use channel
if (socket.getChannel() != null) {
writer = Channels.newWriter(socket.getChannel(), CHARSET);
}
else {
writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), CHARSET));
}
this.backupDeliverer = backupDeliverer; this.backupDeliverer = backupDeliverer;
xmlSerializer = new XMLSocketWriter(writer, this); xmlSerializer = new XMLSocketWriter(writer, this);
......
...@@ -177,8 +177,15 @@ public class TLSStreamHandler { ...@@ -177,8 +177,15 @@ public class TLSStreamHandler {
reader = new TLSStreamReader(wrapper, socket); reader = new TLSStreamReader(wrapper, socket);
writer = new TLSStreamWriter(wrapper, socket); writer = new TLSStreamWriter(wrapper, socket);
rbc = Channels.newChannel(socket.getInputStream()); // DANIELE: Add code to use directly the socket-channel.
wbc = Channels.newChannel(socket.getOutputStream()); if (socket.getChannel() != null) {
rbc = socket.getChannel();
wbc = socket.getChannel();
}
else {
rbc = Channels.newChannel(socket.getInputStream());
wbc = Channels.newChannel(socket.getOutputStream());
}
initialHSStatus = HandshakeStatus.NEED_UNWRAP; initialHSStatus = HandshakeStatus.NEED_UNWRAP;
initialHSComplete = false; initialHSComplete = false;
......
...@@ -47,8 +47,14 @@ public class TLSStreamReader { ...@@ -47,8 +47,14 @@ public class TLSStreamReader {
public TLSStreamReader(TLSWrapper tlsWrapper, Socket socket) throws IOException { public TLSStreamReader(TLSWrapper tlsWrapper, Socket socket) throws IOException {
wrapper = tlsWrapper; wrapper = tlsWrapper;
rbc = Channels.newChannel(socket.getInputStream()); // DANIELE: Add code to use directly the socket channel
inNetBB = ByteBuffer.allocate(wrapper.getNetBuffSize()); if (socket.getChannel() != null) {
rbc = socket.getChannel();
}
else {
rbc = Channels.newChannel(socket.getInputStream());
}
inNetBB = ByteBuffer.allocate(wrapper.getNetBuffSize());
inAppBB = ByteBuffer.allocate(wrapper.getAppBuffSize()); inAppBB = ByteBuffer.allocate(wrapper.getAppBuffSize());
} }
......
...@@ -38,8 +38,14 @@ public class TLSStreamWriter { ...@@ -38,8 +38,14 @@ public class TLSStreamWriter {
public TLSStreamWriter(TLSWrapper tlsWrapper, Socket socket) throws IOException { public TLSStreamWriter(TLSWrapper tlsWrapper, Socket socket) throws IOException {
wrapper = tlsWrapper; wrapper = tlsWrapper;
wbc = Channels.newChannel(socket.getOutputStream()); // DANIELE: Add code to use directly the socket channel
outAppData = ByteBuffer.allocate(tlsWrapper.getAppBuffSize()); if (socket.getChannel() != null) {
wbc = socket.getChannel();
}
else {
wbc = Channels.newChannel(socket.getOutputStream());
}
outAppData = ByteBuffer.allocate(tlsWrapper.getAppBuffSize());
} }
private void doWrite(ByteBuffer buff) throws IOException { private void doWrite(ByteBuffer buff) throws IOException {
......
...@@ -256,42 +256,28 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana ...@@ -256,42 +256,28 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
return ports.iterator(); return ports.iterator();
} }
public void addSocket(Socket sock, boolean isSecure, ServerPort serverPort) { public SocketReader createSocketReader(Socket sock, boolean isSecure, ServerPort serverPort,
try { boolean useBlockingMode) throws IOException {
// the order of these calls is critical (stupid huh?) if (serverPort.isClientPort()) {
SocketReader reader; SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
String threadName; return new ClientSocketReader(router, serverName, sock, conn, useBlockingMode);
if (serverPort.isClientPort()) { }
SocketConnection conn = new SocketConnection(deliverer, sock, isSecure); else if (serverPort.isComponentPort()) {
reader = new ClientSocketReader(router, serverName, sock, conn); SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
threadName = "Client SR - " + reader.hashCode(); return new ComponentSocketReader(router, serverName, sock, conn, useBlockingMode);
}
else if (serverPort.isComponentPort()) {
SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
reader = new ComponentSocketReader(router, serverName, sock, conn);
threadName = "Component SR - " + reader.hashCode();
}
else if (serverPort.isServerPort()) {
SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
reader = new ServerSocketReader(router, serverName, sock, conn);
threadName = "Server SR - " + reader.hashCode();
}
else {
// Use the appropriate packeet deliverer for connection managers. The packet
// deliverer will be configured with the domain of the connection manager once
// the connection manager has finished the handshake.
SocketConnection conn =
new SocketConnection(new MultiplexerPacketDeliverer(), sock, isSecure);
reader = new ConnectionMultiplexerSocketReader(router, serverName, sock, conn);
threadName = "ConnectionMultiplexer SR - " + reader.hashCode();
}
Thread thread = new Thread(reader, threadName);
thread.setDaemon(true);
thread.setPriority(Thread.NORM_PRIORITY);
thread.start();
} }
catch (IOException e) { else if (serverPort.isServerPort()) {
Log.error(LocaleUtils.getLocalizedString("admin.error"), e); SocketConnection conn = new SocketConnection(deliverer, sock, isSecure);
return new ServerSocketReader(router, serverName, sock, conn, useBlockingMode);
}
else {
// Use the appropriate packeet deliverer for connection managers. The packet
// deliverer will be configured with the domain of the connection manager once
// the connection manager has finished the handshake.
SocketConnection conn =
new SocketConnection(new MultiplexerPacketDeliverer(), sock, isSecure);
return new ConnectionMultiplexerSocketReader(router, serverName, sock, conn,
useBlockingMode);
} }
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment