在openfire中与客户端之间的交互代码主要在org.jivesoftware.openfire.nio，org.jivesoftware.openfire.net这个2个包中。
当openfire启动的时候，会去加载连接管理中心ConnectionManagerImpl这个类，其中有几个方法值得去细细阅读，
private synchronized void createListeners() {
if (isSocketStarted || sessionManager == null || deliverer == null || router == null || serverName == null) {
return;
}
// Create the port listener for s2s communication
createServerListener(localIPAddress);
// Create the port listener for Connections Multiplexers
createConnectionManagerListener();
// Create the port listener for external components
createComponentListener();
// Create the port listener for clients
createClientListeners();
// Create the port listener for secured clients
createClientSSLListeners();
}
//这个地方只要是建立5个socket,下面介绍一下：
1. createServerListener，此主要是建立一个基于java.net.ServerSocket。
private void createServerListener(String localIPAddress) {
// Start servers socket unless it's been disabled.
if (isServerListenerEnabled()) {
int port = getServerListenerPort();
try {
serverSocketThread = new SocketAcceptThread(this, new ServerPort(port, serverName,
localIPAddress, false, null, ServerPort.Type.server));
ports.add(serverSocketThread.getServerPort());
serverSocketThread.setDaemon(true);
serverSocketThread.setPriority(Thread.MAX_PRIORITY);
}
catch (Exception e) {
System.err.println("Error creating server listener on port " + port + ": " +
e.getMessage());
Log.error(LocaleUtils.getLocalizedString("admin.error.socket-setup"), e);
}
}
}
进入SocketAcceptThread会发现其真面目，
public class SocketAcceptThread extends Thread {
/**
* Holds information about the port on which the server will listen for connections.
*/
private ServerPort serverPort;
private SocketAcceptingMode acceptingMode;
public SocketAcceptThread(ConnectionManager connManager, ServerPort serverPort)
throws IOException {
super("Socket Listener at port " + serverPort.getPort());
// Listen on a specific network interface if it has been set.
String interfaceName = JiveGlobals.getXMLProperty("network.interface");
InetAddress bindInterface = null;
if (interfaceName != null) {
if (interfaceName.trim().length() > 0) {
bindInterface = InetAddress.getByName(interfaceName);
// Create the new server port based on the new bind address
serverPort = new ServerPort(serverPort.getPort(),
serverPort.getDomainNames().get(0), interfaceName, serverPort.isSecure(),
serverPort.getSecurityType(), serverPort.getType());
}
}
this.serverPort = serverPort;
// Set the blocking reading mode to use
acceptingMode = new BlockingAcceptingMode(connManager, serverPort, bindInterface);
}
/**
* Retrieve the port this server socket is bound to.
*
* @return the port the socket is bound to.
*/
public int getPort() {
return serverPort.getPort();
}
/**
* Returns information about the port on which the server is listening for connections.
*
* @return information about the port on which the server is listening for connections.
*/
public ServerPort getServerPort() {
return serverPort;
}
/**
* Unblock the thread and force it to terminate.
*/
public void shutdown() {
acceptingMode.shutdown();
}
/**
* About as simple as it gets. The thread spins around an accept
* call getting sockets and handing them to the SocketManager.
*/
@Override
public void run() {
acceptingMode.run();
// We stopped accepting new connections so close the listener
shutdown();
}
}
class BlockingAcceptingMode extends SocketAcceptingMode {
private static final Logger Log = LoggerFactory.getLogger(BlockingAcceptingMode.class);
protected BlockingAcceptingMode(ConnectionManager connManager, ServerPort serverPort,
InetAddress bindInterface) throws IOException {
super(connManager, serverPort);
serverSocket = new ServerSocket(serverPort.getPort(), -1, bindInterface);
}
/**
* About as simple as it gets. The thread spins around an accept
* call getting sockets and creating new reading threads for each new connection.
*/
@Override
public void run() {
while (notTerminated) {
try {
Socket sock = serverSocket.accept();
if (sock != null) {
Log.debug("Connect " + sock.toString());
SocketReader reader =
connManager.createSocketReader(sock, false, serverPort, true);
Thread thread = new Thread(reader, reader.getName());
thread.setDaemon(true);
thread.setPriority(Thread.NORM_PRIORITY);
thread.start();
}
}
catch (IOException ie) {
if (notTerminated) {
Log.error(LocaleUtils.getLocalizedString("admin.error.accept"),
ie);
}
}
catch (Throwable e) {
Log.error(LocaleUtils.getLocalizedString("admin.error.accept"), e);
}
}
}
}
//此处只要这个服务器，接受到一个请求，就会创建一个新的线程去解析相关数据，并做逻辑处理。
@Override
public void run() {
try {
socketReader.reader.getXPPParser().setInput(new InputStreamReader(
ServerTrafficCounter.wrapInputStream(socket.getInputStream()), CHARSET));
// Read in the opening tag and prepare for packet stream
try {
socketReader.createSession();
}
catch (IOException e) {
Log.debug("Error creating session", e);
throw e;
}
// Read the packet stream until it ends
if (socketReader.session != null) {
readStream();
}
}
catch (EOFException eof) {
// Normal disconnect
}
catch (SocketException se) {
// The socket was closed. The server may close the connection for several
// reasons (e.g. user requested to remove his account). Do nothing here.
}
catch (AsynchronousCloseException ace) {
// The socket was closed.
}
catch (XmlPullParserException ie) {
// It is normal for clients to abruptly cut a connection
// rather than closing the stream document. Since this is
// normal behavior, we won't log it as an error.
// Log.error(LocaleUtils.getLocalizedString("admin.disconnect"),ie);
}
catch (Exception e) {
if (socketReader.session != null) {
Log.warn(LocaleUtils.getLocalizedString("admin.error.stream") + ". Session: " +
socketReader.session, e);
}
}
finally {
if (socketReader.session != null) {
if (Log.isDebugEnabled()) {
Log.debug("Logging off " + socketReader.session.getAddress() + " on " + socketReader.connection);
}
try {
socketReader.session.close();
}
catch (Exception e) {
Log.warn(LocaleUtils.getLocalizedString("admin.error.connection") + socket.toString());
}
}
else {
// Close and release the created connection
socketReader.connection.close();
Log.debug(LocaleUtils.getLocalizedString("admin.error.connection") + socket.toString());
}
socketReader.shutdown();
}
}
//createSession中进行报文的解析，
protected void createSession()
throws UnauthorizedException, XmlPullParserException, IOException {
XmlPullParser xpp = reader.getXPPParser();
for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;) {
eventType = xpp.next();
}
// Check that the TO attribute of the stream header matches the server name or a valid
// subdomain. If the value of the 'to' attribute is not valid then return a host-unknown
// error and close the underlying connection.
System.out.println(reader.getXPPParser().toString());
String host = reader.getXPPParser().getAttributeValue("", "to");
if (validateHost() && isHostUnknown(host)) {
StringBuilder sb = new StringBuilder(250);
sb.append("<?xml version='1.0' encoding='");
sb.append(CHARSET);
sb.append("'?>");
// Append stream header
sb.append("<stream:stream ");
sb.append("from=\"").append(serverName).append("\" ");
sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
sb.append("version=\"1.0\">");
// Set the host_unknown error
StreamError error = new StreamError(StreamError.Condition.host_unknown);
sb.append(error.toXML());
// Deliver stanza
connection.deliverRawText(sb.toString());
// Close the underlying connection
connection.close();
// Log a warning so that admins can track this cases from the server side
Log.warn("Closing session due to incorrect hostname in stream header. Host: " + host +
". Connection: " + connection);
}
// Create the correct session based on the sent namespace. At this point the server
// may offer the client to secure the connection. If the client decides to secure
// the connection then a <starttls> stanza should be received
else if (!createSession(xpp.getNamespace(null))) {
// No session was created because of an invalid namespace prefix so answer a stream
// error and close the underlying connection
StringBuilder sb = new StringBuilder(250);
sb.append("<?xml version='1.0' encoding='");
sb.append(CHARSET);
sb.append("'?>");
// Append stream header
sb.append("<stream:stream ");
sb.append("from=\"").append(serverName).append("\" ");
sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
sb.append("version=\"1.0\">");
// Include the bad-namespace-prefix in the response
StreamError error = new StreamError(StreamError.Condition.bad_namespace_prefix);
sb.append(error.toXML());
connection.deliverRawText(sb.toString());
// Close the underlying connection
connection.close();
// Log a warning so that admins can track this cases from the server side
Log.warn("Closing session due to bad_namespace_prefix in stream header. Prefix: " +
xpp.getNamespace(null) + ". Connection: " + connection);
}
}
将相关信息转换为XMPP报文格式。
2. createConnectionManagerListener();
private void createConnectionManagerListener() {
// Start multiplexers socket unless it's been disabled.
if (isConnectionManagerListenerEnabled()) {
// Create SocketAcceptor with correct number of processors
multiplexerSocketAcceptor = buildSocketAcceptor();
// Customize Executor that will be used by processors to process incoming stanzas
ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("connectionManager");
int eventThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.threads", 16);
ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor) threadModel.getExecutor();
eventExecutor.setCorePoolSize(eventThreads + 1);
eventExecutor.setMaximumPoolSize(eventThreads + 1);
eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);
multiplexerSocketAcceptor.getDefaultConfig().setThreadModel(threadModel);
// Add the XMPP codec filter
multiplexerSocketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
}
}
从 multiplexerSocketAcceptor.bind(new InetSocketAddress(bindInterface, port), new MultiplexerConnectionHandler(serverName));中可以知道其报文处理是MultiplexerConnectionHandler。
MultiplexerConnectionHandler，ClientConnectionHandler，这个3个类都是继承与ConnectionHandler，下面主要讲解ConnectionHandler。
ConnectionHandler继承与IoHandlerAdapter，重写了IoHandlerAdapter的6个方法。
当客户端和服务器建立连接的时候会调用：
@Override
public void sessionOpened(IoSession session) throws Exception {
// Create a new XML parser for the new connection. The parser will be used by the XMPPDecoder filter.
final XMLLightweightParser parser = new XMLLightweightParser(CHARSET);
session.setAttribute(XML_PARSER, parser);
// Create a new NIOConnection for the new session
final NIOConnection connection = createNIOConnection(session);
session.setAttribute(CONNECTION, connection);
session.setAttribute(HANDLER, createStanzaHandler(connection));
// Set the max time a connection can be idle before closing it. This amount of seconds
// is divided in two, as Openfire will ping idle clients first (at 50% of the max idle time)
// before disconnecting them (at 100% of the max idle time). This prevents Openfire from
// removing connections without warning.
final int idleTime = getMaxIdleTime() / 2;
if (idleTime > 0) {
session.setIdleTime(IdleStatus.READER_IDLE, idleTime);
}
}
//当有一个连接被打开的时候，就会初始化XMPP 的解析器，业务数据解析，以及会话空闲时间。
当客户端向服务端发起消息时会调用：
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
// Get the stanza handler for this session
StanzaHandler handler = (StanzaHandler) session.getAttribute(HANDLER);
// Get the parser to use to process stanza. For optimization there is going
// to be a parser for each running thread. Each Filter will be executed
// by the Executor placed as the first Filter. So we can have a parser associated
// to each Thread
int hashCode = Thread.currentThread().hashCode();
XMPPPacketReader parser = parsers.get(hashCode);
if (parser == null) {
parser = new XMPPPacketReader();
parser.setXPPFactory(factory);
parsers.put(hashCode, parser);
}
// Update counter of read btyes
updateReadBytesCounter(session);
//System.out.println("RCVD: " + message);
// Let the stanza handler process the received stanza
try {
handler.process((String) message, parser);
} catch (Exception e) {
Log.error("Closing connection due to error while processing message: " + message, e);
Connection connection = (Connection) session.getAttribute(CONNECTION);
connection.close();
}
}
public void process(String stanza, XMPPPacketReader reader) throws Exception {
boolean initialStream = stanza.startsWith("<stream:stream") || stanza.startsWith("<flash:stream");
if (!sessionCreated || initialStream) {
if (!initialStream) {
// Allow requests for flash socket policy files directly on the client listener port
if (stanza.startsWith("<policy-file-request/>")) {
String crossDomainText = FlashCrossDomainServlet.CROSS_DOMAIN_TEXT +
XMPPServer.getInstance().getConnectionManager().getClientListenerPort() +
FlashCrossDomainServlet.CROSS_DOMAIN_END_TEXT + '\0';
connection.deliverRawText(crossDomainText);
return;
}
else {
// Ignore <?xml version="1.0"?>
return;
}
}
// Found an stream:stream tag...
if (!sessionCreated) {
sessionCreated = true;
MXParser parser = reader.getXPPParser();
parser.setInput(new StringReader(stanza));
createSession(parser);
}
else if (startedTLS) {
startedTLS = false;
tlsNegotiated();
}
else if (startedSASL && saslStatus == SASLAuthentication.Status.authenticated) {
startedSASL = false;
saslSuccessful();
}
else if (waitingCompressionACK) {
waitingCompressionACK = false;
compressionSuccessful();
}
return;
}
// Verify if end of stream was requested
if (stanza.equals("</stream:stream>")) {
session.close();
return;
}
// Ignore <?xml version="1.0"?> stanzas sent by clients
if (stanza.startsWith("<?xml")) {
return;
}
// Create DOM object from received stanza
Element doc = reader.read(new StringReader(stanza)).getRootElement();
if (doc == null) {
// No document found.
return;
}
String tag = doc.getName();
if ("starttls".equals(tag)) {
// Negotiate TLS
if (negotiateTLS()) {
startedTLS = true;
}
else {
connection.close();
session = null;
}
}
else if ("auth".equals(tag)) {
// User is trying to authenticate using SASL
startedSASL = true;
// Process authentication stanza
saslStatus = SASLAuthentication.handle(session, doc);
}
else if (startedSASL && "response".equals(tag)) {
// User is responding to SASL challenge. Process response
saslStatus = SASLAuthentication.handle(session, doc);
}
else if ("compress".equals(tag)) {
// Client is trying to initiate compression
if (compressClient(doc)) {
// Compression was successful so open a new stream and offer
// resource binding and session establishment (to client sessions only)
waitingCompressionACK = true;
}
}
else {
process(doc);
}
}
//此过程主要的工作是：
1.在调用此方法之前，会调用其编码解码器（XMPPDecoder），对来的报文进行解码。
2.获取XML解析对象XMPPPacketReader，统计接受的数据量，对数据进行解码。
3.根据不同的消息头做不同的处理
此篇文章就到此，稍后会有更多关于openfire的个人解读。
联系方式（qq）:851392159
相关推荐
这一框架已被广泛应用于多个领域，如Red5项目中的RTMP协议实现、SubEtha.SMTP项目中的SMTP协议、Apache Qpid项目中的AMQP协议，以及Openfire项目中的XMPP协议等。这些实例充分展示了Mina框架在不同场景下的强大适用...
目前正在使用 MINA 的软件包括有：Apache Directory Project、AsyncWeb、AMQP（Advanced Message Queuing Protocol）、RED5 Server（Macromedia Flash Media RTMP）、ObjectRADIUS、Openfire 等等。
Openfire采用了Apache MINA框架来处理网络通信，MINA是一个强大的网络应用程序框架，旨在简化高性能和高可靠性网络应用的开发。MINA通过Java NIO提供跨不同传输类型（如TCP/IP、UDP/IP）的统一API，同时支持串口通讯...
Openfire是一个开源的即时通讯（IM）服务器，它基于XMPP（Extensible Messaging and Presence Protocol）协议，该协议使用XML进行消息交换，以实现高度可扩展的通信。XMPP是一种标准化的协议，允许开发者构建即时...
5. Apache MINA是一个提供高性能和高可靠性网络应用框架的Java库，它使用Java NIO框架来实现网络通信。MINA能够为TCP/IP、UDP/IP等不同类型的传输提供统一的事件驱动的异步API。它还提供了过滤器功能、多线程模型、...
具体来说，Openfire中的`ConnectionHandler`类继承自MINA的`IoHandlerAdapter`类，负责客户端连接的创建与销毁，以及XML数据包的传递。此外，Openfire还根据连接类型的不同，进一步细分为`ClientConnectionHandler`...
通过使用MINA框架可以可以省下处理底层I/O和线程并发等复杂工作，开发人员能够把更多的精力投入到业务设计和开发当中。MINA框架的应用比较广泛，应用的开源项目有Apache Directory、AsyncWeb、Apache Qpid、QuickFIX...
通过以上详细介绍，我们可以看到OpenFire二次开发涉及的技术栈相当广泛，包括网络编程、Mina框架的应用以及XMPP协议的理解等多个方面。对于开发者而言，熟悉这些技术和概念是进行有效二次开发的基础。
通过使用MINA框架可以可以省下处理底层I/O和线程并发等复杂工作，开发人员能够把更多的精力投入到业务设计和开发当中。MINA框架的应用比较广泛，应用的开源项目有Apache Directory、AsyncWeb、Apache Qpid、QuickFIX...
在使用Openfire的过程中，可能会遇到一些依赖库或jar包缺失的问题，这将导致服务无法正常启动或功能受限。标题提到的是"openfire3.10.2缺少的jar包"，描述中指出这些jar包是最新的，日期为2015年10月27日。从提供的...
openfire3.10.1源码编译缺少的jar包: jetty-websocket-8.1.17.v20150415 jnsapi mina-core-1.1.7-sources.jar mina-filter-compression-1.1.7-sources.jar mina-filter-ssl-1.1.7-sources.jar mina-filter-ssl.jar ...
《Openfire 3.10.2在Eclipse中的集成与缺失JAR包解析》 Openfire是一款开源的即时通讯服务器软件，它基于XMPP协议，为企业或组织提供了一个强大的、可扩展的通信平台。在将Openfire项目导入Eclipse进行开发时，可能...
Openfire 是一款基于 XMPP（Extensible Messaging and Presence Protocol）协议的即时通讯（IM）服务器，它使用 MINA 库的 Java NIO 实现，通常搭配 MySQL 数据库使用。默认情况下，Openfire 官方宣称可支持约 5000 ...
Openfire 的通信处理基于 Apache MINA 框架实现，提供了一个通过 Java NIO 在不同的传输例如 TCP/IP 和 UDP/IP 上抽象的事件驱动的异步 API。 Openfire 中常见的类名后缀命名包括 Starter、Plugin、Listener、...
在Openfire的运行环境中，Java Archive (JAR) 文件起着至关重要的作用，它们包含了必要的类库和资源，使得Openfire能够正常运行。本文将深入探讨"openfire所需要的jar包"以及与之相关的知识点。 首先，我们关注到的...
Apache MINA 是 Openfire 采用的网络应用程序框架，它为开发者提供了异步、事件驱动的网络编程接口，支持多种传输协议，如 TCP/IP、UDP/IP、串口通信以及虚拟机内部通信。MINA 的特性包括统一的 API、过滤器机制、...
在Openfire中，Apache MINA被用作网络应用程序框架，它提供了一种异步事件驱动的NIO（非阻塞I/O）机制，简化了高性能、高可靠性的网络应用开发。MINA支持多种传输协议，包括TCP/IP、UDP/IP，甚至串口通信，并且具备...