/*
 *  PHEX - The pure-java Gnutella-servent.
 *  Copyright (C) 2001 - 2007 Phex Development Group
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * 
 *  --- SVN Information ---
 *  $Id: MessageDispatcher.java 3859 2007-07-01 20:15:19Z gregork $
 */
package phex.msg;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import phex.common.HorizonTracker;
import phex.common.PongCache;
import phex.common.QueryRoutingTable;
import phex.common.ThreadPool;
import phex.common.address.DefaultDestAddress;
import phex.common.address.DestAddress;
import phex.common.address.IpAddress;
import phex.common.format.NumberFormatUtils;
import phex.connection.NetworkManager;
import phex.host.Host;
import phex.host.HostManager;
import phex.msg.vendor.CapabilitiesVMsg;
import phex.msg.vendor.HopsFlowVMsg;
import phex.msg.vendor.MessagesSupportedVMsg;
import phex.msg.vendor.PushProxyAcknowledgementVMsg;
import phex.msg.vendor.PushProxyRequestVMsg;
import phex.msg.vendor.TCPConnectBackRedirectVMsg;
import phex.msg.vendor.TCPConnectBackVMsg;
import phex.msg.vendor.VendorMsg;
import phex.net.connection.Connection;
import phex.net.connection.ConnectionFactory;
import phex.prefs.core.BandwidthPrefs;
import phex.query.DynamicQueryConstants;
import phex.query.QueryHistoryMonitor;
import phex.query.QueryManager;
import phex.security.AccessType;
import phex.security.PhexSecurityManager;
import phex.share.ShareFile;
import phex.share.ShareManager;
import phex.share.SharedFilesService;
import phex.statistic.MessageCountStatistic;
import phex.upload.PushWorker;
import phex.utils.HexConverter;
import phex.utils.NLogger;
import phex.utils.QueryGUIDRoutingPair;

public class MessageDispatcher
{
    private final SharedFilesService sharedFilesService;
    private final MsgManager messageMgr;
    private final HostManager hostMgr;
    private final QueryHistoryMonitor queryHistory;
    private final NetworkManager networkMgr;
    private final PhexSecurityManager securityManager;
    
    public MessageDispatcher()
    {
        messageMgr = MsgManager.getInstance();
        hostMgr = HostManager.getInstance();
        sharedFilesService = ShareManager.getInstance().getSharedFilesService();
        networkMgr = NetworkManager.getInstance();
        queryHistory = QueryManager.getInstance().getQueryHistoryMonitor();
        securityManager = PhexSecurityManager.getInstance();
    }
    
    public void handlePing( PingMsg pingMsg, Host sourceHost )
    {
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
            NLogger.debug( MessageDispatcher.class, "Received Ping: "
            + pingMsg.getDebugString() + " - " + pingMsg.getHeader().toString());

        // count ping statistic
        MessageCountStatistic.pingMsgInCounter.increment( 1 );

        MsgHeader header = pingMsg.getHeader();

        // Bearshare seems to massively send pings with ttl 6 hops 1 and empty 
        // GUID. We currently drop them all since they are duplicates...
        // if ( header.getMsgID().equals(GUID.EMPTY_GUID) )
        
        // See if I have seen this Ping before.  Drop msg if duplicate.
        if ( !messageMgr.checkAndAddToPingRoutingTable( header.getMsgID(),
                sourceHost ) )
        {
            dropMessage( pingMsg, "Dropping already seen ping", sourceHost );
            return;
        }

        // we dont forward pings... pong caching is handling this..
        // if ( pingMsg.getHeader().getTTL() > 0 )
        // {
        //     messageMgr.forwardPing( pingMsg, connectedHost );
        // }
        
        respondToPing( pingMsg, sourceHost );
    }
    
    private void respondToPing( PingMsg pingMsg, Host sourceHost )
    {
        MsgHeader header = pingMsg.getHeader();
        // to reduce the incomming connection attemps of other clients
        // only response to ping a when we have free incomming slots or this
        // ping has a original TTL ( current TTL + hops ) of 2.
        byte ttl = header.getTTL();
        byte hops = header.getHopsTaken();
        if ( ( ttl + hops > 2 ) && !hostMgr.areIncommingSlotsAdvertised() )
        {
            return;
        }

        // For crawler pings (hops==1, ttl=1) we have a special treatment...
        // We reply with all our leaf connections... in case we have them as a
        // ultrapeer...
        if ( hops == 1 && ttl == 1)
        {// crawler ping
            // respond with leaf nodes pongs, already "hoped" one step. (ttl=1,hops=1)
            Host[] leafs = hostMgr.getNetworkHostsContainer().getLeafConnections();
            for ( int i = 0; i < leafs.length; i++ )
            {
                DestAddress ha = leafs[i].getHostAddress();
                PongMsg pong = PongMsg.createOtherLeafsOutgoingPong( header.getMsgID(),
                    (byte)1, (byte)1, ha );
                sourceHost.queueMessageToSend( pong );                
            }
        }

        // send back my own pong
        byte newTTL = hops++;
        if ( ( hops + ttl ) <= 2)
        {
            newTTL = 1;
        }
        
        // Get my host:port for InitResponse.
        PongMsg pong = PongMsg.createMyOutgoingPong( header.getMsgID(), newTTL );
        sourceHost.queueMessageToSend( pong );
        
        // send pongs from pong cache
        DestAddress orginAddress = sourceHost.getHostAddress();
        IpAddress ip = orginAddress.getIpAddress();
        if ( ip == null )
        {
            return;
        }
        GUID guid = header.getMsgID();
        List<PongMsg> pongs = PongCache.getInstance().getPongs( );
        for( PongMsg pMsg : pongs )
        {
            if( Arrays.equals( pMsg.getIP(), ip.getHostIP() ) )
            {
                continue;
            }
            sourceHost.queueMessageToSend( PongMsg.createFromCachePong(guid, newTTL, pMsg) );
        }
    }
    
    /**
     * 
     */
    public void handlePong( PongMsg pongMsg, Host sourceHost )
    {
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
            NLogger.debug( MessageDispatcher.class, "Received Pong: "
            + pongMsg.getDebugString() + " - " + pongMsg.getHeader().toString());

        // count pong statistic
        MessageCountStatistic.pongMsgInCounter.increment( 1 );
        HorizonTracker.getInstance().trackPong( pongMsg );

        MsgHeader header = pongMsg.getHeader();

        byte[] pongIP = pongMsg.getIP();
        AccessType access = securityManager.controlHostIPAccess( pongIP );
        if ( access == AccessType.ACCESS_STRONGLY_DENIED )
        {
            // drop message
            dropMessage( pongMsg, "IP access strongly denied.", sourceHost );
            return;
        }

        // add address to host catcher...
        if ( access == AccessType.ACCESS_GRANTED )
        {
            boolean isNew = hostMgr.getCaughtHostsContainer().addCaughtHost( pongMsg );
            if ( isNew )
            {
                PongCache.getInstance().addPong( pongMsg );
            }
        }

        // this port is unsinged and always valid.
        int pongPort = pongMsg.getPort();
        byte hopsTaken = pongMsg.getHeader().getHopsTaken();
        
        // check if this is the response to my Ping message
        if ( hopsTaken == 1 )
        {
            DestAddress connectedAddress = sourceHost.getHostAddress();
            byte[] connectedIP = connectedAddress.getIpAddress().getHostIP();
            if ( Arrays.equals( connectedIP, pongIP ) )
            {
                sourceHost.setFileCount( pongMsg.getFileCount() );
                sourceHost.setTotalFileSize( pongMsg.getFileSizeInKB() );
                // I guess that a hops == 1 with equal ip address is a pong from my
                // direct neighbor, therefore I also update the obviously wrong port.
                int connectedPort = connectedAddress.getPort();
                if ( connectedPort != pongPort )
                {
                    connectedAddress.setPort( pongPort );
                }
            }
        }
        
        
        // Did I forward that Pong GUID before?
        Host returnHost = messageMgr.getPingRouting( header.getMsgID() );
        if ( returnHost == null || returnHost == Host.LOCAL_HOST )
        { // pong was for me or timed out.
            return;
        }

        // Ok, I did forward the Init msg on behalf of returnHost.
        // The InitResponse is for returnHost.  Better route it back.
        if ( pongMsg.getHeader().getTTL() > 0 )
        {
            returnHost.queueMessageToSend( pongMsg );
        }
    }
    
    public void handleQuery( QueryMsg msg, Host sourceHost )
    {
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
            NLogger.debug( MessageDispatcher.class, "Received Query: "
            + msg.toString() + " - " + msg.getHeader().toString());
        
        // count query statistic
        MessageCountStatistic.queryMsgInCounter.increment( 1 );

        MsgHeader header = msg.getHeader();

        // See if I have seen this Query before.  Drop msg if duplicate.
        // This drop is done even though this could be an extension of a 
        // probe query. Only Limewire is doing this extension of a probe
        // query currently and as stated by themselfs the efficency of it
        // is doubtfull. 
        if ( !messageMgr.checkAndAddToQueryRoutingTable( header.getMsgID(),
                sourceHost ) )
        {
            dropMessage( msg, "Dropping already seen query", sourceHost );
            return;
        }
        
        // a leaf is not supposed to forward me queries not comming from iteself.
        if ( sourceHost.isUltrapeerLeafConnection() && header.getHopsTaken() > 2 )
        {
            dropMessage( msg, "Dropping Query from Leaf with hops > 2.", sourceHost );
        }

        // logging a msg can be very expensive!
        //mRemoteHost.log( Logger.FINEST, "Received Msg: " + msg + " Hex: " +
        //    HexConverter.toHexString( body ) + " Data: " + new String( body) );

        // Add to the net search history.
        queryHistory.addSearchQuery( msg );

        // TTL > 0 checks for querys depends on routing
        messageMgr.forwardQuery( msg, sourceHost );
        
        // Search the sharefile database and get groups of sharefiles.
        List<ShareFile> resultFiles = sharedFilesService.handleQuery( msg );
        if ( resultFiles == null || resultFiles.size() == 0)
        {
            return;
        }
        respondToQuery( header, resultFiles, sourceHost );
    }

    private void respondToQuery( MsgHeader header, List<ShareFile> resultFiles, 
        Host sourceHost )
    {
        // Construct QueryResponse msg.  Copy the original Init's GUID.
        // TTL expansion on query hits doesn't matter very much so it doesn't
        // hurt us to give query hits a TTL boost.
        // Bearshare sets QueryHit TTL to 10
        // gtk-gnutella sets QueryHit TTL to (hops + 5)
        MsgHeader newHeader = new MsgHeader( header.getMsgID(),
            MsgHeader.QUERY_HIT_PAYLOAD,
            // Will take as many hops to get back.
            // hops + 1 decided in gdf 2002-12-04
            (byte)(header.getHopsTaken() + 1),
            //(byte)(Math.min( 10, header.getHopsTaken() + 5 ) ),
            (byte)0, 0 );

        int resultCount = resultFiles.size();
        if ( resultCount > 255 )
        {
            resultCount = 255;
        }

        QueryResponseRecord[] records = new QueryResponseRecord[ resultCount ];
        QueryResponseRecord record;
        int recPos = 0;
        for( ShareFile shareFile : resultFiles )
        {
            record = QueryResponseRecord.createFromShareFile( shareFile );
            records[ recPos ] = record;
            recPos ++;
        }

        DestAddress hostAddress = networkMgr.getLocalAddress();
        QueryResponseMsg response = new QueryResponseMsg(
            newHeader, networkMgr.getServentGuid(), hostAddress,
            Math.round( BandwidthPrefs.MaxUploadBandwidth.get().floatValue() / NumberFormatUtils.ONE_KB ),
            records );

        sourceHost.queueMessageToSend( response );
    }
    
    public void handleQueryResponse( QueryResponseMsg queryResponseMsg, Host sourceHost )
    {
        // Logging is expensive...
//        if ( Logger.isLevelLogged( Logger.FINEST ) )
//        {
//            Logger.logMessage( Logger.FINEST, Logger.NETWORK,
//                connectedHost, "Received QueryResponse: " + queryResponseMsg + " - " +
//                queryResponseMsg.toDebugString() );
//        }
        
        // count query hit statistic
        MessageCountStatistic.queryHitMsgInCounter.increment( 1 );

        DestAddress queryAddress = queryResponseMsg.getDestAddress();
        AccessType access = securityManager.controlHostAddressAccess( queryAddress );
        if ( access == AccessType.ACCESS_STRONGLY_DENIED )
        {
            // drop message
            dropMessage( queryResponseMsg, "IP access strongly denied.", sourceHost );
            return;
        }

        
        if ( access == AccessType.ACCESS_GRANTED )
        {
            try
            {
                messageMgr.processQueryResponse( sourceHost, queryResponseMsg );
            }
            catch ( InvalidMessageException exp )
            {// drop invalid message
                dropMessage(queryResponseMsg, exp.getMessage(), sourceHost);
                return;
            }
        }
        messageMgr.addToPushRoutingTable( queryResponseMsg.getRemoteClientID(),
                sourceHost );
        
        MsgHeader responseHeader = queryResponseMsg.getHeader();
        
        // check if I forwarded the Query with the same message id as this QueryResponse. 
        QueryGUIDRoutingPair routingPair = null;
        try
        {
            routingPair = messageMgr.getQueryRouting(
                responseHeader.getMsgID(), queryResponseMsg.getUniqueResultCount() );
        }
        catch ( InvalidMessageException exp )
        {// drop invalid message
            dropMessage(queryResponseMsg, exp.getMessage(), sourceHost);
            return;
        }
        if ( routingPair == null )
        {
            return;
        }
        Host returnHost = routingPair.getHost();
        
        // This QueryResponse needs to be routed to returnHost, I forwarded the query
        // on behalf of returnHost. The message is routed back to returnHost
        // if there is enough ttl left and I'm not the return host
        // and I have not yet routed enough results back.
        if ( responseHeader.getTTL() > 0 && returnHost != Host.LOCAL_HOST &&
            routingPair.getRoutedResultCount() < DynamicQueryConstants.DESIRED_ULTRAPEER_RESULTS )
        {
            returnHost.queueMessageToSend( queryResponseMsg );
        }
    }
    
    public void handleRouteTableUpdate( RouteTableUpdateMsg message, Host sourceHost )
    {
        // no specific stat so count to total
        MessageCountStatistic.totalInMsgCounter.increment( 1 );
        if ( !(sourceHost.isQueryRoutingSupported() ||
                sourceHost.isUPQueryRoutingSupported()) )
        {
            dropMessage( message, "QRP not supported from host.", sourceHost);
            return;
        }

        QueryRoutingTable qrTable = sourceHost.getLastReceivedRoutingTable();
        if ( qrTable == null )
        {
            // create new table... TODO3 maybe makes not much sense because we might
            // recreate table. maybe there is a way to initialise the qrt lazy
            qrTable = new QueryRoutingTable();
            sourceHost.setLastReceivedRoutingTable( qrTable );
        }
        try
        {
            qrTable.updateRouteTable( message );
            if ( sourceHost.isUltrapeerLeafConnection() )
            {// in case this is a leaf connection, we need to update our
             // local query routing table. This needs to be done since
             // have our leaves QRT aggregated our QRT and are checking
             // during a query against our QRT if leaves might have a hit.
                messageMgr.triggerQueryRoutingTableUpdate();
            }
        }
        catch ( InvalidMessageException exp )
        {// drop message
            dropMessage( message, "Invalid QRT update message.", sourceHost );
        }
    }
    
    public void handleVendorMessage( VendorMsg vendorMsg, Host sourceHost )
    {
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
            NLogger.debug( MessageDispatcher.class, "Received VendorMsg: "
            + vendorMsg.toString() + " - " + vendorMsg.getHeader().toString());
        
        if ( vendorMsg instanceof MessagesSupportedVMsg )
        {
            handleMessagesSupportedVMsg( (MessagesSupportedVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof TCPConnectBackVMsg )
        {
            handleTCPConnectBackVMsg( (TCPConnectBackVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof TCPConnectBackRedirectVMsg )
        {
            handleTCPConnectBackRedirectVMsg((TCPConnectBackRedirectVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof PushProxyRequestVMsg )
        {
            handlePushProxyRequestVMsg( (PushProxyRequestVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof PushProxyAcknowledgementVMsg )
        {
            handlePushProxyAcknowledgementVMsg( (PushProxyAcknowledgementVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof HopsFlowVMsg )
        {
            handleHopsFlowVMsg( (HopsFlowVMsg)vendorMsg, sourceHost );
        }
        else if ( vendorMsg instanceof CapabilitiesVMsg )
        {
            handleCapabilitiesVMsg( (CapabilitiesVMsg)vendorMsg, sourceHost );
        }
    }
    
    private void handleMessagesSupportedVMsg(MessagesSupportedVMsg msg, Host sourceHost )
    {
        sourceHost.setSupportedVMsgs( msg );
        
        // if push proxy is supported request it..
        boolean isFirewalled = networkMgr.hasConnectedIncoming();
        // if we are a leave or are firewalled and connected to a ultrapeer
        // and the connection supports push proxy.
        if ( ( sourceHost.isLeafUltrapeerConnection() ||
             ( isFirewalled && sourceHost.isUltrapeer() ) ) 
          && sourceHost.isPushProxySupported() )
        {
            PushProxyRequestVMsg pprmsg = new PushProxyRequestVMsg();
            // TODO2 remove this once Limewire support PPR v2
            if ( sourceHost.getVendor() != null &&
                    sourceHost.getVendor().indexOf( "LimeWire" ) != -1 )
            {
                pprmsg.setVersion( 1 );
            }
            sourceHost.queueMessageToSend( pprmsg );
        }
        if ( !networkMgr.hasConnectedIncoming() && 
              messageMgr.isTCPRedirectAllowed() &&
              sourceHost.isTCPConnectBackSupported()  )
        {
            DestAddress localAddress = networkMgr.getLocalAddress();
            VendorMsg tcpConnectBack = new TCPConnectBackVMsg( localAddress.getPort() );
            sourceHost.queueMessageToSend( tcpConnectBack );
            messageMgr.incNumberOfTCPRedirectsSent();
        }
    }
    
    private void handleCapabilitiesVMsg(CapabilitiesVMsg msg, Host sourceHost )
    {
        sourceHost.setCapabilitiesVMsgs( msg );
    }

    /**
     * @param msg
     */
    private void handleTCPConnectBackVMsg(TCPConnectBackVMsg msg, Host sourceHost )
    {
        int port = msg.getPort();
        DestAddress address = sourceHost.getHostAddress();
        if ( address.getPort() != port )
        {
            address = new DefaultDestAddress( address.getHostName(), port );
        }
        VendorMsg redirectMsg = new TCPConnectBackRedirectVMsg( address );
        
        Host[] hosts = hostMgr.getNetworkHostsContainer().getUltrapeerConnections();
        int sentCount = 0;
        for ( int i = 0; sentCount <= 5 && i < hosts.length; i++ )
        {
            if ( sourceHost == hosts[i] )
            {
                // skip sending redirect to my host.
                continue;
            }
            if ( hosts[i].isTCPConnectBackRedirectSupported() )
            {
                hosts[i].queueMessageToSend( redirectMsg );
                sentCount ++;
            }
        }
    }
    
    private void handleTCPConnectBackRedirectVMsg( TCPConnectBackRedirectVMsg msg, Host sourceHost )
    {
        final DestAddress address = msg.getAddress();
        Runnable connectBackRunner = new Runnable()
        {
            public void run()
            {
                Connection connection = null;
                try
                {
                    DestAddress connectBackAddress = new DefaultDestAddress( address.getHostName(),
                        address.getPort() );
                    connection = ConnectionFactory.createConnection(
                        connectBackAddress, 2000 );
                    connection.write( "\n\n".getBytes( ) );
                    connection.flush();
                }
                catch ( IOException exp )
                { // failed.. dont care..
                }
                catch ( Exception exp )
                {
                    NLogger.error( MessageDispatcher.class, exp, exp);
                }
                finally
                {
                    if (connection != null)
                    {
                        connection.disconnect();
                    }
                }
            }
        };
        ThreadPool.getInstance().addJob( connectBackRunner, "TCPConnectBackJob");
    }
    
    private void handlePushProxyRequestVMsg( PushProxyRequestVMsg pprvmsg, Host sourceHost )
    {
        if ( !sourceHost.isUltrapeerLeafConnection() ) 
        {
            return;
        }
        DestAddress localAddress = networkMgr.getLocalAddress();
        // PP only works if we have a valid IP to use in the PPAck message.
        if( localAddress.getIpAddress() == null )
        {
            NLogger.warn( MessageDispatcher.class, 
                "Local address has no IP to use for PPAck." );
            return;
        }
        GUID requestGUID = pprvmsg.getHeader().getMsgID();        
        PushProxyAcknowledgementVMsg ppavmsg = 
            new PushProxyAcknowledgementVMsg( localAddress,
            requestGUID );
        sourceHost.queueMessageToSend( ppavmsg );
        
        messageMgr.addToPushRoutingTable( requestGUID,
                sourceHost );            
    }
    
    private void handlePushProxyAcknowledgementVMsg( PushProxyAcknowledgementVMsg ppavmsg, Host sourceHost )
    {
        // the candidate is able to be a push proxy if the ack contains my guid.
        if ( networkMgr.getServentGuid().equals( ppavmsg.getHeader().getMsgID() ) )
        {
            sourceHost.setPushProxyAddress( ppavmsg.getHostAddress() );
        }
    }
    
    private void handleHopsFlowVMsg( HopsFlowVMsg hopsFlowVMsg, Host sourceHost )
    {
        byte hopsFlowValue = hopsFlowVMsg.getHopsValue();
        sourceHost.setHopsFlowLimit(hopsFlowValue);
    }
    
    public void handlePushRequest( PushRequestMsg msg, Host sourceHost )
    {
        // count push statistic
        MessageCountStatistic.pushMsgInCounter.increment( 1 );

        // logging a msg can be very expensive! the toString() calls are bad
        //mRemoteHost.log( Logger.FINEST, "Received Msg: " + msg);

        AccessType access = securityManager.controlHostAddressAccess(
            msg.getRequestAddress() );
        if ( access == AccessType.ACCESS_STRONGLY_DENIED )
        {
            // drop message
            dropMessage( msg, "IP access strongly denied.", sourceHost );
            return;
        }

        if ( networkMgr.getServentGuid().equals(msg.getClientGUID() ) )
        {
            if ( access == AccessType.ACCESS_GRANTED )
            {
                new PushWorker(msg);
            }
            return;
        }

        Host returnHost = messageMgr.getPushRouting(msg.getClientGUID());
        if (returnHost == null)
        {
//          mRemoteHost.log("Don't route the PushRequest since I didn't forward the QueryResponse msg.  " + msg);
            return;
        }

        // Ok, I did forward the QueryResponse msg on behalf of returnHost.
        // The PushRequest is for the returnHost.  Better route it back.
        if ( msg.getHeader().getTTL() > 0 )
        {
            returnHost.queueMessageToSend( msg );
        }
    }

    private void dropMessage( Message msg, String reason, Host sourceHost )
    {
        NLogger.info( MessageDispatcher.class, 
            "Dropping message: " + reason + " from: " + sourceHost );
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
        {
            NLogger.debug( MessageDispatcher.class,
                "Header: [" + msg.getHeader().toString() + "] - Message: [" +
                msg.toDebugString() + "].");
        }
        sourceHost.incDropCount();
        MessageCountStatistic.dropedMsgInCounter.increment( 1 );
    }
    
    public void dropMessage( MsgHeader header, byte[] body, String reason, Host sourceHost )
    {
        NLogger.info( MessageDispatcher.class, 
            "Dropping message: " + reason + " from: " + sourceHost );
        if ( NLogger.isDebugEnabled( MessageDispatcher.class ) )
        {
            NLogger.debug( MessageDispatcher.class,
                "Header: [" + header.toString() + "] - Body: [" +
                HexConverter.toHexString( body, 0, header.getDataLength() ) + "]." );
        }
        sourceHost.incDropCount();
        MessageCountStatistic.dropedMsgInCounter.increment( 1 );
    }
}
