Netzwerkprotokolle

Jede nicht triviale Netzwerkbibliothek nutzt Netzwerk- bzw. Anwendungsprotokolle. Sobald Sie mit Sockets arbeiten, müssen Sie in der Regel auch ein höheres Protokoll entwerfen. Ein Netzwerkprotokoll ist eine exakte Vereinbarung, nach der Daten zwischen Computern bzw. Prozessen ausgetauscht werden, die durch ein Netz miteinander verbunden sind. Auch in einem P2P-Netz muss eine Vereinbarung darüber herrschen, wie die gesendeten Daten interpretiert werden sollen, wie Daten kodiert werden, wer welche Daten sendet und wie eine Kommunikation abgebrochen wird.

Das Internet-Protokoll definiert bereits Regeln, wie Pakete innerhalb des Internets versendet und empfangen werden. Aufbauend auf dem IP befindet sich das Transmission Control Protocol (TCP) (zu dt. Übertragungssteuerungsprotokoll). Es befindet sich in der Transportschicht im OSI-Schichtenmodell.

Das TCP/IP-Protokoll transportiert Daten, ohne diese zu verändern oder zu prüfen. Dies stattet Anwendungen mit einer großen Flexibilität aus, wie sie ihre Informationen für die Übertragung kodieren. Viele Anwendungen nutzen Anwendungsprotokolle, die sich aus diskreten Nachrichten, kodiert in sequentiellen Feldern, zusammensetzen. Jedes Feld beinhaltet spezifische Informationen, kodiert als Bitsequenz. An diese Stelle tritt das Anwendungsprotokoll. Es spezifiziert wie die Bitsequenzen formatiert sind, wie sie interpretiert und ggf. geparsed werden müssen, so dass der Empfänger die Nachricht lesen kann.

Das TCP verlangt lediglich das die zu übertragenden Daten Vielfache von 8-Bit sein müssen. Auf Netzwerkebene werden alle Daten in Sequenzen von Bytes übertragen. Ein Byte sind 8-Bit und liegen in einem Bereich zwischen 0 und 255. Unabhängig von der Kodierung werden die Daten stets in Bytes übertragen.

Anwendungen, die einen TCP-Transport verwenden, senden einen fortlaufenden Bytestrom (Stream) über die aufgebaute Datenverbindung und nicht aufeinanderfolgende Einzelnachrichten. Nachrichtengrenzen werden daher auch bei einer Ende-zu-Ende-Verbindung nicht bewahrt.

Ein Problem in TCP ist, das TCP keine Nachrichtengrenzen kennt. Das Framing beschreibt das Problem des Empfängers, den Beginn und das Ende der diskreten Nachrichten innerhalb des Streams im TCP-Buffer zu lokalisieren. Ob die Information als Text in ASCII kodiert ist, als Multibyte in Unicode oder in einer Kombination aus beidem - das Anwendungsprotokoll muss den Empfänger der Nachricht in die Lage versetzen zu entscheiden wann die komplette Nachricht empfangen wurde und wie diese zurück in ihre Felder geparsed werden kann.

Falls die Felder innerhalb einer Nachricht stets eine feste Größe haben und die Nachricht ebenfalls immer aus einer festgelegten Anzahl von Feldern besteht, dann ist die Größe der Nachricht bekannt und der Empfänger muss nur noch die erwartete Zahl an Bytes in den byte[] Puffer laden. Sobald aber die Nachrichtengröße variiert, muss die Nachricht Framing-Informationen beinhalten. Es gibt in der Regel zwei unterschiedliche Ansätze dies zu bewerkstelligen.

• Delimiter: Das Ende der Nachricht, mit variabler Länge, wird durch eine einzigartige Zeichenfolge oder ein einzigartiges Trennzeichen beschrieben.
• Explizite Länge: Die Länge der Nachricht wird zu Beginn der Transmission in einer zuvor definierten Weise übermittelt.

Der Ansatz mithilfe eines Delimiters wird oft bei textbasierten Nachrichten gewählt. In dem Bild oben, wurde ein mit ASCII kodierter Text versendet. Als Delimiter wurde ein einzelnes Zeichen, nämlich die Raute gewählt. Sobald komplexe Datenströme übermittelt werden, ist der Ansatz mit einem Delimiter allerdings ineffizient und unzureichend. Wir werden in unserem Protokoll deshalb den zweiten Ansatz wählen. Dazu wird zunächst eine sogenannte Message (zu dt. Nachricht) entworfen.

Nachrichten

Netzwerkbibliotheken müssen Nachrichten übertragen können. Viele Anwendungen, wie der Bittorrent Client Azureus oder ICQ verwenden Nachrichten. Das Instant-Messaging-Programm ICQ basiert auf einem komplexen Anwendungsprotokoll, namens OSCAR. Das OSCAR-Protokoll ist sehr umfassend und wurde über Jahre hinweg entwickelt. Die offizielle Dokumentation des OSCAR-Protokolls ist auf http://dev.aol.com/aim/oscar/ zu finden. Falls Sie an komplexen Anwendungsprotokollen von großen Programmen interessiert sind, können Sie sich dort mit dem Aufbau eines solchen Protokolls vertraut machen.

Auch das OSCAR-Protokoll definiert Nachrichten und Nachrichtentypen. So definiert beispielsweise die Konstante MTYPE_FILEREQ im Feld 0x03 einen File Request in ICQ. Die hier vorgestellte Netzwerkbibliothek wird ebenfalls auf Nachrichten aufbauen. Das konkrete Anwendungsprotokoll soll zu einem späteren Zeitpunkt definiert werden können. Zunächst muss ein Peer im Netzwerk in die Lage versetzt werden Nachrichten zu senden und wieder zu empfangen. Wir definieren zuerst ein Interface, das den Namen Message trägt und alle Methoden definiert, die eine reale Nachricht in unserem Netzwerk bereitstellen muss.

public interface Message
{   
    /**
     * Get message type.
     *
     * @return type the message type
     */
    String getType();
 
    /**
     * Get textual data of this particular message.
     *
     * @return  the message data
     */
    String getData();
 
    /**
     * Returns a list with tokens. If there are more messages in the message
     * data buffer, this method will tokenize the buffer and return the
     * tokens in a list.
     *
     * @return  a list of byte[] tokens
     */
    List<byte[]> getByteTokens();
 
    /**
     * Allows to clone an object.
     *
     * @return  an object copy
     * @throws java.lang.CloneNotSupportedException
     */
    Object clone() throws CloneNotSupportedException;
 
    /**
     * Returns the total number of data tokens stored in the data buffer.
     *
     * @return  num of tokens
     */
    int countTokens();
 
    /**
     * Returns a packed representation of this message as an
     * array of bytes.
     *
     * @return byte array of message data
     */
    byte[] toBytes() throws MessageException;
 
    /**
     * Returns a list with String tokens from the message data buffer.
     *
     * @return  the message data tokens
     */
    List<String> getStringTokens();
 
    /**
     * Returns the bytes of the message data.
     *
     * @return the message data bytes
     */
    byte[] getDataBytes();
 
    /**
     * Returns the message type bytes.
     *
     * @return the message type (4-byte array)
     */
    byte[] getTypeBytes();
 
    /**
     * General delimiter for all messages. Do not choose a regex metacharacter.
     *
     * @return the delimiter
     */
    String getDelimiter();
 
    /**
     * Overrides toString().
     *
     * @return  the string
     */
    @Override
    String toString();
 
    /**
     * Gets the protocol payload of this message in bytes.
     *
     * @return  total protocol payload
     */
    int getProtocolPayload();
 
    /**
     * Gets the data payload of this message in bytes.
     *
     * @return  total data payload
     */
    int getDataPayload();  
}

Sie können an den ersten beiden Methoden im Interface zwei wichtige Funktionen erkennen, die eine Nachricht besitzen muss. Eine Nachricht im Battleship-Netzwerk besteht aus einem Typ und aus Daten. Gleichzeitig muss eine Nachricht die Methode toBytes implementieren, so dass die Nachricht als Bytefolge versendet werden kann.

Die konkrete Nachricht hat den folgenden Aufbau in unserem System.

Der Kopfbereich (Header) besteht aus 64-Bit. Die ersten 32-Bit beschreiben den Nachrichtentyp. Die nächsten 32-Bit stehen für die Nachrichtenlänge. Darauf folgen die eigentlichen Daten der Nachricht. Sie können theoretisch eine Länge von 32-Bit haben. Das entspricht einem Integer in Java. Eine Nachricht kann also maximal 4 Gigabyte an Daten fassen.

Die Implementierung einer konkreten Nachrichtenklasse gestaltet sich sehr einfach, aufbauend auf dem zuvor definierten Interface. Die Klasse PeerMessage implementiert das Interface Message und stellt alle Methoden bereit, die für das Versenden von Nachrichten erforderlich sind.

public final class PeerMessage implements Message, Cloneable
{
    /**
     * Constructs a new PeerMessage object.
     *
     * @param type the message type (4 ascii characters)
     * @param pid  the peer identification of the sender
     * @param data the message data
     */
    public PeerMessage(String type, String pid, byte[] data)
    {
        this( type.getBytes(), pid, data );
    }
 
 
    /**
     * Constructs a new PeerMessage object by reading data
     * from the given socket connection. Automatically decodes the
     * message data.
     * 
     * @param s a socket connection object
     * @throws IOException if I/O error occurs
     */
    public PeerMessage(SocketInterface s) throws IOException 
    {
        type_ = new byte[4];
        byte[] length = new byte[4];    // for reading length of message data
 
        // Check if it is a valid message type
        if( s.read( type_ ) != 4 )
            throw new IOException( "PeerMessage IOException: Undefined message type."  );
 
        // Check if we now can read 4 data bytes, holding the *size* 
        // of the data in the message
        if( s.read( length ) != 4 )
            throw new IOException( "PeerMessage IOException: Undefined data length." );
 
        // Convert the 4 bytes to an integer
        int size = Bytes.byteArrayToInt( length );
 
        // Reserve space for the message
        data_ = new byte[size];
 
        // Read it...
        if( s.read( data_ ) != size )
            throw new IOException( "PeerMessage IOException: Unexpected message data length." );
    }  
 
    /** 
     * Returns the message type as a String.
     * 
     * @return the message type (4-character String)
     */
    @Override
    public String getType()
    {
        return new String( type_ );
    }    
 
    /**
     * Returns a copy of the message type buffer.
     * 
     * @return the message type (4-byte array)
     */
    @Override
    public byte[] getTypeBytes()
    {
        return (byte[])type_.clone();
    }
 
    /**
     * <p>Retrieves the <b>maximum allowed size</b> of a message. This is necessary
     * if a block compressing algorithm is used at the socket layer,
     * transmitting a PeerMessage by wire.</p>
     *
     * <p>Since a message is an atomic object, the message is not allowed to be
     * bigger than the compressed block size. Otherwise the message length
     * would not match the received length, causing an invalid message
     * length exception.</p>
     *
     * @return  maximum message size
     */
    public static int getMaxSize()
    {
        return CompressedSocket.DEFAULT_SYNC_SIZE_GZIP_BYTES;
    }
 
    /** The type buffer (header composed of a 4-byte identifier) */
    private byte[] type_;
 
    /** The data buffer */
    private byte[] data_;
 
    // Rest siehe Quellcode...
}

Die Klasse hat diverse Konstruktoren, einige sind im Code oben gelistet. Ein wichtiger Konstruktor nimmt als Argument das SocketInterface entgegen. Aus dem Socket wird anschließend mit read die Nachricht geparsed. Zunächst werden die ersten 4-Byte eingelesen, dann die zweiten 4-Byte. Anschließend wird die in Big-Endian vorliegende Länge mithilfe der statischen Methode byteArrayToInt aus der Klasse Bytes in einen Integer transformiert, so dass anschließend n Bytes aus dem restlichen Stream gelesen werden können. Die Nachricht wurde erfolgreich aus dem TCP-Buffer konstruiert.

Handler

Bei der Implementierung des Servers wurde eine innere Klasse mit dem Namen PeerHandler definiert. Die Klasse PeerHandler hat Nachrichten in Abhängigkeit des Nachrichtentyps an bestimmte Handler weitergeleitet. Bei einem Handler handelt es sich um eine Aufrufroutine, bei der Nachrichten automatisch an die richtigen Methoden weitergeleitet werden.

In der Basisklasse Node, in welcher auch die Klasse PeerHandler implementiert ist, wurde für die Verwaltung der Handler eine spezielle HashMap verwendet.

    /** 
     * A hash table supporting full concurrency containing key-value pairs
     * of unique commands and handlers
     */
    private final ConcurrentMap<String, MessageHandlerInterface> handlers_ =
            new ConcurrentHashMap<String, MessageHandlerInterface>();

Bei der ConcurrentHashMap handelt es sich um eine Thread-sichere Collection. Diese besonderen Collections sind im Package java.util.concurrent versammelt. In Multi-Threaded Applikationen ist es von entscheidender Bedeutung Zugriffe von Threads zu koordinieren. Geschieht das nicht, erhält man früher oder später mit Sicherheit eine ConcurrentModificationException sobald mehr als ein Thread versucht auf die Collection schreibend zuzugreifen. Im schlimmsten Fall erhalten Sie keine ConcurrentModificationException und Ihre Anwendung produziert mysteriöse Fehler.

Die ausgeklügelten Algorithmen dieser speziellen Collections blockieren niemals die gesamte Tabelle und minimieren Konkurrenzsituationen, indem sie gleichzeitigen Zugriff auf verschiedene Teile der Datenstruktur zulassen. Da unser Server mithilfe von Thread-Pools automatisch beliebig viele neue Threads produziert, muss sichergestellt sein, dass sich diese Threads nicht in die Quere kommen. Die ConcurrentHashMap hat nützliche Methoden für atomares Einfügen und Entfernen von Zuordnungen.

Der PeerHandler ruft je nach Typ der Nachricht das MessageHandlerInterface auf. Dieses Interface hat nur eine einzige Methode, die von allen Handlern implementiert werden muss, nämlich handleMessage. Die Methode nimmt die aktuelle Verbindung, sowie die Nachricht entgegen.

public interface MessageHandlerInterface
{
    /**
     * Invoked when a peer connects and sends a message to this node.
     * 
     * @param peerconn  the peer connection
     * @param msg   the message
     */
    public void handleMessage(PeerConnection peerconn, Message msg);
}

In der Klasse Node sind weitere wichtige Methoden und innere Klassen definiert. Neben einer Stabilisierungsroutine, die kontinuierlich Teilnehmer im Netzwerk anpingt und bei gescheiterter Verbindung aus der eigenen Peerliste entfernt, sind auch Methoden zum Versenden von Daten in dieser abstrakten Klasse implementiert. Die Klasse verfügt über eine Startmethode, namens start. Diese Methode startet die wesentlichen Threads der Klasse Node und muss nach der Instanzierung der Subklasse aufgerufen werden.

Die Klasse Node ist bekanntlich abstrakt. Sie erweitert, wie das Model, die Java-Klasse Observable. Von Node leiten wir nun eine konkrete Klasse für unsere Anwendung ab. Diese Klasse nennen wir BattleshipNode.

public final class BattleshipNode extends Node
{
    /**
     * Constructs a new battleship node.
     * 
     * @param maxPeers  the max. allowed number of peers
     * @param myInfo    the own peer info (host, port)
     */
    public BattleshipNode(int maxPeers, PeerInfo myInfo) 
    {
        super( myInfo );
 
        maxPeers_ = maxPeers;
 
        addRouter( new SimpleRouter( this ) );
 
        // Add protocol handlers to the message types
        addHandler( MessageType.ERROR.toString(), new ErrorHandler( this ) );
        addHandler( MessageType.SHOT.toString(), new ShotHandler( this ) );
        addHandler( MessageType.JOIN.toString(), new JoinHandler( this ) );
        addHandler( MessageType.PEERINFO.toString(), new PeerInfoHandler( this ) );
        addHandler( MessageType.LISTPEERS.toString(), new ListPeersHandler( this ) );
        addHandler( MessageType.TEXTMESSAGE.toString(), new TextMessageHandler( this ) );
        addHandler( MessageType.NEWGAME.toString(), new NewGameHandler( this ) );
        addHandler( MessageType.FLEET.toString(), new FleetHandler( this ) );
        addHandler( MessageType.NEWGAMEACCEPT.toString(), new NewGameAcceptedHandler( this ) );
        addHandler( MessageType.STYLEDDOCUMENT.toString(), new StyledDocumentHandler( this ) );
        addHandler( MessageType.QUIT.toString(), new QuitHandler( this ) );
    }
 
    /**
     * <p>This enum handles the common 4-byte message types, used in the
     * peer-to-peer protocol. This types can be viewed as a string. Each type
     * identifies the various messages exchanged in the system. When the 
     * NodeService of a peer receives a message, it dispatches it to the
     * appropriate handler based on the message type.</p>
     */    
    public enum MessageType
    {
        /** Indicates an error message */
        ERROR             ("ERRO"),
        /** A fleet message with information about all ships */
        FLEET             ("FLET"),        
        /** Indicates a welcome hello packet */
        HELLO             ("HELO"),
        /** Indicates a joining peer */
        JOIN              ("JOIN"),
        /** Indicates the peer listing */
        LISTPEERS         ("LIST"),
        /** Indicates that a peer wants to start a new game with us */
        NEWGAME           ("NEWG"),      
        /** Indicates that we accept a new game */
        NEWGAMEACCEPT     ("NWGA"),         
        /** Indicates a peer info */
        PEERINFO          ("PIFO"),
        /** Indicates that this peer wants to quit */
        QUIT              ("QUIT"),
        /** Indicates a reply message */
        REPLY             ("REPL"),
        /** Indicates a shot message */
        SHOT              ("SHOT"),
        /** Indicates a serialized styled document */
        STYLEDDOCUMENT    ("SDOC"),
        /** Indicates a simple text message */
        TEXTMESSAGE       ("TMSG");
 
        MessageType(String type) 
        {
            if( type.length() == 4 ) {
                type_ = type;
            } else {
                type_ = "####";
            }
        }
 
        @Override
        public String toString()
        { 
            return type_; 
        }
 
        /** The message type */
        private final String type_;
    }
 
    // ...

Die Klasse BattleshipNode implementiert alle wichtigen Handler und definiert auch alle Nachrichtentypen, die von unserem Anwendungsprotokoll gebraucht werden. Mithilfe einer Enumeration namens MessageType werden die jeweiligen Typen definiert. Es handelt sich entsprechend der Vereinbarung um 4-Byte große Strings. Die Enumeration kann beliebig um weitere Nachrichtentypen erweitert werden.

Im Konstruktor der Klasse BattleshipNode werden für alle Nachrichtentypen die Handler definiert. Ein Handler wird als innere Klasse von BattleshipNode definiert. Eingehende Nachrichten werden auf diese Weise, abhängig vom Typ, direkt an die jeweilige Handlerroutine weitergeleitet. An dieser Stelle können Sie einen Blick auf einen Handler werfen. Dieser wird aufgerufen, sobald eine Nachricht vom Typ »TEXTMESSAGE« empfangen wird.

/**
 * Handles simple text messages. If a peer sends a message
 * to us, this handler will be called.
 * 
 * Message syntax: TYPE pid message
 */
private class TextMessageHandler implements MessageHandlerInterface
{     
    public TextMessageHandler(Node peer)
    { 
        peer_ = peer; 
    }
 
    @Override
    public void handleMessage(PeerConnection peerconn, Message msg) 
    {
        // Check for correct number of arguments 
        List<String> data = msg.getStringTokens();
 
        if( data.size() != 2 ) {
            Message reply = new PeerMessage( MessageType.ERROR.toString(), peer_.getMyInfo().getId(),
                    msg.getType() + ": " + "Incorrect arguments." );
            peerconn.sendData( reply );
            return;
        }            
 
        // Put the message in the queue
        addMessage( msg );
    }
 
    private Node peer_;
}

Sie können bereits jetzt erkennen, dass das System sehr flexibel konzipiert ist. Die Netzwerkbibliothek erlaubt es beliebige Nachrichtentypen zu definieren, so dass ein ganz individuelles Anwendungsprotokoll entworfen werden kann.

Peers

Die Klasse BattleshipNode implementiert einige Methoden der abstrakten Superklasse Node. Neben den Handlern verwaltet die Klasse auch eine Tabelle mit den Peers. Alle Netzwerkteilnehmer werden anhand ihrer IP-Adresse und Portnummer identifiziert. Nimmt ein neuer Netzwerkteilnehmer Kontakt mit uns auf, wird seine Adresse in einer gewöhnlichen Hashtabelle gespeichert.

In P2P-Systemen kommen oft sogenannte verteilte Hashtabellen (engl. distributed hash table) zum Einsatz. Es handelt sich dabei um eine Datenstruktur, die versucht, das allgemeine Problem in P2P-Systemen – das Finden des Speicherorts einer gesuchten Datei – mit möglichst geringem Aufwand effizient zu lösen und zu dezentralisieren, um es von Ausfall, zum Beispiel durch Abschalten eines Trackers unabhängig zu gestalten. Kademlia ist ein bekanntes Beispiel für ein Protokoll das auf einer DHT basiert.

Die Datenobjekte sollen in einer DHT möglichst gleichmäßig über die Knotenmenge verteilt sein und ein von jedem beliebigen Einstiegsort ortsunabhängiges Routing zum verantwortlichen Knoten ermöglichen. Jeder Knoten ist dabei analog zu einem Behälter einer Hashtabelle. Die Datenstruktur muss ständige Anpassungen durch Ausfall, Beitritt und Austritt von Knoten überstehen, sich selbst organisieren und skalierbar sein. Die Grundlage für verteilte Hashtabellen bilden konsistente Hash-Funktionen, die für alle gesuchten Werte bzw. Dateien in einem P2P-System einen eindeutigen Hash abbilden.

Für die Verwaltung der Peers und der empfangenen Nachrichten in Battleship, werden zwei Thread-sichere Collections benötigt. Während die ConcurrentHashMap alle Peerdaten speichert, werden Nachrichten in einer ConcurrentLinkedQueue hinterlegt. Es handelt sich dabei um eine Warteschlange, die auf dem FIFO-Prinzip beruht.

/**
 * A hash table supporting full concurrency containing key-value
 * pairs of unique id's and the PeerInfo's.
 */
private final ConcurrentMap<String, PeerInfo> peers_= new ConcurrentHashMap<String, PeerInfo>();
 
/**
 * An efficient scalable thread-safe non-blocking FIFO queue. This
 * queue will hold all messages send by other p2p network members.
 */
private final ConcurrentLinkedQueue<Message> queue_ = new ConcurrentLinkedQueue<Message>();

Alle Peers im Netzwerk werden in der speziellen HashMap gespeichert. Der Schüssel ist eine eindeutige ID, der Wert ein Objekt vom Typ PeerInfo. Die Klasse PeerInfo speichert relevante Daten von einem Peer. Dazu zählt neben der IP-Adresse auch die Portnummer und der Name. In der Klasse sind weitere Felder definiert. Auch diese Klasse kann bei Bedarf erweitert werden. In BattleshipNode sind entsprechende Methoden implementiert, die es der Anwendung gestatten Peers in das eigene Netzwerk aufzunehmen. Eine Methode trägt den Namen addPeer.

/**
 * Add new peer information to the peer list, indexed by the given
 * key.
 *
 * @param key the key associated with the peer
 * @param peerInfo the peer information
 * @return true if successful; false if maxPeers is reached, or if the
 * peer list already contains the given key, or if peer is corrupt.
 */
@Override
public boolean addPeer(String key, PeerInfo peerInfo)
{
    // Before we add a new peer, sanitize variables. Call the class check.
    if( !peerInfo.sanitize() ) return false;
 
    if( (maxPeers_ == 0 || peers_.size() < maxPeers_) && !peers_.containsKey( key ) ) {
        peers_.put( key, peerInfo );
        setChanged();   // New peers, tell observers
        notifyObservers();
        return true;
    }
 
    return false;
}

Die Methode fügt einen neuen Teilnehmer in die Map ein. Sie können hier erstmals zwei Methoden der Klasse Observable erkennen. Sobald ein neuer Peer in die Liste eingetragen wurde, werden alle Beobachter der Klasse davon unterrichtet. Dies erreicht man durch den Aufruf von setChanged und dem anschließenden Aufruf von notifyObservers.

Neben den Peers, stehen auch etliche Methoden für den Zugriff auf die empfangenen Nachrichten zur Verfügung. Die Methode getMsgOfType ist eine davon. Sie liefert die neueste Nachricht eines bestimmten Typs zurück. Bei Bedarf wird die Nachricht aus der Warteschlange endgültig entfernt.

/**
 * Gets the first message of a given type from the message queue. Removes
 * this message from the queue, if <code>remove</code> is true.
 *
 * @param mt    the message type
 * @param remove    if true, the message will be removed from the queue
 * @return  the last message of the given type; null if no such message
 */
public Message getMsgOfType(MessageType mt, boolean remove)
{
    Message message = null;
 
    for( Message msg : queue_ ) {
        if( msg.getType().equals( mt.toString() ) ) {
            message = msg;
            if( remove ) {
                queue_.remove( msg );   // Iterator.remove is safe
            }
            break;
        }
    }
 
    return message;
}