tigase源码分析5:SessionManager

jianfulove

浏览: 120237 次
性别:
来自: 湛江

最近访客更多访客>>

zhd214

hy811014

dyhjjk

hh_qq_love_hi

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

tigase

这节主要是分析数据包packet是如何被SessionManager和插件处理的，首先分析一些开发的理论知识先：

一、 Tigase服务器插件开发重要的是要了解它是如何工作的。不同类型的插件负责处理数据流中不同阶段的数据包（packet）。

在Tigase服务器中，插件代码负责处理特定的XMPP节。一个单独的插件可能会负责处理<message>,另一个用于处理<presence>,和还有些插件是单独负责<iq>处理，还有些处理不同版本等等。

一个插件应该提供它感兴趣信息，如哪些XML元素节点，和叫什么名，它具有什么样xmlns名称空间等，这将使得插件在sm中匹配到相对应的packet，从而可以对匹配的packet进行处理。所以你可以创建一个插件来处理包含有感兴趣的caps子节点的packet。也可能都没有一个插件处理某个特定xml节,然而系统会默认的行为是简单转发这个xml节到目的地址。也可能有一个以上的插件处理一个特定的XML元素,然后他们在单独的线程上同时处理同一xml节,所以不保证的不同的插件能按顺序处理xml节。

 //一个插件要处理哪些xml元素，是通过以下两个方法来告知系统他将处理那些它感兴趣的xml元素的。
         @Override
	public String[][] supElementNamePaths() {
		return ELEMENTS; //感兴趣的xml元素。如<message> <iq>
	}

	@Override
	public String[] supNamespaces() {
		return XMLNSS; //supElementNamePaths中一一对应的xmlns
	}

每个packet通过SessionManager组件时被处理的几个步骤：

照片显示,每个packet 通过SessionManager处理的4个步骤：

1，Pre-processing

为了不影响Session Manager的性能需要限制该方法处理时间为极小值，用于判断当前package是否应该被阻塞，如果返回为true，则表示阻塞。（只要有一个pre-processor阻塞就算阻塞）

2，Processing

如果一个Package没有被任何的pre-processors阻塞，则才继续执行该方法。所有对当前XML段感兴趣的processor都会将该段加入到独立的线程里运行，这些现成使用内部固定的队列。当所有感兴趣的processor都执行完后就可以得到通知进入下一步。

3，post-processor

对于在第2步中没有被任何processor处理的package将会通过所有的post-processors，并被最后一个post-processor转发到一个目的地，大多数情况是以<message/>的形式被转发。例如MessageAmp.postProcess(..)，它判断用户不在线时，massage信息将被保存入数据库。

4，filter

对于以上三步任何形式的结果result输出，都会被所有的filters拦截过滤，这些结果可能最终被拦截也可能被放行。

由于session manager和processors都属于消费者，所以在所有的processors中应该至少有一个processor复制一个新package并发送给某个目标。当然processor处理过程中可以生成任意数量的packet。以上4个步骤任何一个processor都可以生成packet作为结果返出。

看看以下的图片：

如果UserA发送packet-P1到服务器以外的其它节点,例如另一个服务器的用户或一些组件(Muc,PubSub),那么其中一个processor必须新创建一个副本P2的packet并正确地设置所有属性和目的地地址。packet-P1 在被SessionManager处理后消毁了，然而某一个插件生成了一个新的packet-P2。

当然同理，packet从组件转发给用户：

packet的处理组件或插件，必须生成新packet的副本提供给用户。当然，如果没有被任何插件处理的package将会通过系统默认方式转发到一个目的地，注意实现这种方式,因为输入数据包P1可以被许多插件同时处理，因此实际上packet一旦到了SessionManager进行处理应该不得变它。

很明显，下图的处理流程是当用户发送请求到服务器和期望服务器的响应：

这里有一个令人惊讶的设计结果。如果你看看下图显示2个用户之间的交流，可以看到packet送到最终目的地前复制了两次：

上图方式的packet必须被Session Manager处理两次。处理第一次代表packet作为一个即将离开用户A的包和第二次处理它代表传入用户B的packet。这是确保用户A有权限发送一个packet,确保用户B有权利接收一个packet。另外,如果用户B是离线或脱机的，<message> processors应该把packet信息保存到一个数据库。

二、当需要编写一个自己的PLUGIN的时候根据以上SM分析可以知道，

可以按需求来分别实现以下四个接口，可以实现一个也可以实现多个：
1,XMPPPreprocessorIfc
2,XMPPProcessorIfc
3,XMPPPostprocessorIfc
4,XMPPPacketFilterIfc
这四个接口各需要实现一个简单的方法，每个方法的参数类似，参数描述如下：
-Packet packet 需要被处理的PACKET，该PACKET不能为NULL虽然在PROCESS处理过程中无法修改它
-XMPPResourceConnection session 用于保存所有用户的数据，它提供权限访问用户的仓库数据，在没有在线用户SESSION的情况下该参数可以为NULL
-NonAuthUserRepository repo 该参数往往在参数session为NULL的时候被使用，它用于为不在线的用户保存私有或公开的数据信息。
-Queue<Packet> results 这个为输入packet的处理结果产生的packet集合，它总被要求一定要存放一个输入数据包packet的副本到里面，其实包含了所有需要进一步处理的packet,包括process生成的结果packet。

-Map<String, Object> settings 为PLUGIN制定配置信息，一般情况下不需要使用，然而如果需要访问额外的数据库则可以通过配置文件将数据库连接字符串传给plugin

以下就是具体的代码分析了：

SessionManager

public void SessionManager.processPacket(final Packet packet) {

	if (packet.isCommand() && processCommand(packet)) {
	  packet.processedBy("SessionManager");

	}    // end of if (pc.isCommand())
        //这个方法查找相对应的connection，这方法很重要，from-to ,to-from思想。后面详细分析
	XMPPResourceConnection conn = getXMPPResourceConnection(packet);

	if ((conn == null) && (isBrokenPacket(packet)) || processAdminsOrDomains(packet)) {
		return;
	}
	processPacket(packet, conn);
}

protected XMPPResourceConnection getXMPPResourceConnection(Packet p) {
		XMPPResourceConnection conn = null;
      //首先先根据发用户发送的packet信息来找连接session，这样的设计正如前面理论中所指出的，发用户是否有权利发packet，发用户A的packet到了SM这时处理为一个阶段，这个packet在这一阶段处理完成就会被消毁了，经过processor处理会生成该packet的副本继续往后面流转
		JID   from = p.getPacketFrom();

		if (from != null) {
			conn = connectionsByFrom.get(from);
			if (conn != null) {
				return conn;
			}
		}

		// It might be a message _to_ some user on this server
		// so let's look for established session for this user...
     //如果packet没有含有发用户接连信息则查询接收目标用户连接session，这为下一阶段-这是经过某一个processor处理后转发出来的packet，也就是说packet由系统转发到目的地阶段，判断目的地 接收用户B 是否在线是否有连接在在，如果在线则packet再次经过processor就会转发到用户B了，如果不在线，则最终会被保存入库。
		JID to = p.getStanzaTo();

		if (to != null) {
			conn = getResourceConnection(to);
		} else {	
		}    // end of else

		return conn;
	}

     public XMPPResourceConnection getResourceConnection(JID jid) {
		XMPPSession session = getSession(jid.getBareJID());

		if (session != null) {
			XMPPResourceConnection res = session.getResourceConnection(jid);
			}

			return res;
		}    // end of if (session != null)

		// Maybe this is a call for the server session?
		if (isLocalDomain(jid.toString(), false)) {
			return smResourceConnection;
		}

		return null;
	}

protected void processPacket(Packet packet, XMPPResourceConnection conn) {
		long startTime = System.currentTimeMillis();
		int  idx       = tIdx;

		tIdx = (tIdx + 1) % maxIdx;

		packet.setPacketTo(getComponentId());
		

		Queue<Packet> results = new ArrayDeque<Packet>(2);
		boolean       stop    = false;

	if (!stop) {
	  if (defPacketHandler.preprocess(packet, conn, naUserRepository, results)) {
		packet.processedBy("filter-foward");
                  
		 addOutPackets(packet, conn, results);

		 return;
		 }
	   }

	
	if (!stop) {
	for (XMPPPreprocessorIfc preproc : preProcessors.values()) {
	stop |= preproc.preProcess(packet, conn, naUserRepository, results, plugin_config
						.get(preproc.id()));
		if (stop && log.isLoggable(Level.FINEST)) {
					break;
				}
			}    // end of for (XMPPPreprocessorIfc preproc: preProcessors)
		}

		// prepTm = System.currentTimeMillis() - startTime;
		if (!stop) {
		if (defPacketHandler.forward(packet, conn, naUserRepository, results)) {
				packet.processedBy("filter-foward");
				addOutPackets(packet, conn, results);

				return;
			}
		}

		// defForwTm = System.currentTimeMillis() - startTime;
      if (!stop) {
	walk(packet, conn);
	try {
	if ((conn != null) && conn.getConnectionId().equals(packet.getPacketFrom())) {
		handleLocalPacket(packet, conn);
	}
	} catch (NoConnectionIdException ex) { 
        }
     }

	
	if (!stop) {
	  for (XMPPPostprocessorIfc postproc : postProcessors.values()) {
		String plug_id      = postproc.id();
		long[] postProcTime = null;
		synchronized (postTimes) {
			postProcTime = postTimes.get(plug_id);
			if (postProcTime == null) {
				postProcTime = new long[maxIdx];
				postTimes.put(plug_id, postProcTime);
			}
			}

			long stTime = System.currentTimeMillis();

	              postproc.postProcess(packet, conn, naUserRepository, results, 
                                         plugin_config.get(postproc.id()));
				postProcTime[idx] = System.currentTimeMillis() - stTime;
			}    // end of for (XMPPPostprocessorIfc postproc: postProcessors)
		}      // end of if (!stop)

		// postTm = System.currentTimeMillis() - startTime;
		if (!stop &&!packet.wasProcessed() && ((packet.getStanzaTo() == null) ||
				(!isLocalDomain(packet.getStanzaTo().toString())))) {
			if (defPacketHandler.canHandle(packet, conn)) {
		           ProcessingThreads<ProcessorWorkerThread> pt = workerThreads.get(                            defHandlerProc.id());
                         if (pt == null) {
				pt = workerThreads.get(defPluginsThreadsPool);
			   }
				pt.addItem(defHandlerProc, packet, conn);
				packet.processedBy(defHandlerProc.id());
			}
		}
		setPermissions(conn, results);
		addOutPackets(packet, conn, results);
	if (packet.wasProcessed() || processAdminsOrDomains(packet)) {
			
			Packet error = null;

        if (stop || ((conn == null) && (packet.getStanzaFrom() != null) && (packet
	.getStanzaTo() != null) &&!packet.getStanzaTo().equals(getComponentId()) &&
	((packet.getElemName() == Iq.ELEM_NAME) || (packet.getElemName() == Message
					.ELEM_NAME)))) {
	try {
	error = Authorization.SERVICE_UNAVAILABLE.getResponseMessage(packet,
							"Service not available.", true);
	} catch (PacketErrorTypeException e) {
		
		}    // end of else

	}

private void walk(final Packet packet, final XMPPResourceConnection connection) {

  for (XMPPProcessorIfc proc_t : processors.values()) {
      XMPPProcessorIfc processor = proc_t;
      Authorization    result    = processor.canHandle(packet, connection);

      if (result == Authorization.AUTHORIZED) {
	ProcessingThreads<ProcessorWorkerThread> pt = workerThreads.get(processor.id());
       //获取processor的处理线程，如果预先没有放入，则使用默认的处理线程集处理
      if (pt == null) {
	pt = workerThreads.get(defPluginsThreadsPool);
	}
       //加入待处理队列中，则用单独线程去执行process()这样可以提高效率，
       //而不是用sessionmanager线程执行
       if (pt.addItem(processor, packet, connection)) {
	packet.processedBy(processor.id());
	} else {
	if (result != null) {

	// TODO: A plugin returned an error, the packet should be bounced back
	// with an appropriate error
	}
			
  }    // end of for ()
}

public abstract class WorkerThread extends Thread {
private LinkedBlockingQueue<QueueItem> queue = null;
public void setQueueMaxSize(int maxSize) {
		LinkedBlockingQueue<QueueItem> oldQueue = queue;

		queue = new LinkedBlockingQueue<QueueItem>(maxSize);

		if (oldQueue != null) {
			queue.addAll(oldQueue);
		}
	}
public void run() {
		QueueItem item = null;

		while ( !stopped) {
			try {
				item = queue.take();

				long start = System.currentTimeMillis();

				process(item);

				long end = System.currentTimeMillis() - start;

				if (end > 0) {
         averageProcessingTime = (averageProcessingTime + end) / 2;
				}
			} catch (Exception e) {
				
			}

			++runsCnt;
		}
	}



}

private class ProcessorWorkerThread  extends WorkerThread {
private ArrayDeque<Packet> local_results = new ArrayDeque<Packet>(100);

public void process(QueueItem item) {
	XMPPProcessorIfc processor = item.getProcessor();

	try {
	processor.process(item.getPacket(), item.getConn(), naUserRepository,
					local_results, plugin_config.get(processor.id()));
	if (item.getConn() != null) {
		setPermissions(item.getConn(), local_results);
	}
		addOutPackets(item.getPacket(), item.getConn(), local_results);
	   } catch (PacketErrorTypeException e) {

		} catch (XMPPException e) {
		
	   }
        }
}

public class ProcessingThreads<E extends WorkerThread> {
  private int numWorkerThreads = 1;
  private ArrayList<E> workerThreads = null;
public ProcessingThreads(E worker, int numWorkerThreads, int maxQueueSize, String name)
			throws ClassNotFoundException, InstantiationException, IllegalAccessException {

		
		this.numWorkerThreads = numWorkerThreads;
		workerThreads = new ArrayList<E>(numWorkerThreads);
		this.name = name;

		for (int j = 0; j < numWorkerThreads; j++) {
			WorkerThread t = worker.getNewInstance();

			t.setQueueMaxSize(maxQueueSize);
			t.setDaemon(true);
			t.setName(name + " Queue Worker " + j);
			t.start();
			workerThreads.add((E) t);
			
		}

//  }
	}

public boolean addItem(XMPPProcessorIfc processor, Packet packet, XMPPResourceConnection conn) {
	boolean ret = false;
	QueueItem item = new QueueItem(processor, packet, conn);

	try {
	if ((item.getConn() != null) && item.getConn().isAuthorized()) {

				// Queueing packets per user...
	ret = workerThreads.get(Math.abs(conn.getJID().getBareJID().hashCode())
						% numWorkerThreads).offer(item);


	} else {
	if (packet.getPacketFrom() != null) {

	 // Queueing packets per user's connection...
	ret = workerThreads.get(Math.abs(packet.getPacketFrom().hashCode())
							% numWorkerThreads).offer(item);
	} else {

	// Otherwise per destination address
	// If the packet elemTo is set then used it, otherwise just packetTo:
      if (packet.getStanzaTo() != null) {
	ret = workerThreads.get(Math.abs(packet.getStanzaTo().getBareJID().hashCode())
							% numWorkerThreads).offer(item);
	} else {
	ret = workerThreads.get(Math.abs(packet.getTo().hashCode())).offer(item);

	 }
       }
    }
    ..........
 }//end of addItem

}

SessionManager

//
private Map<String, ProcessingThreads<ProcessorWorkerThread>> workerThreads= 
       new ConcurrentHashMap<String, ProcessingThreads<ProcessorWorkerThread>>(32);
//
private Map<String, XMPPPresenceUpdateProcessorIfc> presenceProcessors =
			new ConcurrentHashMap<String, XMPPPresenceUpdateProcessorIfc>();
//
private Map<String, XMPPPreprocessorIfc> preProcessors = new ConcurrentHashMap<String,
			XMPPPreprocessorIfc>(10);
//
private Map<String, XMPPProcessorIfc> processors = new ConcurrentHashMap<String,
			XMPPProcessorIfc>(32);
//
private Map<String, XMPPPostprocessorIfc> postProcessors =
			new ConcurrentHashMap<String, XMPPPostprocessorIfc>(10);
//
private Map<String, XMPPPacketFilterIfc> outFilters = new ConcurrentHashMap<String,
			XMPPPacketFilterIfc>(10);

public void SessionManager.setProperties(Map<String, Object> props) throws ConfigurationException {
   super.setProperties(props);
     .......
if (!isInitializationComplete()) {
  String sm_threads_pool = (String) props.get(SM_THREADS_POOL_PROP_KEY);

  if (!sm_threads_pool.equals(SM_THREADS_POOL_PROP_VAL)) {
	String[] threads_pool_params = sm_threads_pool.split(":");
	int      def_pool_size       = 100;
	if (threads_pool_params.length > 1) {
	try {
	  def_pool_size = Integer.parseInt(threads_pool_params[1]);
	} catch (Exception e) {
	def_pool_size = 100;
     }
    }
   //设置默认线程沲default-threads-pool对应的处理线程
   ProcessorWorkerThread worker = new ProcessorWorkerThread();
   ProcessingThreads<ProcessorWorkerThread> pt =
             new ProcessingThreads<ProcessorWorkerThread>(worker, def_pool_size,
						maxQueueSize, defPluginsThreadsPool);

	      workerThreads.put(defPluginsThreadsPool, pt);

	}
       }
    //--------
	String[] plugins = SessionManagerConfig.getActivePlugins(props);
	for (String plug_id : plugins) {
	keys.remove(plug_id);
        //根据配置文件加载插件，在这过程中也设定了插件的处理线程
	XMPPImplIfc plugin = addPlugin(plug_id, plugins_concurrency.get(plug_id));
	if (plugin != null) {
	Map<String, Object> plugin_settings = getPluginSettings(plug_id, props);
	if (plugin_settings.size() > 0) {
	plugin_config.put(plug_id, plugin_settings);
					}
	try {
	       plugin.init(plugin_settings);
	} catch (TigaseDBException ex) {

	}
       }
    }    // end of for (String comp_id: plugins)
  .......
}

public XMPPImplIfc SessionManager.addPlugin(String plug_id, String conc).. {
		XMPPImplIfc      result = null;
		XMPPProcessorIfc proc   = null;
		String           version;

		if (plug_id.equals(sessionOpenProcId)) {
			sessionOpenProc = new SessionOpenProc();
			proc            = sessionOpenProc;
		}
		if (plug_id.equals(sessionCloseProcId)) {
			sessionCloseProc = new SessionCloseProc();
			proc             = sessionCloseProc;
		}
		if (plug_id.equals(defaultHandlerProcId)) {
			defHandlerProc = new DefaultHandlerProc();
			proc           = defHandlerProc;
		}
		if (proc == null) {
			proc = SessionManagerConfig.getProcessor(plug_id);
		}

		boolean loaded = false;
		if (proc != null) {
			int threadsNo = proc.concurrentQueuesNo();
			int queueSize = maxQueueSize / threadsNo;
			if (conc != null && !conc.trim().isEmpty()) {
					String[] plug_conc = conc.split(":");
					try {
				threadsNo = Integer.parseInt(plug_conc[0]);
				queueSize = maxQueueSize / threadsNo;
				
					} catch (Exception e) {
						
					}
					if (plug_conc.length > 1) {
					try {
				queueSize = Integer.parseInt(plug_conc[1]);
					
					} catch (Exception e) {
						
					}
					}
			}

			
if ((workerThreads.get(defPluginsThreadsPool) == null) || (conc != null)) {
      //如果插件还没有对应的处理线程集，则为他创建对应的线程集合。
     if (!workerThreads.containsKey(proc.id())) {
           ProcessorWorkerThread  worker = new ProcessorWorkerThread();	
	   ProcessingThreads<ProcessorWorkerThread> pt =	
    new ProcessingThreads<ProcessorWorkerThread>(worker, threadsNo,queueSize, proc.id());
          workerThreads.put(proc.id(), pt); //加入map中，等待使用时获取
	}
     }
           processors.put(proc.id(), proc);
			
			loaded  = true;
			result  = proc;
			version = result.getComponentInfo().getComponentVersion();
			
		}

		XMPPPreprocessorIfc preproc = SessionManagerConfig.getPreprocessor(plug_id);

		if (preproc != null) {
			preProcessors.put(plug_id, preproc);
			loaded = true;
			result = preproc;
		}

		XMPPPostprocessorIfc postproc = SessionManagerConfig.getPostprocessor(plug_id);

		if (postproc != null) {
			postProcessors.put(plug_id, postproc);
			loaded = true;
			result = postproc;
		}

	XMPPStopListenerIfc stoplist = SessionManagerConfig.getStopListener(plug_id);

		if (stoplist != null) {
			stopListeners.put(plug_id, stoplist);
			loaded = true;
			result = stoplist;
		}

		XMPPPacketFilterIfc filterproc = SessionManagerConfig.getPacketFilter(plug_id);

		if (filterproc != null) {
			outFilters.put(plug_id, filterproc);
			loaded = true;
			result = filterproc;
		}
		if (!loaded) {
	  log.log(Level.WARNING, "No implementation found for plugin id: {0}", plug_id);
		}    // end of if (!loaded)
		if (result != null) {
		allPlugins.add(result);
	if (result instanceof PresenceCapabilitiesManager.PresenceCapabilitiesListener) {
	PresenceCapabilitiesManager.registerPresenceHandler((PresenceCapabilitiesManager
						.PresenceCapabilitiesListener) result);
	}
	if (result instanceof XMPPPresenceUpdateProcessorIfc) {
	 presenceProcessors.put(result.id(), (XMPPPresenceUpdateProcessorIfc) result);
	}
	}

	return result;
}