/** * $RCSfile$ * $Revision$ * $Date$ * * Copyright (C) 2002-2003 Jive Software. All rights reserved. * ==================================================================== * The Jive Software License (based on Apache Software License, Version 1.1) * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * * 1. Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * * 2. Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in * the documentation and/or other materials provided with the * distribution. * * 3. The end-user documentation included with the redistribution, * if any, must include the following acknowledgment: * "This product includes software developed by * Jive Software (http://www.jivesoftware.com)." * Alternately, this acknowledgment may appear in the software itself, * if and wherever such third-party acknowledgments normally appear. * * 4. The names "Smack" and "Jive Software" must not be used to * endorse or promote products derived from this software without * prior written permission. For written permission, please * contact webmaster@jivesoftware.com. * * 5. Products derived from this software may not be called "Smack", * nor may "Smack" appear in their name, without prior written * permission of Jive Software. * * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL JIVE SOFTWARE OR * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF * SUCH DAMAGE. * ==================================================================== */ package org.jivesoftware.smack; import org.jivesoftware.smack.packet.Packet; import org.jivesoftware.smack.filter.PacketFilter; import java.util.LinkedList; /** * Provides a mechanism to collect packets into a result queue that pass a * specified filter. The collector lets you perform blocking and polling * operations on the result queue. So, a PacketCollector is more suitable to * use than a {@link PacketListener} when you need to wait for a specific * result.
* * Each packet collector will queue up to 2^16 packets for processing before * older packets are automatically dropped. * * @see XMPPConnection#createPacketCollector(PacketFilter) * @author Matt Tucker */ public class PacketCollector { /** * Max number of packets that any one collector can hold. After the max is * reached, older packets will be automatically dropped from the queue as * new packets are added. */ private static final int MAX_PACKETS = 65536; private PacketFilter packetFilter; private LinkedList resultQueue; private PacketReader packetReader; private boolean cancelled = false; /** * Creates a new packet collector. If the packet filter is null, then * all packets will match this collector. * * @param packetReader the packetReader the collector is tied to. * @param packetFilter determines which packets will be returned by this collector. */ protected PacketCollector(PacketReader packetReader, PacketFilter packetFilter) { this.packetReader = packetReader; this.packetFilter = packetFilter; this.resultQueue = new LinkedList(); // Add the collector to the packet reader's list of active collector. synchronized (packetReader.collectors) { packetReader.collectors.add(this); } } /** * Explicitly cancels the packet collector so that no more results are * queued up. Once a packet collector has been cancelled, it cannot be * re-enabled. Instead, a new packet collector must be created. */ public void cancel() { // If the packet collector has already been cancelled, do nothing. if (cancelled) { return; } else { cancelled = true; // Remove object from collectors list by setting the value in the // list at the correct index to null. The collector thread will // automatically remove the actual list entry when it can. synchronized (packetReader.collectors) { int index = packetReader.collectors.indexOf(this); packetReader.collectors.set(index, null); } } } /** * Returns the packet filter associated with this packet collector. The packet * filter is used to determine what packets are queued as results. * * @return the packet filter. */ public PacketFilter getPacketFilter() { return packetFilter; } /** * Polls to see if a packet is currently available and returns it, or * immediately returns null if no packets are currently in the * result queue. * * @return the next packet result, or null if there are no more * results. */ public synchronized Packet pollResult() { if (resultQueue.isEmpty()) { return null; } else { return (Packet)resultQueue.removeLast(); } } /** * Returns the next available packet. The method call will block (not return) * until a packet is available. * * @return the next available packet. */ public synchronized Packet nextResult() { // Wait indefinitely until there is a result to return. while (resultQueue.isEmpty()) { try { wait(); } catch (InterruptedException ie) { } } return (Packet)resultQueue.removeLast(); } /** * Returns the next available packet. The method call will block (not return) * until a packet is available or the timeout has elapased. If the * timeout elapses without a result, null will be returned. * * @param timeout the amount of time to wait for the next packet (in milleseconds). * @return the next available packet. */ public synchronized Packet nextResult(long timeout) { // Wait up to the specified amount of time for a result. if (resultQueue.isEmpty()) { try { wait(timeout); } catch (InterruptedException ie) { } } // If still no result, return null. if (resultQueue.isEmpty()) { return null; } else { return (Packet)resultQueue.removeLast(); } } /** * Processes a packet to see if it meets the criteria for this packet collector. * If so, the packet is added to the result queue. * * @param packet the packet to process. */ protected synchronized void processPacket(Packet packet) { if (packet == null) { return; } if (packetFilter == null || packetFilter.accept(packet)) { // If the max number of packets has been reached, remove the oldest one. if (resultQueue.size() == MAX_PACKETS) { resultQueue.removeLast(); } // Add the new packet. resultQueue.addFirst(packet); // Notify waiting threads a result is available. notifyAll(); } } }