`

tigase component (二)

阅读更多

除了直接使用在《Tigase如何创建一个新Component 第一部分》当中介绍的那些接口外,我推荐extend一些现成的抽象类,因为它们已经实现了大部分“繁琐和令人厌倦”的接口实现工作。这里有一个你可能会感兴趣的抽象类列表。

  1. tigase.server.AbstractMessageReceiver – 它已经实现了四个接口:ServerComponent,MessageReceiver,Configurable和StatisticsContainer。它通过自己的多个线程来管理内部数据队列,且能避免死锁。它使用事件驱动的方式来处理数据,当packet被发送到AbstractMessageReceiver实例的abstract void processPacket(Packet packet)方法时,就立即启动了packet的处理工作。当然你还是需要实现抽象类当中的抽象方法,如果你还希望输出packet数据(例如当它收到请求时还需要发送响应),可以调用boolean addOutPacket(Packet packet)方法。
    /**
     * 它是所有用来处理用户packet的对象的父类。它使用内部队列和
     * 多个独立线程(线程个数取决于cpu个数)来应对高负载的packet处理工作。如果对它进行适当扩展
     * 还可以通过ad-hoc指令来处理普通的用户和管理员发送的packets。这一类的组件有 MUC,PubSub,SessionManager。
     * 它为管理员发送的ad-hoc指令提供脚本API支持。
     * 它内部使用的优先级队列在某些罕见情况下可能导致packet的顺序重新排列。如果你不允许在部署时
     * 发生这种情况,那么可以使用非优先级队列。队列的长度是有限制的,这取决于可用内存的大小。
     * packet在调用“processPacket(Packet packet)”时被处理,方法的调用是多线程并发处理的。
     */
    public abstract class AbstractMessageReceiver extends BasicComponent
            implements StatisticsContainer, MessageReceiver {
    
        /**
         * “在component层设置的传入packet过滤器”键名
         */
        public static final String INCOMING_FILTERS_PROP_KEY = "incoming-filters";
    
        /**
         * 被Tigase服务器载入的传入packet过滤器值名
         * 这是一个由“,”分割的packet过滤器类名列表,过滤器必须实现“PacketFilterIfc”接口
         */
        public static final String INCOMING_FILTERS_PROP_VAL = "tigase.server.filters.PacketCounter";
    
        /**
         * component的内部队列长度键名,如果修改这个键值会覆写默认的队列长度,在默认情况下
         * 这个值会根据内存的大小来进行调整。
         */
        public static final String MAX_QUEUE_SIZE_PROP_KEY = "max-queue-size";
    
        /**
         * 最大队列长度的默认值。这个值在服务启动时被计算出来,计算公式为:
         * Runtime.getRuntime().maxMemory() / 400000L
         * 你可以通过修改服务配置项“MAX_QUEUE_SIZE_PROP_KEY”的值来修改队列长度
         */
        public static final Integer MAX_QUEUE_SIZE_PROP_VAL = new Long(Runtime.getRuntime().maxMemory()
                / 400000L).intValue();
    
        /**
         * “在component层设置的传出packet过滤器”键名
         */
        public static final String OUTGOING_FILTERS_PROP_KEY = "outgoing-filters";
    
        /**
         * 被Tigase服务器载入的传出packet过滤器值名
         * 这是一个由“,”分割的packet过滤器类名列表,过滤器必须实现“PacketFilterIfc”接口
         */
        public static final String OUTGOING_FILTERS_PROP_VAL = "tigase.server.filters.PacketCounter";
    
        /**
         * 在时间计算程序中被用到的常量,1秒=1000毫秒
         */
        protected static final long SECOND = 1000;
    
        /**
         * 在时间计算程序中被用到的常量,1分钟=60秒
         */
        protected static final long MINUTE = 60 * SECOND;
    
        /**
         * 在时间计算程序中被用到的常量,1小时=60分钟
         */
        protected static final long HOUR = 60 * MINUTE;
    
        /**
         * logger
         */
        private static final Logger log = Logger.getLogger("tigase.debug.AbstractMessageReceiver");
    
        //~--- fields ---------------------------------------------------------------
    
        private int in_queues_size = 1;
        private long last_hour_packets = 0;
        private long last_minute_packets = 0;
        private long last_second_packets = 0;
        protected int maxQueueSize = MAX_QUEUE_SIZE_PROP_VAL;
        private QueueListener out_thread = null;
        private long packetId = 0;
        private long packets_per_hour = 0;
        private long packets_per_minute = 0;
        private long packets_per_second = 0;
        private MessageReceiver parent = null;
        private int pptIdx = 0;
    
        // 队列缓存可以提高程序运行效能
        private final Priority[] pr_cache = Priority.values();
        private final CopyOnWriteArrayList<PacketFilterIfc> outgoing_filters =
                new CopyOnWriteArrayList<PacketFilterIfc>();
        private final PriorityQueueAbstract<Packet> out_queue =
                PriorityQueueAbstract.getPriorityQueue(pr_cache.length, maxQueueSize);
        private final CopyOnWriteArrayList<PacketFilterIfc> incoming_filters =
                new CopyOnWriteArrayList<PacketFilterIfc>();
        private final List<PriorityQueueAbstract<Packet>> in_queues =
                new ArrayList<PriorityQueueAbstract<Packet>>(pr_cache.length);
        private final long[] processPacketTimings = new long[100];
        private Timer receiverTasks = null;
    
        /**
         * 变量statAddedMessagesEr用来记录因为队列满而无法成功添加到队列的消息个数
         */
        private long statReceivedPacketsEr = 0;
    
        /**
         * 变量statAddedMessagesOk用来记录已经成功添加到队列的消息个数
         */
        private long statReceivedPacketsOk = 0;
        private long statSentPacketsEr = 0;
        private long statSentPacketsOk = 0;
        private ArrayDeque<QueueListener> threadsQueue = null;
        private final ConcurrentHashMap<String, PacketReceiverTask> waitingTasks =
                new ConcurrentHashMap<String, PacketReceiverTask>(16, 0.75f, 4);
        private final Set<Pattern> regexRoutings =
                new ConcurrentSkipListSet<Pattern>(new PatternComparator());
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * 这是packet处理的主方法。由于它会被多线程并发调用,因此在实现它的时候要养成首先对packet进行备份的习惯
         * 所有被定位到该组件的packet都会通过这个方法被处理。
         * 需要提醒的是,packet的实例可能在运行时被多个component,多个plug-in同时处理。
         * 所以最好把packet视作为immutable对象
         * 这个方法的处理过程是异步的,因此它没有返回值。如果在某些情况下需要返回值,可以把返回值传给
         * “addOutPacket(Packet)”方法
         *
         * @param packet 作为入口参数的packet实例
         */
        public abstract void processPacket(Packet packet);
    
        /**
         * 这个方法把一个packet实例添加到内部的输入队列。队列里面的packet将异步的传递到“processPacket(Packet)”
         * 方法进行处理。当队列满了的时候,这会是一个阻塞性的方法,直到有了足够的空间能够容纳新加入的packet。
         * 方法被设计为阻塞性的目的是避免系统过载,并避免因产生压根无法处理的packet时所造成的资源浪费。
         * 方法返回布尔类型变量,如果成功添加到队列则返回true,否则返回false。
         * 在运行时可以有多个队列和多个线程处理packet,这个方法可以有效保证packet按照正确顺序
         * 进行处理。举个例子,同一个用户的的多个packet会出现在同一个队列里面。你可以调整“hashCodeForPacket(Packet)”
         * 方法来改变packets的分发策略。如果有N个线程,那么packet的分发会依照下面的逻辑:
         * int threadNo = Math.abs(hashCodeForPacket(packet) % N);
         * 大部分的Tigase组件的都会使用这个方法。
         *
         * @param packet 被添加到内部队列的packet实例
         * @return 布尔类型,如果成功返回true,否则返回false
         */
        @Override
        public boolean addPacket(Packet packet) {
            int queueIdx = Math.abs(hashCodeForPacket(packet) % in_queues_size);
    
            if (log.isLoggable(Level.FINEST)) {
                log.log(Level.FINEST, "[{0}] queueIdx={1}, {2}", new Object[]{getName(), queueIdx,
                        packet.toStringSecure()});
            }
    
            try {
                in_queues.get(queueIdx).put(packet, packet.getPriority().ordinal());
                ++statReceivedPacketsOk;
            } catch (InterruptedException e) {
                ++statReceivedPacketsEr;
    
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "Packet dropped for unknown reason: {0}", packet);
                }
    
                return false;
            }    // end of try-catch
    
            return true;
        }
    
        /**
         * 这是一个添加packet到内部队列的非阻塞性方法。
         * 方法返回布尔型变量,如果成功添加到队列则返回true,否则返回false。
         * 在大多数情况,这个方法是不被推荐的。唯一一个允许使用该方法的组件是“MessageRouter”,
         * 因为它不允许因为任何原因造成阻塞。
         * 使用非阻塞性方法添加packet可能造成死锁,所有非MessageRounter组件必须使用阻塞性方法,
         * 在内部队列满的高负载情况下必须等待。请参考阻塞性“addPacket(Packet packet)”方法
         *
         * @param packet 被添加到内部队列的packet实例
         * @return 布尔类型,如果成功返回true,否则返回false
         *         请参考阻塞式的“addPacket(Packet packet)”方法
         */
        @Override
        public boolean addPacketNB(Packet packet) {
            int queueIdx = Math.abs(hashCodeForPacket(packet) % in_queues_size);
    
            if (log.isLoggable(Level.FINEST)) {
                log.log(Level.FINEST, "[{0}] queueIdx={1}, {2}", new Object[]{getName(), queueIdx,
                        packet.toStringSecure()});
            }
    
            boolean result = in_queues.get(queueIdx).offer(packet, packet.getPriority().ordinal());
    
            if (result) {
                ++statReceivedPacketsOk;
            } else {
    
                // Queue overflow!
                ++statReceivedPacketsEr;
    
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "Packet dropped due to queue overflow: {0}", packet);
                }
            }
    
            return result;
        }
    
        /**
         * 这是通过传递一个packet队列达到批量添加packet到component内部队列的方法。
         * 这个方法循环调用“addPacket(Packet)”方法
         * 如果入口参数队列中的所有packet对象都被移出并添加到component内部队列,返回true。否则中断
         * 循环,返回false。
         * 需要提醒的是,如果方法返回true,那么入口参数队列中的每一个packet实例都已经被传递
         * 给“addPacket(Packet)”方法,那么这时入口参数队列应该已经被清空。如果返回false,
         * 那么入口参数队列中应该至少含有一个packet没有成功的被addPacket(Packet)方法执行,
         * 和所有压根还没有被来得及被传递给addPacket(Packet)方法的packet。
         *
         * @param packets 一个包含所有需要被添加到内部队列的packet对象队列。所有的packet对象如果由
         *                一个线程处理,那么它们会严格按照在入口参数队列的顺序被处理。请参考“hashCodeForPacket(Packet)”
         *                的方法备注或文档来了解如何分发packet到特定的线程。
         * @return 如果入口参数的队列中的所有packet对象都被移出并添加到component内部队列,返回true。否则
         *         返回false。
         *         请参考“hashCodeForPacket(Packet)”的方法备注或文档来了解如何分发packet到特定的线程。
         */
        @Override
        public boolean addPackets(Queue<Packet> packets) {
            boolean result = true;
            Packet p = packets.peek();
    
            while (p != null) {
                result = addPacket(p);
    
                if (result) {
                    packets.poll();
                } else {
                    break;
                }    // end of if (result) else
    
                p = packets.peek();
            }      // end of while ()
    
            return result;
        }
    
        /**
         * 这个方法为组件添加一个新的路由地址。MessageRouter会通过路由地址用来计算得出packet的
         * 下一步去向。如果packet的最终地址与一个组件的路由地址匹配。那么这个packet将会被添加到该
         * 组件的内部输入队列。
         * 默认情况下,所有的component接受两种方式的寻址方式,一为组件id
         * 二是component.getName() + '@' + any virtual domain
         *
         * @param address 是一个java正则表达式字符串,所有与该正则表达式匹配的packet地址都将被
         *                component接收。在将来这个地址很可能被调整为any virtual domain + '/' + component.getName()
         *                作为入口参数传递的java正则表达式是component能够接收的packet的地址的严格匹配。在大多数情况下,
         *                这个方法为外部component的协议调用准备的,这个方法为外部component提供了动态修改component的接收
         *                packet地址实现。
         */
        public void addRegexRouting(String address) {
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "{0} - attempt to add regex routing: {1}", new Object[]{getName(),
                        address});
            }
    
            regexRoutings.add(Pattern.compile(address, Pattern.CASE_INSENSITIVE));
    
            if (log.isLoggable(Level.FINE)) {
                log.log(Level.FINE, "{0} - success adding regex routing: {1}", new Object[]{getName(),
                        address});
            }
        }
    
        /**
         * 清除路由正则表达式方法,将所有的component路由正则表达式地址移除。当这个方法被执行后,
         * component只接收默认的路由地址packet,仍然是两种方式:一为componentid
         * 二是component.getName() + '@' + any virtual domain
         */
        public void clearRegexRoutings() {
            regexRoutings.clear();
        }
    
        /**
         * 每个小时都会执行的工具类方法。一个实际的组件可以覆写这个方法,把自己的实现代码放进去,
         * 达到每小时执行一次的效果。
         * 需要提醒的是,复杂的计算或者需要运行很久的操作应该避免被放进这个方法,这个方法的执行需要确保
         * 在一个小时内完成!覆写方法时必须首先调用父类的everyHour()方法,然后执行自己的实现代码。
         */
        public synchronized void everyHour() {
            packets_per_hour = statReceivedPacketsOk - last_hour_packets;
            last_hour_packets = statReceivedPacketsOk;
        }
    
        /**
         * 每分钟都会执行的工具类方法。一个实际的组件可以覆写这个方法,把自己的实现代码放进去,
         * 达到每分钟执行一次的效果。
         * 需要提醒的是,复杂的计算或者需要运行很久的操作应该避免被放进这个方法,这个方法的执行需要确保
         * 在一分钟内完成!覆写方法时必须首先调用父类的everyMinute()方法,然后执行自己的实现代码。
         */
        public synchronized void everyMinute() {
            packets_per_minute = statReceivedPacketsOk - last_minute_packets;
            last_minute_packets = statReceivedPacketsOk;
            receiverTasks.purge();
        }
    
        /**
         * 每一秒都会执行的工具类方法。一个实际的component可以覆写这个方法,把自己的实现代码放进去,
         * 达到每秒执行一次的效果。
         * 需要提醒的是,复杂的计算或者需要运行很久的操作应该避免被放进这个方法,这个方法的执行需要确保
         * 在一秒钟内完成!覆写方法时必须首先调用父类的everySecond()方法,然后执行自己的实现代码。
         */
        public synchronized void everySecond() {
            packets_per_second = statReceivedPacketsOk - last_second_packets;
            last_second_packets = statReceivedPacketsOk;
        }
    
        //~--- get methods ----------------------------------------------------------
    
        /**
         * 返回组件的缺省配置信息。返回值为Map类型,配置的属性id作为key,属性值作为value。
         * 通过调用该方法获得的缺省配置项将会在之后传递给“setProperties(...)”方法,它们当中的一些
         * 值可能会因为服务器配置文件被改变,配置项的值是一下几种元类型:int,long,boolean,String
         *
         * @param params 是一个在启动服务器时被初始化的属性Map,这些属性将会在生成组件缺省
         *               配置信息的时作为暗示和预设值
         * @return a 组件的默认配置Map
         */
        @Override
        public Map<String, Object> getDefaults(Map<String, Object> params) {
            Map<String, Object> defs = super.getDefaults(params);
            String queueSize = (String) params.get(GEN_MAX_QUEUE_SIZE);
            int queueSizeInt = MAX_QUEUE_SIZE_PROP_VAL;
    
            if (queueSize != null) {
                try {
                    queueSizeInt = Integer.parseInt(queueSize);
                } catch (NumberFormatException e) {
                    queueSizeInt = MAX_QUEUE_SIZE_PROP_VAL;
                }
            }
    
            defs.put(MAX_QUEUE_SIZE_PROP_KEY, getMaxQueueSize(queueSizeInt));
            defs.put(INCOMING_FILTERS_PROP_KEY, INCOMING_FILTERS_PROP_VAL);
            defs.put(OUTGOING_FILTERS_PROP_KEY, OUTGOING_FILTERS_PROP_VAL);
    
            return defs;
        }
    
        /**
         * 这个方法返回所有欲编译完(正则表达式)之后的组件路由信息。返回值Set的长度可以为0,但是
         * set本身不可以为null。
         *
         * @return 所有欲编译完(正则表达式)之后的组件路由信息,封装为Set
         */
        public Set<Pattern> getRegexRoutings() {
            return regexRoutings;
        }
    
        /**
         * 方法返回组件的统计信息(多个统计项)。需要提醒的是,因为方法可能被服务监控系统每秒
         * 调用一次,因此方法体中不能有复杂/大运算量/耗时久的计算操作。
         * 如果统计信息当中有些项是需要较长时间才能产生的(比如查询数据库),那么它们必须要设置为
         * Level.FINEST,并且把它放到level guard当中,以阻止监控系统要求组件产生这些
         * 统计信息。监控系统不会收集Level.FINEST级别的统计项。
         * level guard的代码看起来就像下面的代码实例一样:
         * if (list.checkLevel(Level.FINEST)) {
         * // 一些cpu密集计算或长耗时的操作
         * list.add(getName(), "Statistic description", stat_value, Level.FINEST);
         * }
         * 这种方法可以避免你的密集计算型操作每秒中都执行,避免影响服务器性能。
         *
         * @param list 是一个StatistcsList实例,它保存了所有的统计项信息
         */
        @Override
        public void getStatistics(StatisticsList list) {
            list.add(getName(), "Last second packets", packets_per_second, Level.FINE);
            list.add(getName(), "Last minute packets", packets_per_minute, Level.FINE);
            list.add(getName(), "Last hour packets", packets_per_hour, Level.FINE);
            list.add(getName(), StatisticType.MSG_RECEIVED_OK.getDescription(), statReceivedPacketsOk,
                    Level.FINE);
            list.add(getName(), StatisticType.MSG_SENT_OK.getDescription(), statSentPacketsOk, Level.FINE);
    
            if (list.checkLevel(Level.FINEST)) {
                int[] in_priority_sizes = in_queues.get(0).size();
    
                for (int i = 1; i < in_queues.size(); i++) {
                    int[] tmp_pr_sizes = in_queues.get(i).size();
    
                    for (int j = 0; j < tmp_pr_sizes.length; j++) {
                        in_priority_sizes[j] += tmp_pr_sizes[j];
                    }
                }
    
                int[] out_priority_sizes = out_queue.size();
    
                for (int i = 0; i < in_priority_sizes.length; i++) {
                    Priority queue = Priority.values()[i];
    
                    list.add(getName(), "In queue: " + queue.name(), in_priority_sizes[queue.ordinal()],
                            Level.FINEST);
                    list.add(getName(), "Out queue: " + queue.name(), out_priority_sizes[queue.ordinal()],
                            Level.FINEST);
                }
            }
    
            int in_queue_size = 0;
    
            for (PriorityQueueAbstract<Packet> total_size : in_queues) {
                in_queue_size += total_size.totalSize();
            }
    
            int out_queue_size = out_queue.totalSize();
    
            list.add(getName(), "Total In queues wait", in_queue_size, Level.INFO);
            list.add(getName(), "Total Out queues wait", out_queue_size, Level.INFO);
            list.add(getName(), StatisticType.MAX_QUEUE_SIZE.getDescription(), maxQueueSize, Level.FINEST);
            list.add(getName(), StatisticType.IN_QUEUE_OVERFLOW.getDescription(), statReceivedPacketsEr,
                    Level.INFO);
            list.add(getName(), StatisticType.OUT_QUEUE_OVERFLOW.getDescription(), statSentPacketsEr,
                    Level.INFO);
    
            long res = 0;
    
            for (long ppt : processPacketTimings) {
                res += ppt;
            }
    
            long prcessingTime = res / processPacketTimings.length;
    
            list.add(getName(),
                    "Average processing time on last " + processPacketTimings.length + " runs [ms]",
                    prcessingTime, Level.FINE);
    
            for (PacketFilterIfc packetFilter : incoming_filters) {
                packetFilter.getStatistics(list);
            }
    
            for (PacketFilterIfc packetFilter : outgoing_filters) {
                packetFilter.getStatistics(list);
            }
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * 这个方法决定了传入packet被分发给哪个线程进行处理。不同的组件需要不同的分发逻辑,
         * 因为这样可以让所有的线程更有效率的工作,并避免packet重新排序。
         * 如果有N个处理线程,packet的分发规则逻辑如下:
         * int threadNo = Math.abs(hashCodeForPacket(packet) % N);
         * 比如对于PubSub组件,基于PubSub频道名称进行分发是一个较好选择;对于SessionMananger而言
         * 基于目的地地址分发是一个较好选择。
         *
         * @param packet 一个需要多线程处理的packet
         * @return 为线程提供的packet哈希值
         */
        public int hashCodeForPacket(Packet packet) {
            if ((packet.getFrom() != null) && (packet.getFrom() != packet.getStanzaFrom())) {
    
                // 这个packet来源于connection manager,所以最好是通过connectionid生成哈希值,
                // getFrom()方法返回connectionid。
                return packet.getFrom().hashCode();
            }
    
            // 否则,一个较好的方法是通过elemTo地址,因为elemTo地址是目的地用户的名称
            if (packet.getStanzaTo() != null) {
                return packet.getStanzaTo().getBareJID().hashCode();
            }
    
            if (packet.getTo() != null) {
                return packet.getTo().hashCode();
            }
    
            return 1;
        }
    
        //~--- get methods ----------------------------------------------------------
    
        /**
         * Method description
         *
         * @param address
         * @return
         */
        @Override
        public boolean isInRegexRoutings(String address) {
    
            // log.finest(getName() + " looking for regex routings: " + address);
            for (Pattern pat : regexRoutings) {
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "{0} matching: {1} against {2}", new Object[]{getName(), address,
                            pat.toString()});
                }
    
                if (pat.matcher(address).matches()) {
                    return true;
                }
    
                // log.finest(getName() + " matching failed against pattern: " + pat.toString());
            }
    
            return false;
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * Method description
         *
         * @param prefix
         * @return
         */
        public String newPacketId(String prefix) {
            StringBuilder sb = new StringBuilder(32);
    
            if (prefix != null) {
                sb.append(prefix).append("-");
            }
    
            sb.append(getName()).append(++packetId);
    
            return sb.toString();
        }
    
        /**
         * Method description
         *
         * @param packet
         * @param results
         */
        @Override
        public final void processPacket(final Packet packet, final Queue<Packet> results) {
            addPacketNB(packet);
        }
    
        /**
         * Method description
         *
         * @return
         */
        public int processingThreads() {
            return 1;
        }
    
        /**
         * Method description
         */
        @Override
        public void release() {
            stop();
        }
    
        /**
         * Method description
         *
         * @param address
         * @return
         */
        public boolean removeRegexRouting(String address) {
            return regexRoutings.remove(Pattern.compile(address, Pattern.CASE_INSENSITIVE));
        }
    
        //~--- set methods ----------------------------------------------------------
    
        /**
         * Method description
         *
         * @param maxQueueSize
         */
        public void setMaxQueueSize(int maxQueueSize) {
            if ((this.maxQueueSize != maxQueueSize) || (in_queues.size() == 0)) {
    
                // out_queue = PriorityQueueAbstract.getPriorityQueue(pr_cache.length, maxQueueSize);
                this.maxQueueSize = maxQueueSize / processingThreads();
    
                if (in_queues.size() == 0) {
                    for (int i = 0; i < in_queues_size; i++) {
                        PriorityQueueAbstract<Packet> queue =
                                PriorityQueueAbstract.getPriorityQueue(pr_cache.length, maxQueueSize);
    
                        in_queues.add(queue);
                    }
                } else {
                    for (int i = 0; i < in_queues.size(); i++) {
                        in_queues.get(i).setMaxSize(maxQueueSize);
                    }
                }
    
                out_queue.setMaxSize(maxQueueSize);
            }    // end of if (this.maxQueueSize != maxQueueSize)
        }
    
        /**
         * Method description
         *
         * @param name
         */
        @Override
        public void setName(String name) {
            super.setName(name);
            in_queues_size = processingThreads();
            setMaxQueueSize(maxQueueSize);
        }
    
        /**
         * Method description
         *
         * @param parent
         */
        @Override
        public void setParent(MessageReceiver parent) {
            this.parent = parent;
        }
    
        /**
         * 为Component设置配置项
         *
         * @param props
         */
        @Override
        @TODO(note = "Replace fixed filers loading with configurable options for that")
        public void setProperties(Map<String, Object> props) {
            super.setProperties(props);
    
            int queueSize = (Integer) props.get(MAX_QUEUE_SIZE_PROP_KEY);
    
            setMaxQueueSize(queueSize);
            incoming_filters.clear();
            outgoing_filters.clear();
    
            String filters = (String) props.get(INCOMING_FILTERS_PROP_KEY);
    
            if ((filters != null) && !filters.trim().isEmpty()) {
                String[] incoming = filters.trim().split(",");
    
                for (String inc : incoming) {
                    try {
                        PacketFilterIfc filter = (PacketFilterIfc) Class.forName(inc).newInstance();
    
                        filter.init(getName(), QueueType.IN_QUEUE);
                        incoming_filters.add(filter);
                        log.log(Level.CONFIG, "{0} loaded incoming filter: {1}", new Object[]{getName(), inc});
                    } catch (Exception e) {
                        log.log(Level.WARNING, "Problem loading filter: " + inc + " in component: " + getName(),
                                e);
                    }
                }
            }
    
            filters = (String) props.get(OUTGOING_FILTERS_PROP_KEY);
    
            if ((filters != null) && !filters.trim().isEmpty()) {
                String[] outgoing = filters.trim().split(",");
    
                for (String out : outgoing) {
                    try {
                        PacketFilterIfc filter = (PacketFilterIfc) Class.forName(out).newInstance();
    
                        filter.init(getName(), QueueType.OUT_QUEUE);
                        outgoing_filters.add(filter);
                        log.log(Level.CONFIG, "{0} loaded outgoing filter: {1}", new Object[]{getName(), out});
                    } catch (Exception e) {
                        log.log(Level.WARNING, "Problem loading filter: " + out + " in component: " + getName(),
                                e);
                    }
                }
            }
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * Method description
         */
        @Override
        public void start() {
            if (log.isLoggable(Level.FINER)) {
                log.log(Level.INFO, "{0}: starting queue management threads ...", getName());
            }
            startThreads();
        }
    
        /**
         * Method description
         */
        public void stop() {
            if (log.isLoggable(Level.FINER)) {
                log.log(Level.INFO, "{0}: stopping queue management threads ...", getName());
            }
    
            stopThreads();
        }
    
        protected boolean addOutPacket(Packet packet) {
            if (log.isLoggable(Level.FINEST)) {
                log.log(Level.FINEST, "[{0}]  {1}", new Object[]{getName(), packet.toStringSecure()});
            }
    
            try {
                out_queue.put(packet, packet.getPriority().ordinal());
                ++statSentPacketsOk;
            } catch (InterruptedException e) {
                ++statSentPacketsEr;
    
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "Packet dropped for unknown reason: {0}", packet);
                }
    
                return false;
            }    // end of try-catch
    
            return true;
        }
    
        /**
         * 非阻塞的addOutPacket版本
         *
         * @param packet 一个packet对象
         * @return 成功的布尔类型
         */
        protected boolean addOutPacketNB(Packet packet) {
            if (log.isLoggable(Level.FINEST)) {
                log.log(Level.FINEST, "[{0}]  {1}", new Object[]{getName(), packet.toStringSecure()});
            }
    
            boolean result = false;
    
            result = out_queue.offer(packet, packet.getPriority().ordinal());
    
            if (result) {
                ++statSentPacketsOk;
            } else {
    
                // Queue overflow!
                ++statSentPacketsEr;
    
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "Packet dropped due to queue overflow: {0}", packet);
                }
            }
    
            return result;
        }
    
        protected boolean addOutPacketWithTimeout(Packet packet, ReceiverTimeoutHandler handler,
                                                  long delay, TimeUnit unit) {
    
            new PacketReceiverTask(handler, delay, unit, packet);
    
            return addOutPacket(packet);
        }
    
        protected boolean addOutPackets(Queue<Packet> packets) {
            Packet p = null;
            boolean result = true;
    
            while ((p = packets.peek()) != null) {
                result = addOutPacket(p);
    
                if (result) {
                    packets.poll();
                } else {
                    return false;
                }    // end of if (result) else
            }      // end of while ()
    
            return true;
        }
    
        protected void addTimerTask(TimerTask task, long delay, TimeUnit unit) {
            receiverTasks.schedule(task, unit.toMillis(delay));
        }
    
        protected void addTimerTask(TimerTask task, long delay) {
            receiverTasks.schedule(task, delay);
        }
    
        //~--- get methods ----------------------------------------------------------
    
        protected Integer getMaxQueueSize(int def) {
            return def;
        }
    
        //~--- methods --------------------------------------------------------------
    
        private Packet filterPacket(Packet packet, CopyOnWriteArrayList<PacketFilterIfc> filters) {
            Packet result = packet;
    
            for (PacketFilterIfc packetFilterIfc : filters) {
                result = packetFilterIfc.filter(result);
    
                if (result == null) {
                    break;
                }
            }
    
            return result;
        }
    
        private void startThreads() {
            if (threadsQueue == null) {
                threadsQueue = new ArrayDeque<QueueListener>(8);
    
                for (int i = 0; i < in_queues_size; i++) {
                    QueueListener in_thread = new QueueListener(in_queues.get(i), QueueType.IN_QUEUE);
    
                    in_thread.setName("in_" + i + "-" + getName());
                    in_thread.start();
                    threadsQueue.add(in_thread);
                }
            }    // end of if (thread == null || ! thread.isAlive())
    
            if ((out_thread == null) || !out_thread.isAlive()) {
                out_thread = new QueueListener(out_queue, QueueType.OUT_QUEUE);
                out_thread.setName("out_" + getName());
                out_thread.start();
            }    // end of if (thread == null || ! thread.isAlive())
    
            receiverTasks = new Timer(getName() + " tasks", true);
            receiverTasks.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    everySecond();
                }
            }, SECOND, SECOND);
            receiverTasks.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    everyMinute();
                }
            }, MINUTE, MINUTE);
            receiverTasks.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    everyHour();
                }
            }, HOUR, HOUR);
        }
    
        private void stopThreads() {
    
            // stopped = true;
            try {
                if (threadsQueue != null) {
                    for (QueueListener in_thread : threadsQueue) {
                        in_thread.threadStopped = true;
                        in_thread.interrupt();
    
                        while (in_thread.isAlive()) {
                            Thread.sleep(100);
                        }
                    }
                }
    
                if (out_thread != null) {
                    out_thread.threadStopped = true;
                    out_thread.interrupt();
    
                    while (out_thread.isAlive()) {
                        Thread.sleep(100);
                    }
                }
            } catch (InterruptedException e) {
            }
    
            threadsQueue = null;
            out_thread = null;
    
            if (receiverTasks != null) {
                receiverTasks.cancel();
                receiverTasks = null;
            }
        }
    
        //~--- inner classes --------------------------------------------------------
    
        private class PacketReceiverTask extends TimerTask {
            private ReceiverTimeoutHandler handler = null;
            private String id = null;
            private Packet packet = null;
    
            //~--- constructors -------------------------------------------------------
    
            private PacketReceiverTask(ReceiverTimeoutHandler handler, long delay, TimeUnit unit,
                                       Packet packet) {
                super();
                this.handler = handler;
                this.packet = packet;
                id = packet.getFrom().toString() + packet.getStanzaId();
                waitingTasks.put(id, this);
                receiverTasks.schedule(this, unit.toMillis(delay));
    
    //    log.finest("[" + getName() + "]  " + "Added timeout task for: " + id);
            }
    
            //~--- methods ------------------------------------------------------------
    
            /**
             * Method description
             *
             * @param response
             */
            public void handleResponse(Packet response) {
    
                this.cancel();
    
                handler.responseReceived(packet, response);
            }
    
            /**
             * Method description
             */
            public void handleTimeout() {
    
                waitingTasks.remove(id);
                handler.timeOutExpired(packet);
            }
    
            /**
             * Method description
             */
            @Override
            public void run() {
                handleTimeout();
            }
        }
    
        private class QueueListener extends Thread {
            private String compName = null;
            private QueueType type = null;
            private boolean threadStopped = false;
            private PriorityQueueAbstract<Packet> queue;
    
            //~--- constructors -------------------------------------------------------
    
            private QueueListener(PriorityQueueAbstract<Packet> q, QueueType type) {
                this.queue = q;
                this.type = type;
                compName = AbstractMessageReceiver.this.getName();
            }
    
            //~--- methods ------------------------------------------------------------
    
            /**
             * Method description
             */
            @Override
            public void run() {
                if (log.isLoggable(Level.FINEST)) {
                    log.finest(getName() + " starting queue processing.");
                }
    
                Packet packet = null;
                Queue<Packet> results = new ArrayDeque<Packet>(2);
    
                while (!threadStopped) {
                    try {
    
                        // 现在处理下一个等待的packet
                        packet = queue.take();
    
                        switch (type) {
                            case IN_QUEUE:
                                long startPPT = System.currentTimeMillis();
    
                                PacketReceiverTask task = null;
    
                                if (packet.getTo() != null) {
                                    String id = packet.getTo().toString() + packet.getStanzaId();
    
                                    task = waitingTasks.remove(id);
                                }
    
                                if (task != null) {
                                    task.handleResponse(packet);
                                } else {
    
                                    // 也许这是一个本地处理指令
                                    boolean processed = false;
    
                                    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);
                                    }
    
                                    processPacketTimings[pptIdx] = System.currentTimeMillis() - startPPT;
                                    pptIdx = (pptIdx + 1) % processPacketTimings.length;
                                }
    
                                break;
    
                            case OUT_QUEUE:
                                if ((packet = filterPacket(packet, outgoing_filters)) != null) {
                                    if (parent != null) {
                                        parent.addPacket(packet);
                                    } else {
    
                                        // 非阻塞性的addPacket()方法只能出现在MessageRouter当中
                                        addPacketNB(packet);
                                    }    // end of else
                                }
    
                                break;
    
                            default:
                                log.severe("Unknown queue element type: " + type);
    
                                break;
                        }          // end of switch (qel.type)
                    } catch (InterruptedException e) {
    
                    } catch (Exception e) {
                        log.log(Level.SEVERE,
                                "[" + getName() + "] Exception during packet processing: " + packet, e);
                    }    // end of try-catch
                }      // end of while (! threadStopped)
            }
        }
    }
    
  2. tigase.server.ConnectionManager – 这是一个extend AbstractMessageReceiver的抽象类。正如其名,这个类专注于对连接进行管理工作。如果你的组件需要通过网络直接发送或接受数据(比如c2s connection,s2s connection 或者 连接到外部第三方jabber服务),你可以把它作为基类进行扩展。它会帮你把所有和网络有关的工作都打理好(例如io,重连,socket监听和连接握手等工作)。如果你extend这个类,你需要知道数据来源于哪里:如果来源于MessageRouter,那么abstract void processPacket(Packet packet)方法会被调用; 如果来源于网络连接,那么abstract Queue processSocketData(XMPPIOService serv)方法会被调用。
    public abstract class ConnectionManager<IO extends XMPPIOService> extends AbstractMessageReceiver
            implements XMPPIOServiceListener<IO> {
        private static final Logger log = Logger.getLogger(ConnectionManager.class.getName());
    
        /**
         * Field description
         */
        public static final String NET_BUFFER_ST_PROP_KEY = "--net-buff-standard";
    
        /**
         * Field description
         */
        public static final String NET_BUFFER_HT_PROP_KEY = "--net-buff-high-throughput";
        protected static final String PORT_KEY = "port-no";
        protected static final String PROP_KEY = "connections/";
        protected static final String PORTS_PROP_KEY = PROP_KEY + "ports";
        protected static final String PORT_TYPE_PROP_KEY = "type";
        protected static final String PORT_SOCKET_PROP_KEY = "socket";
        protected static final String PORT_IFC_PROP_KEY = "ifc";
        protected static final String PORT_CLASS_PROP_KEY = "class";
        protected static final String PORT_REMOTE_HOST_PROP_KEY = "remote-host";
        protected static final String PORT_REMOTE_HOST_PROP_VAL = "localhost";
        protected static final String TLS_PROP_KEY = PROP_KEY + "tls/";
        protected static final String TLS_USE_PROP_KEY = TLS_PROP_KEY + "use";
        protected static final boolean TLS_USE_PROP_VAL = true;
        protected static final String TLS_REQUIRED_PROP_KEY = TLS_PROP_KEY + "required";
        protected static final boolean TLS_REQUIRED_PROP_VAL = false;
    
        protected static final String MAX_RECONNECTS_PROP_KEY = "max-reconnects";
        protected static final String NET_BUFFER_PROP_KEY = "net-buffer";
        protected static final int NET_BUFFER_ST_PROP_VAL = 2 * 1024;
        protected static final int NET_BUFFER_HT_PROP_VAL = 64 * 1024;
    
        /**
         * Field description
         */
        public static final String PORT_LOCAL_HOST_PROP_KEY = "local-host";
        private static ConnectionOpenThread connectThread = ConnectionOpenThread.getInstance();
        private static SocketReadThread readThread = SocketReadThread.getInstance();
    
        //~--- fields ---------------------------------------------------------------
    
        /**
         * Field description
         */
        public String[] PORT_IFC_PROP_VAL = {"*"};
    
        private int services_size = 0;
        private Thread watchdog = null;
        private long watchdogRuns = 0;
        private long watchdogStopped = 0;
        private long watchdogTests = 0;
        private LinkedList<Map<String, Object>> waitingTasks = new LinkedList<Map<String, Object>>();
        private ConcurrentHashMap<String, IO> services = new ConcurrentHashMap<String, IO>();
        private Set<ConnectionListenerImpl> pending_open =
                Collections.synchronizedSet(new HashSet<ConnectionListenerImpl>());
    
        protected int net_buffer = NET_BUFFER_ST_PROP_VAL;
    
        private boolean initializationCompleted = false;
        protected long connectionDelay = 2 * SECOND;
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * Method description
         *
         * @param serv
         * @return
         */
        public abstract Queue<Packet> processSocketData(IO serv);
    
        /**
         * Method description
         *
         * @param port_props
         */
        public abstract void reconnectionFailed(Map<String, Object> port_props);
    
        //~--- get methods ----------------------------------------------------------
    
        protected abstract long getMaxInactiveTime();
    
        protected abstract IO getXMPPIOServiceInstance();
    
        /**
         * Method description
         *
         * @param params
         * @return
         */
        @Override
        public Map<String, Object> getDefaults(Map<String, Object> params) {
            log.config(getName() + " defaults: " + params.toString());
    
            Map<String, Object> props = super.getDefaults(params);
    
            props.put(TLS_USE_PROP_KEY, TLS_USE_PROP_VAL);
    
            int buffSize = NET_BUFFER_ST_PROP_VAL;
    
            if (isHighThroughput()) {
                buffSize = DataTypes.parseSizeInt((String) params.get(NET_BUFFER_HT_PROP_KEY),
                        NET_BUFFER_HT_PROP_VAL);
            } else {
                buffSize = DataTypes.parseSizeInt((String) params.get(NET_BUFFER_ST_PROP_KEY),
                        NET_BUFFER_ST_PROP_VAL);
            }
    
            props.put(NET_BUFFER_PROP_KEY, buffSize);
    
            int[] ports = null;
            String ports_str = (String) params.get("--" + getName() + "-ports");
    
            if (ports_str != null) {
                String[] ports_stra = ports_str.split(",");
    
                ports = new int[ports_stra.length];
    
                int k = 0;
    
                for (String p : ports_stra) {
                    try {
                        ports[k++] = Integer.parseInt(p);
                    } catch (Exception e) {
                        log.warning("Incorrect ports default settings: " + p);
                    }
                }
            }
    
            int ports_size = 0;
    
            if (ports != null) {
                log.config("Port settings preset: " + Arrays.toString(ports));
    
                for (int port : ports) {
                    putDefPortParams(props, port, SocketType.plain);
                }        // end of for (int i = 0; i < idx; i++)
    
                props.put(PORTS_PROP_KEY, ports);
            } else {
                int[] plains = getDefPlainPorts();
    
                if (plains != null) {
                    ports_size += plains.length;
                }        // end of if (plains != null)
    
                int[] ssls = getDefSSLPorts();
    
                if (ssls != null) {
                    ports_size += ssls.length;
                }        // end of if (ssls != null)
    
                if (ports_size > 0) {
                    ports = new int[ports_size];
                }        // end of if (ports_size > 0)
    
                if (ports != null) {
                    int idx = 0;
    
                    if (plains != null) {
                        idx = plains.length;
    
                        for (int i = 0; i < idx; i++) {
                            ports[i] = plains[i];
                            putDefPortParams(props, ports[i], SocketType.plain);
                        }    // end of for (int i = 0; i < idx; i++)
                    }      // end of if (plains != null)
    
                    if (ssls != null) {
                        for (int i = idx; i < idx + ssls.length; i++) {
                            ports[i] = ssls[i - idx];
                            putDefPortParams(props, ports[i], SocketType.ssl);
                        }    // end of for (int i = 0; i < idx + ssls.length; i++)
                    }      // end of if (ssls != null)
    
                    props.put(PORTS_PROP_KEY, ports);
                }        // end of if (ports != null)
            }
    
            return props;
        }
    
        /**
         * 生成组件统计信息
         *
         * @param list 是一个统计项列表
         */
        @Override
        public void getStatistics(StatisticsList list) {
            super.getStatistics(list);
            list.add(getName(), "Open connections", services_size, Level.INFO);
    
            if (list.checkLevel(Level.FINEST)) {
                int waitingToSendSize = 0;
    
                for (IO serv : services.values()) {
                    waitingToSendSize += serv.waitingToSendSize();
                }
    
                list.add(getName(), "Waiting to send", waitingToSendSize, Level.FINEST);
            }
    
            list.add(getName(), "Watchdog runs", watchdogRuns, Level.FINER);
            list.add(getName(), "Watchdog tests", watchdogTests, Level.FINE);
            list.add(getName(), "Watchdog stopped", watchdogStopped, Level.FINE);
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * Method description
         *
         * @param binds
         */
        @Override
        public void initBindings(Bindings binds) {
            super.initBindings(binds);
            binds.put(CommandIfc.SERVICES_MAP, services);
        }
    
        /**
         * Method description
         */
        @Override
        public void initializationCompleted() {
            initializationCompleted = true;
    
            for (Map<String, Object> params : waitingTasks) {
                reconnectService(params, connectionDelay);
            }
    
            waitingTasks.clear();
        }
    
        /**
         * Method description
         *
         * @param serv
         * @throws IOException
         */
        @Override
        public void packetsReady(IO serv) throws IOException {
            writePacketsToSocket(serv, processSocketData(serv));
        }
    
        /**
         * Method description
         *
         * @param packet
         */
        @Override
        public void processPacket(Packet packet) {
            writePacketToSocket(packet);
        }
    
        /**
         * Method description
         */
        @Override
        public void release() {
    
            releaseListeners();
            super.release();
        }
    
        /**
         * Method description
         *
         * @param service
         */
        @TODO(note = "Do something if service with the same unique ID is already started, "
                + "possibly kill the old one...")
        public void serviceStarted(final IO service) {
    
            String id = getUniqueId(service);
    
            if (log.isLoggable(Level.FINER)) {
                log.log(Level.FINER, "[[{0}]] Connection started: {1}", new Object[]{getName(), service});
            }
    
            IO serv = services.get(id);
    
            if (serv != null) {
                if (serv == service) {
                    log.log(Level.WARNING,
                            "{0}: That would explain a lot, adding the same service twice, ID: {1}",
                            new Object[]{getName(),
                                    serv});
                } else {
    
                    // Is it at all possible to happen???
                    // let's log it for now....
                    log.log(Level.WARNING, "{0}: Attempt to add different service with the same ID: {1}",
                            new Object[]{getName(),
                                    service});
    
                    // And stop the old service....
                    serv.stop();
                }
            }
    
            services.put(id, service);
            ++services_size;
    
        }
    
        /**
         * @param service
         * @return
         */
        @Override
        public boolean serviceStopped(IO service) {
    
            String id = getUniqueId(service);
    
            if (log.isLoggable(Level.FINER)) {
                log.log(Level.FINER, "[[{0}]] Connection stopped: {1}", new Object[]{getName(), service});
            }
    
            if (id != null) {
                boolean result = services.remove(id, service);
    
                if (result) {
                    --services_size;
                } else {
    
                    // Is it at all possible to happen???
                    // let's log it for now....
                    log.log(Level.WARNING, "[[{0}]] Attempt to stop incorrect service: {1}",
                            new Object[]{getName(),
                                    service});
                    Thread.dumpStack();
                }
    
                return result;
            }
    
            return false;
    
        }
    
        //~--- set methods ----------------------------------------------------------
    
        /**
         * Method description
         *
         * @param name
         */
        @Override
        public void setName(String name) {
            super.setName(name);
            watchdog = new Thread(new Watchdog(), "Watchdog - " + name);
            watchdog.setDaemon(true);
            watchdog.start();
        }
    
        /**
         * Method description
         *
         * @param props
         */
        @Override
        public void setProperties(Map<String, Object> props) {
            super.setProperties(props);
            net_buffer = (Integer) props.get(NET_BUFFER_PROP_KEY);
            releaseListeners();
    
            int[] ports = (int[]) props.get(PORTS_PROP_KEY);
    
            if (ports != null) {
                for (int i = 0; i < ports.length; i++) {
                    Map<String, Object> port_props = new LinkedHashMap<String, Object>(20);
    
                    for (Map.Entry<String, Object> entry : props.entrySet()) {
                        if (entry.getKey().startsWith(PROP_KEY + ports[i])) {
                            int idx = entry.getKey().lastIndexOf('/');
                            String key = entry.getKey().substring(idx + 1);
    
                            log.log(Level.CONFIG, "Adding port property key: {0}={1}", new Object[]{key,
                                    entry.getValue()});
                            port_props.put(key, entry.getValue());
                        }    // end of if (entry.getKey().startsWith())
                    }      // end of for ()
    
                    port_props.put(PORT_KEY, ports[i]);
                    addWaitingTask(port_props);
    
                }        // end of for (int i = 0; i < ports.length; i++)
            }          // end of if (ports != null)
    
            if ((Boolean) props.get(TLS_USE_PROP_KEY)) {
                Map<String, String> tls_params = new LinkedHashMap<String, String>(20);
    
            }    // end of if (use.equalsIgnoreCase())
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * Method description
         */
        @Override
        public void start() {
            super.start();
        }
    
        /**
         * Method description
         *
         * @param ios
         * @param p
         * @return
         */
        public boolean writePacketToSocket(IO ios, Packet p) {
            if (ios != null) {
                if (log.isLoggable(Level.FINER) && !log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINER, "{0}, Processing packet: {1}, type: {2}", new Object[]{ios,
                            p.getElemName(), p.getType()});
                }
    
                if (log.isLoggable(Level.FINEST)) {
                    log.log(Level.FINEST, "{0}, Writing packet: {1}", new Object[]{ios, p});
                }
    
                ios.addPacketToSend(p);
    
                try {
                    ios.processWaitingPackets();
                    readThread.addSocketService(ios);
    
                    return true;
                } catch (Exception e) {
                    log.log(Level.WARNING, ios + "Exception during writing packets: ", e);
    
                    try {
                        ios.stop();
                    } catch (Exception e1) {
                        log.log(Level.WARNING, ios + "Exception stopping XMPPIOService: ", e1);
                    }    // end of try-catch
                }      // end of try-catch
    
            } else {
                if (log.isLoggable(Level.FINE)) {
                    log.log(Level.FINE, "Can''t find service for packet: <{0}> {1}, service id: {2}",
                            new Object[]{p.getElemName(),
                                    p.getTo(), getServiceId(p)});
                }
            }    // end of if (ios != null) else
    
            return false;
        }
    
        /**
         * Method description
         *
         * @param serv
         * @param packets
         */
        public void writePacketsToSocket(IO serv, Queue<Packet> packets) {
            if (serv != null) {
    
                if ((packets != null) && (packets.size() > 0)) {
                    Packet p = null;
    
                    while ((p = packets.poll()) != null) {
                        if (log.isLoggable(Level.FINER) && !log.isLoggable(Level.FINEST)) {
                            log.log(Level.FINER, "{0}, Processing packet: {1}, type: {2}", new Object[]{serv,
                                    p.getElemName(), p.getType()});
                        }
    
                        if (log.isLoggable(Level.FINEST)) {
                            log.log(Level.FINEST, "{0}, Writing packet: {1}", new Object[]{serv, p});
                        }
    
                        serv.addPacketToSend(p);
                    }      // end of for ()
    
                    try {
                        serv.processWaitingPackets();
                        readThread.addSocketService(serv);
                    } catch (Exception e) {
                        log.log(Level.WARNING, serv + "Exception during writing packets: ", e);
    
                        try {
                            serv.stop();
                        } catch (Exception e1) {
                            log.log(Level.WARNING, serv + "Exception stopping XMPPIOService: ", e1);
                        }    // end of try-catch
                    }      // end of try-catch
                }
    
            } else {
                if (log.isLoggable(Level.FINE)) {
                    log.log(Level.FINE, "Can''t find service for packets: [{0}] ", packets);
                }
            }          // end of if (ios != null) else
        }
    
        protected void addWaitingTask(Map<String, Object> conn) {
            if (initializationCompleted) {
                reconnectService(conn, connectionDelay);
            } else {
                waitingTasks.add(conn);
            }
        }
    
        /**
         * 返回活动的网络连接数(IOServices).
         *
         * @return 活动的网络连接数(IOServices).
         */
        protected int countIOServices() {
            return services.size();
        }
    
        /**
         * 对所有活动的“IOService(活动的网络连接)”对象执行一个由“ServiceChecker”定义的工作
         *
         * @param checker 是一个定义了动作内容的“ServiceChecker”实例
         */
        protected void doForAllServices(ServiceChecker checker) {
            for (IO service : services.values()) {
                checker.check(service);
            }
        }
    
        //~--- get methods ----------------------------------------------------------
    
        protected int[] getDefPlainPorts() {
            return null;
        }
    
        protected int[] getDefSSLPorts() {
            return null;
        }
    
        protected Map<String, Object> getParamsForPort(int port) {
            return null;
        }
    
        protected String getServiceId(Packet packet) {
            return getServiceId(packet.getTo());
        }
    
        protected String getServiceId(JID jid) {
            return jid.getResource();
        }
    
        protected String getUniqueId(IO serv) {
            return serv.getUniqueId();
        }
    
        protected IO getXMPPIOService(String serviceId) {
            return services.get(serviceId);
        }
    
        protected IO getXMPPIOService(Packet p) {
            return services.get(getServiceId(p));
        }
    
        protected boolean isHighThroughput() {
            return false;
        }
    
        //~--- methods --------------------------------------------------------------
    
        /**
         * @param p
         * @return
         */
        protected boolean writePacketToSocket(Packet p) {
            IO ios = getXMPPIOService(p);
    
            if (ios != null) {
                return writePacketToSocket(ios, p);
            } else {
                return false;
            }
        }
    
        protected boolean writePacketToSocket(Packet p, String serviceId) {
            IO ios = getXMPPIOService(serviceId);
    
            if (ios != null) {
                return writePacketToSocket(ios, p);
            } else {
                return false;
            }
        }
    
        protected void writeRawData(IO ios, String data) {
    
            try {
                ios.writeRawData(data);
                readThread.addSocketService(ios);
            } catch (Exception e) {
                log.log(Level.WARNING, ios + "Exception during writing data: " + data, e);
    
                try {
                    ios.stop();
                } catch (Exception e1) {
                    log.log(Level.WARNING, ios + "Exception stopping XMPPIOService: ", e1);
                }    // end of try-catch
            }
    
        }
    
        private void putDefPortParams(Map<String, Object> props, int port, SocketType sock) {
            log.log(Level.CONFIG, "Generating defaults for port: {0}", port);
            props.put(PROP_KEY + port + "/" + PORT_TYPE_PROP_KEY, ConnectionType.accept);
            props.put(PROP_KEY + port + "/" + PORT_SOCKET_PROP_KEY, sock);
            props.put(PROP_KEY + port + "/" + PORT_IFC_PROP_KEY, PORT_IFC_PROP_VAL);
            props.put(PROP_KEY + port + "/" + PORT_REMOTE_HOST_PROP_KEY, PORT_REMOTE_HOST_PROP_VAL);
            props.put(PROP_KEY + port + "/" + TLS_REQUIRED_PROP_KEY, TLS_REQUIRED_PROP_VAL);
    
            Map<String, Object> extra = getParamsForPort(port);
    
            if (extra != null) {
                for (Map.Entry<String, Object> entry : extra.entrySet()) {
                    props.put(PROP_KEY + port + "/" + entry.getKey(), entry.getValue());
                }    // end of for ()
            }      // end of if (extra != null)
        }
    
        private void reconnectService(final Map<String, Object> port_props, long delay) {
            if (log.isLoggable(Level.FINER)) {
                log.log(Level.FINER, "Reconnecting service for: {0}, scheduling next try in {1}secs",
                        new Object[]{getName(),
                                delay / 1000});
            }
    
            addTimerTask(new TimerTask() {
                @Override
                public void run() {
                    String host = (String) port_props.get(PORT_REMOTE_HOST_PROP_KEY);
    
                    if (host == null) {
                        host = (String) port_props.get("remote-hostname");
                    }
    
                    int port = (Integer) port_props.get(PORT_KEY);
    
                    if (log.isLoggable(Level.FINE)) {
                        log.log(Level.FINE,
                                "Reconnecting service for component: {0}, to remote host: {1} on port: {2}",
                                new Object[]{getName(),
                                        host, port});
                    }
    
                    startService(port_props);
                }
            }, delay);
        }
    
        private void releaseListeners() {
            for (ConnectionListenerImpl cli : pending_open) {
                connectThread.removeConnectionOpenListener(cli);
            }
    
            pending_open.clear();
        }
    
        private void startService(Map<String, Object> port_props) {
            ConnectionListenerImpl cli = new ConnectionListenerImpl(port_props);
    
            if (cli.getConnectionType() == ConnectionType.accept) {
                pending_open.add(cli);
            }
    
            connectThread.addConnectionOpenListener(cli);
        }
    
        //~--- inner classes --------------------------------------------------------
    
        private class ConnectionListenerImpl implements ConnectionOpenListener {
            private Map<String, Object> port_props = null;
    
            //~--- constructors -------------------------------------------------------
    
            private ConnectionListenerImpl(Map<String, Object> port_props) {
                this.port_props = port_props;
            }
    
            //~--- methods ------------------------------------------------------------
    
            /**
             * Method description
             *
             * @param sc
             */
            @SuppressWarnings({"unchecked"})
            @Override
            public void accept(SocketChannel sc) {
                IO serv = getXMPPIOServiceInstance();
    
                serv.setIOServiceListener(ConnectionManager.this);
                serv.setSessionData(port_props);
    
                try {
                    serv.accept(sc);
    
                    if (getSocketType() == SocketType.ssl) {
                        serv.startSSL(false);
                    }    // end of if (socket == SocketType.ssl)
    
                    serviceStarted(serv);
                    readThread.addSocketService(serv);
                } catch (SocketException e) {
                    if (getConnectionType() == ConnectionType.connect) {
    
                        // 组件服务的接收方还没有准备好?
                        // 让我们等几秒然后再试一下
                        log.log(Level.FINEST, "Problem reconnecting the service: {0}", serv);
    
                        boolean reconnect = false;
                        Integer reconnects = (Integer) port_props.get(MAX_RECONNECTS_PROP_KEY);
    
                        if (reconnects != null) {
                            int recon = reconnects.intValue();
    
                            if (recon != 0) {
                                port_props.put(MAX_RECONNECTS_PROP_KEY, (--recon));
                                reconnect = true;
                            }    // end of if (recon != 0)
                        }
    
                        if (reconnect) {
                            reconnectService(port_props, connectionDelay);
                        } else {
                            reconnectionFailed(port_props);
                        }
                    } else {
    
                    }
                } catch (Exception e) {
                    log.log(Level.WARNING, "Can not accept connection.", e);
                    serv.stop();
                }          // end of try-catch
            }
    
            //~--- get methods --------------------------------------------------------
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public ConnectionType getConnectionType() {
                String type = null;
    
                if (port_props.get(PORT_TYPE_PROP_KEY) == null) {
                    log.warning(getName() + ": connection type is null: "
                            + port_props.get(PORT_KEY).toString());
                } else {
                    type = port_props.get(PORT_TYPE_PROP_KEY).toString();
                }
    
                return ConnectionType.valueOf(type);
            }
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public String[] getIfcs() {
                return (String[]) port_props.get(PORT_IFC_PROP_KEY);
            }
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public int getPort() {
                return (Integer) port_props.get(PORT_KEY);
            }
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public int getReceiveBufferSize() {
                return net_buffer;
            }
    
            /**
             * Method description
             *
             * @return
             */
            public SocketType getSocketType() {
                return SocketType.valueOf(port_props.get(PORT_SOCKET_PROP_KEY).toString());
            }
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public int getTrafficClass() {
                if (isHighThroughput()) {
                    return IPTOS_THROUGHPUT;
                } else {
                    return DEF_TRAFFIC_CLASS;
                }
            }
    
            //~--- methods ------------------------------------------------------------
    
            /**
             * Method description
             *
             * @return
             */
            @Override
            public String toString() {
                return port_props.toString();
            }
        }
    
        /**
         * 检查所有已经建立的连接,看看它们当中有没有已经被断开的
         */
        private class Watchdog implements Runnable {
    
            /**
             * Method description
             */
            @Override
            public void run() {
                while (true) {
                    try {
    
                        // Sleep...
                        Thread.sleep(10 * MINUTE);
                        ++watchdogRuns;
    
                        // 遍历所有的连接,检查它们是否都真正的活着。尝试对每一个非激活状态的服务
                        // 每小时(或更久)发送一个空格,关闭那些出现异常的服务。
                        doForAllServices(new ServiceChecker() {
                            @Override
                            public void check(final XMPPIOService service) {
    
                                try {
                                    if (null != service) {
                                        long curr_time = System.currentTimeMillis();
                                        long lastTransfer = service.getLastTransferTime();
    
                                        if (curr_time - lastTransfer >= getMaxInactiveTime()) {
    
                                            // 停止那些超过了keep-alive时间的非活动服务
                                            if (log.isLoggable(Level.INFO)) {
                                                log.info(getName() + ": Max inactive time exceeded, stopping: " + service);
                                            }
    
                                            ++watchdogStopped;
                                            service.stop();
                                        } else {
                                            if (curr_time - lastTransfer >= (29 * MINUTE)) {
    
                                                // 如果连接还活动着,那么至少每小时检查一次
                                                service.writeRawData(" ");
                                                ++watchdogTests;
                                            }
                                        }
                                    }
                                } catch (Exception e) {
    
                                    // 关闭服务
                                    try {
                                        if (service != null) {
                                            log.info(getName() + "Found dead connection, stopping: " + service);
                                            ++watchdogStopped;
                                            service.forceStop();
                                        }
                                    } catch (Exception ignore) {
    
                                        // 如果抛出异常,则什么事都不做
                                    }
                                }
    
                            }
                        });
                    } catch (InterruptedException e) {    /* Do nothing here */
                    }
                }
            }
        }
    }
    
分享到:
评论

相关推荐

    tigase-server-tigase-server-8.0.0.zip 源码

    Tigase服务器是一款开源的XMPP(Extensible Messaging and Presence Protocol)服务器,用于构建实时通信系统,如即时消息、语音和视频通话、群聊等。它的源码是用Java语言编写的,具有高度可扩展性和模块化设计,...

    Tigase 概述

    * XEP-0114:Jabber Component Protocol,定义了服务器和外部组件如(MUC)或其他消息网络之间的通信。 * XEP-0133:服务管理者,允许管理员管理用户,发送消息给所有用户,改变每天登陆的欢迎消息,管理管理员账户...

    tigase-local

    **二、Tigase安装** 1. **系统要求**:Tigase可以在多种操作系统上运行,包括Linux、Windows和macOS。确保你的系统满足Java运行环境的要求。 2. **下载**:访问Tigase官方网站下载最新版本的Tigase服务器软件包。 ...

    tigase 内部处理流程

    #### 二、Tigase启动流程及组件初始化 ##### 2.1 启动流程概述 Tigase的启动流程始于`TigaseServer`类中的`start`方法。在这个过程中,系统首先会加载配置信息,并基于这些信息初始化各个组件。 1. **配置加载**: ...

    Tigase学习笔记整理

    组件(Component)是Tigase服务器的基础,其中`tigase.server.ServerComponent`是最顶层的组件接口,所有组件都必须实现这个接口。`tigase.server.MessageReceiver`接口用于接收和处理数据包,如果你的组件需要像...

    tigase-utils-3.5.1.jar

    tigase相关jar包

    Tigase Server 7.0.1 源代码

    《Tigase Server 7.0.1 源代码详解》 Tigase Server,作为一款基于Java语言开发的轻量级、可伸缩的Jabber/XMPP服务器,自诞生以来就以其高效能和高并发能力受到业界的广泛关注。在7.0.1版本中,Tigase Server继续...

    tigase-server_配置好的.zip

    《Tigase服务器的配置详解》 Tigase服务器是一款基于Java开发的XMPP(Extensible Messaging and Presence Protocol)服务器,广泛应用于即时通讯、在线状态服务以及多用户聊天室等场景。本文将深入探讨Tigase服务器...

    tigase http-api 源码部署

    Tigase是一个开源的XMPP(可扩展消息与出席协议,最初被称为 Jabber)服务器,它允许实时通讯和在线状态信息的交换。HTTP API是Tigase服务器的一个可选模块,它允许通过HTTP请求进行与服务器的交互,这样的方式对于...

    tigase 集群设置

    Tigase是一个用Java编写的开源XMPP服务器,它具有高性能、高度可伸缩和模块化的特点。在构建需要处理大量并发用户和高消息吞吐量的系统时,部署一个Tigase集群变得非常必要。集群模式允许将多个Tigase服务器实例组合...

    XMPP_tigase_IM服务部署安装

    - **Tigase PubSub Component**:实现了消息发布和订阅机制,适用于实时通知场景。 - **Tigase Web Messenger**:基于Web的聊天客户端,可以直接集成到网站中。 #### 安装配置步骤 ##### Windows环境下的安装指南 ...

    tigase开发手册

    而“TigaseServerElements”(Tigase服务器元素)则可能会解释构成服务器的关键组件,如Component(组件)、Plug-in(插件)和Connector(连接器),以及它们之间的数据流和处理方式。这些都是开发者需要掌握的基础...

    tigase组件

    Tigase是一款开源的XMPP服务器,用于实现即时通讯和在线状态服务。它采用Java语言编写,具有跨平台性,并且支持多种扩展协议,如多用户聊天(MUC)、消息存档(Message Archiving)和发布订阅(PubSub)等。在你提供...

    tigase-server.7.0.2

    二、Tigase服务器7.0.2的核心特性 1. 性能优化:7.0.2版本着重提升了处理并发连接和大规模用户群的能力,降低了系统资源消耗,确保在高负载下依然稳定运行。 2. 安全强化:新增和更新了安全策略,加强了对SSL/TLS...

    tigase monitor配置

    Tigase Monitor配置详解 Tigase Monitor是一款用于监控Tigase XMPP服务器性能和状态的工具,特别适用于Tigase 7.0版本。本文将详细介绍如何在Tigase 7.0上配置Monitor模块,包括客户端和服务器部分。 首先,确保你...

    tigase-web-chat

    【Tigase Web Chat】是基于Tigase服务器的一个实时通信解决方案,专为构建Web端的聊天和协作应用而设计。Tigase是一款开源、跨平台的XMPP(Extensible Messaging and Presence Protocol)服务器,支持多种协议,如...

    tigase-server 组件

    Tigase服务器是一款开源的XMPP(Extensible Messaging and Presence Protocol)服务器,它支持即时通讯、在线状态服务、多用户聊天室以及各种扩展协议。在IT行业中,掌握Tigase服务器的配置和使用对于开发实时通信...

    Spark客户端连接Tigase

    Spark连接Tigase服务器,完整的步骤,很清晰的看到。大家可以参考。

    tigase 5.2完整代码数据库eclipse完美启动版本

    Tigase是一个开源的XMPP(Extensible Messaging and Presence Protocol)服务器,用于构建即时通讯和在线状态服务。Tigase 5.2是该服务器的一个版本,提供了强大的可扩展性和稳定性,支持多种协议如XMPP、Jabber等。...

    Tigase-MUC简介(群组相关功能)

    全面:tigase 完全实现了XMPP协议,除了全面实施的两个核心协议,它支持大多数的你可能永远都需要的扩展协议。 Tigase的MUC组件以单独的jar包,提供群组相关功能

Global site tag (gtag.js) - Google Analytics