tigase源码分析7:用户连接登录流程

//在看本节之前一定要先了解下xmpp协议，建议仔细看下 tigase源码分析6:了解xmpp协议


//在看下面代码之前，要知道，每一个用户User通过某一资源连接到服务器时，
 //每一个User在不同的资源上登录都各对应着一个IOService，
 //每一个资源对应着一个XMPPResourceConnection,
 //同一个用户User多个XMPPResourceConnection可能共同引用着同一个XMPPSession
public IOService<?> IOService.call() throws IOException {
......
//当socket有数据要处理的时候，进行解析
processSocketData();
if ((receivedPackets() > 0) && (serviceListener != null)) {
	serviceListener.packetsReady(this);
}    // end of if (receivedPackets.size() > 0)
.....
}


protected void XMPPIOService.processSocketData() throws IOException {
.....  
       //解析socket数据
	parser.parse(domHandler, data, 0, data.length);
....

}


public final void SimpleParser.parse(SimpleHandler handler, char[] data, int off, int len) {
.....
 //遇到<stream：stream>
handler.startElement(parser_state.element_name, null, null);
......
}


public void XMPPDomBuilderHandler.startElement(StringBuilder name, StringBuilder[] attr_names,
			StringBuilder[] attr_values) {
......
       //服务端也打开一个对应的<stream:stream>
	service.xmppStreamOpened(attribs);
........
}

protected void XMPPIOService.xmppStreamOpened(Map<String, String> attribs) {
...
String response = serviceListener.xmppStreamOpened(this, attribs);
...
}


public String ClientConnectionManager.xmppStreamOpened(XMPPIOService<Object> serv,
 Map<String,String> attribs) {
....

if (id == null) {
//生成一些属性
 id = UUID.randomUUID().toString();
 serv.getSessionData().put(IOService.SESSION_ID_KEY, id);
 serv.setXMLNS(XMLNS);
 serv.getSessionData().put(IOService.HOSTNAME_KEY, hostname);
 serv.setDataReceiver(JID.jidInstanceNS(routings.computeRouting(hostname)));

 String streamOpenData = prepareStreamOpen(serv, id, hostname);
//给客户端回一个<stream:stream>告诉他服务端也打开了stream
 writeRawData(serv, streamOpenData);
//生成一个新的iq请求packet，主要作用是通知打开session connection
Packet streamOpen = Command.STREAM_OPENED.getPacket(serv.getConnectionId(), serv
 .getDataReceiver(), StanzaType.set, this.newPacketId("c2s-"), Command.DataType.submit);
//设置一些属性
Command.addFieldValue(streamOpen, "session-id", id);
Command.addFieldValue(streamOpen, "hostname", hostname);
Command.addFieldValue(streamOpen, "xml:lang", lang);
//把刚新生成的packet投递到MessageRouter去路由到目的地
addOutPacketWithTimeout(streamOpen, startedHandler, 45l, TimeUnit.SECONDS);
}

       

 

//这是上面生成的一个iq command packet
<iq from="c2s@dell-pc/192.168.3.10_5222_192.168.3.10_53597" type="set" id="c2s--c2s5"
     to="sess-man@dell-pc">
  <command node="STREAM_OPENED" xmlns="http://jabber.org/protocol/commands">
    <x type="submit" xmlns="jabber:x:data">
         <field var="session-id">
          <value>4f0e26c9-aeac-442e-aaea-84c788ab73d2</value>
         </field>
          <field var="hostname">
            <value>192.168.3.10</value>
          </field>
          <field var="xml:lang">
          <value>en</value>
          </field>
    </x>
   </command>
</iq>

 

 

   

//packet被路由到SessionManager后，由继承的QueueListener线程进行处理
QueueListener为内部类，所以他能访问外部类的方法
 public void QueueListener.run() {
  .........
//由于属于command，所以进入以下代码块
  if (packet.isCommand() && (packet.getStanzaTo() != null)
  && compName.equals(packet.getStanzaTo().getLocalpart())
   && isLocalDomain(packet.getStanzaTo().getDomain())) {
	processed = processScriptCommand(packet, results);
	if (processed) {
	Packet result = null;

	while ((result = results.poll()) != null) {
	  addOutPacket(result);
	}
	}
  }
   if (!processed && ((packet = filterPacket(packet, incoming_filters)) != null)) {
	processPacket(packet);//此方法是执行真正的实现类的方法，
    }
.........
}

//再执行到processCommand
public void SessionManager.processPacket(final Packet packet) {
	//为command，则执行processCommand
	 if (packet.isCommand() && processCommand(packet)) {
			packet.processedBy("SessionManager");

			// No more processing is needed for command packet
			return;
		}    // end of if (pc.isCommand())

		XMPPResourceConnection conn = getXMPPResourceConnection(packet);

...
		processPacket(packet, conn);
	}


//private SessionOpenProc   sessionOpenProc  = null;
protected boolean SessionManager.processCommand(Packet pc) {
....
Iq      iqc= (Iq) pc;
XMPPResourceConnection connection = connectionsByFrom.get(iqc.getFrom());
  switch (iqc.getCommand()) {
....
case STREAM_OPENED : {		
 //获取session processor的处理线程集
  ProcessingThreads<ProcessorWorkerThread> pt = workerThreads.get(sessionOpenProc.id());
   if (pt == null) {
      pt = workerThreads.get(defPluginsThreadsPool);
    }
    //把packet交给session processor插件来进行下一步的处理，它是运行在单独的线程上的。
    pt.addItem(sessionOpenProc, iqc, connection);
    processing_result = true;
  }



public void SessionOpenProc.process(Packet packet, XMPPResourceConnection session,
	NonAuthUserRepository repo, Queue<Packet> results, Map<String, Object> settings)
				throws XMPPException {
...
 //每一个客户端都会生成一个对应的XMPPResourceConnection 资源连接器，它持有XMPPSession的引用
  conn = createUserSession(packet.getFrom(), hostname);
  conn.setSessionId(Command.getFieldValue(packet, "session-id"));
  conn.setDefLang(Command.getFieldValue(packet, "xml:lang"));
..
 //回一个应答packet
  fastAddOutPacket(packet.okResult((String) null, 0));
}


protected XMPPResourceConnection SessionManager.createUserSession(JID conn_id, String domain) throws TigaseStringprepException {
  XMPPResourceConnection connection = new XMPPResourceConnection(conn_id,
				user_repository, auth_repository, this);
//放进SessionManager.connectionsByFrom中，以便在processPacket(..)中得到相关packet的conn
   connectionsByFrom.put(conn_id, connection);
  return connection;
 }

 

<iq from="c2s@dell-pc/192.168.3.10_5222_192.168.3.10_64739" type="get" id="ead3ca01-9469-414f-9f0a-a7a52c164a72" to="sess-man@dell-pc"><command node="GETFEATURES" xmlns="http://jabber.org/protocol/commands"/>
</iq>

<iq from="sess-man@dell-pc" type="result" id="ead3ca01-9469-414f-9f0a-a7a52c164a72" to="c2s@dell-pc/192.168.3.10_5222_192.168.3.10_64739"><command node="GETFEATURES" xmlns="http://jabber.org/protocol/commands"><ver xmlns="urn:xmpp:features:rosterver"/></command>
</iq>

<iq from="sess-man@dell-pc" type="result" id="73c47636-536c-4f6b-a306-8ff912313497" to="c2s@dell-pc/192.168.3.10_5222_192.168.3.10_49257"><command node="GETFEATURES" xmlns="http://jabber.org/protocol/commands"><ver xmlns="urn:xmpp:features:rosterver"/><starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"/><mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl"><mechanism>PLAIN</mechanism><mechanism>ANONYMOUS</mechanism></mechanisms><register xmlns="http://jabber.org/features/iq-register"/><compression xmlns="http://jabber.org/features/compress"><method>zlib</method></compression><auth xmlns="http://jabber.org/features/iq-auth"/></command>
</iq>

 

<!-- 客户端发来认证请求-->
<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">ADc4OTQ1NgA3ODk0NTY=</auth>
<!--服务端产生的对应的iq 作为流转命令，不返回给客户端的-->
<iq from="sess-man@dell-pc" type="set" id="tig2" to="c2s@dell-pc/192.168.3.10_5222_192.168.3.10_49257"><command node="USER_LOGIN" xmlns="http://jabber.org/protocol/commands"><x type="submit" xmlns="jabber:x:data"><field var="user-jid"><value>789456@192.168.3.10</value></field></x></command></iq>
<!-- 认证成功返回给客户端的-->
<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>

   

   当服务端接收到<auth>认证请求后，执行SaslAuth 认证处理通过以后，则生成用户对应的XMPPSession

 

public void SaslAuth.process(final Packet packet, final XMPPResourceConnection session,final NonAuthUserRepository repo, final Queue<Packet> results, final Map<String, Object> settings) {
.....
if (ss.isComplete() && (ss.getAuthorizationID() != null)) {
 .....
//检查有没有XMPPSession，没有则创建一个新的绑定相关的属性
 session.authorizeJID(jid, anonymous);
//返回一个<success>成功标志的packet
results.offer(packet.swapFromTo(createReply(ElementType.success,challengeData),null,null));

}

  public void XMPPResourceConnection.authorizeJID(BareJID jid, boolean anonymous) {
		authState    = Authorization.AUTHORIZED;
		is_anonymous = anonymous;
		loginHandler.handleLogin(jid, this);
		login();
	}

    public void SessionManager.handleLogin(BareJID userId, XMPPResourceConnection conn) {
	registerNewSession(userId, conn);
	}

    protected void SessionManager.registerNewSession(BareJID userId, XMPPResourceConnection conn) {
   .....
       //一个用户在不同一资源上登录，共用这个xmppsession
       XMPPSession session = sessionsByNodeId.get(userId);
         if (session == null) {
	    session = new XMPPSession(userId.getLocalpart());
	    sessionsByNodeId.put(userId, session);
      ....
         } else {
        // Check all other connections whether they are still alive....
        //检查session.activeResources中其它的XMPPResourceConnection是否有效的
        List<XMPPResourceConnection> connections = session.getActiveResources();
        .........
       }
       //session和connection相互关联起来，双方都持有对方的引用
       session.addResourceConnection(conn);

if ((!"USER_STATUS".equals(conn.getSessionId())) &&!conn.isServerSession() &&!conn
						.isTmpSession()) {
  //生成一个USER_LOGIN 事件的命令packet，好让生成的jid关联到用户的IOService上		
 Packet user_login_cmd = Command.USER_LOGIN.getPacket(getComponentId(), 
  conn.getConnectionId(), StanzaType.set, conn.nextStanzaId(), Command.DataType.submit);
  Command.addFieldValue(user_login_cmd, "user-jid", userId.toString());
  ddOutPacket(user_login_cmd);
}
   .....
 }

 

 

 

 

protected void ClientConnectionManager.processCommand(Packet packet) {
		XMPPIOService<Object> serv = getXMPPIOService(packet);
switch (iqc.getCommand()) {
case GETFEATURES :
..
break;
case USER_LOGIN :
  String jid = Command.getFieldValue(iqc, "user-jid");
 .....
  serv.setUserJid(jid);
}

       下一步，客户端会再次打开流，服务端也会打开一个新的流，这时客户端会请求绑定资源名称

<iq type="set" id="bind_1">
<bind xmlns="urn:ietf:params:xml:ns:xmpp-bind">
<resource>DELL-PC</resource>
</bind>
</iq>

    服务端 BindResource （绑定插件）会处理这个packet，所以只要明白，tigase都是基于插件和组件组合来处理请求的，不同的插件来处理不同的请求，我们也可以开发相关的插件来处理我们自定义的请求了。