<commit_before>package com.ardverk.dht.io;

import java.io.Closeable;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;

import com.ardverk.dht.io.transport.Transport;
import com.ardverk.dht.io.transport.TransportListener;
import com.ardverk.dht.message.Message;
import com.ardverk.dht.message.MessageCodec;
import com.ardverk.dht.message.MessageFactory;
import com.ardverk.dht.message.MessageId;
import com.ardverk.dht.message.RequestMessage;
import com.ardverk.dht.message.ResponseMessage;
import com.ardverk.logging.LoggerUtils;
import com.ardverk.utils.ExecutorUtils;

/**
 * 
 */
public abstract class MessageDispatcher implements Closeable {
    
    private static final Logger LOG 
        = LoggerUtils.getLogger(MessageDispatcher.class);
    
    private final TransportListener listener 
            = new TransportListener() {
        @Override
        public void received(SocketAddress src, 
                byte[] message) throws IOException {
            MessageDispatcher.this.handleMessage(src, message);
        }
    };
    
    private final MessageEntityManager entityManager 
        = new MessageEntityManager();
    
    private final Transport transport;
    
    private final MessageFactory factory;
    
    private final MessageCodec codec;
    
    /**
     * 
     */
    public MessageDispatcher(Transport transport, 
            MessageFactory factory, MessageCodec codec) {
        
        if (transport == null) {
            throw new NullPointerException("transport");
        }
        
        if (factory == null) {
            throw new NullPointerException("factory");
        }
        
        if (codec == null) {
            throw new NullPointerException("codec");
        }
        
        this.transport = transport;
        this.factory = factory;
        this.codec = codec;
        
        transport.addTransportListener(listener);
    }
    
    /**
     * 
     */
    public Transport getTransport() {
        return transport;
    }
    
    /**
     * 
     */
    public MessageFactory getMessageFactory() {
        return factory;
    }
    
    /**
     * 
     */
    public MessageCodec getMessageCodec() {
        return codec;
    }
    
    @Override
    public void close() {
        transport.removeTransportListener(listener);
        entityManager.close();
    }
    
    /**
     * 
     */
    public void send(ResponseMessage message) throws IOException {
        byte[] data = codec.encode(message);
        transport.send(message.getAddress(), data);
    }
    
    /**
     * 
     */
    public void send(MessageCallback callback, 
            RequestMessage message, long timeout, 
            TimeUnit unit) throws IOException {
        
        byte[] data = codec.encode(message);
        
        entityManager.add(callback, message, timeout, unit);
        transport.send(message.getAddress(), data);
    }
    
    /**
     * 
     */
    private void handleMessage(SocketAddress src, 
            byte[] data) throws IOException {
        Message message = codec.decode(data);
        handleMessage(src, message);
    }
    
    /**
     * 
     */
    protected void handleMessage(SocketAddress src, 
            Message message) throws IOException {
        
        if (message instanceof RequestMessage) {
            handleRequest(src, (RequestMessage)message);
        } else {
            handleResponse(src, (ResponseMessage)message);
        }
    }
    
    /**
     * 
     */
    protected abstract void handleRequest(SocketAddress src, 
            RequestMessage message) throws IOException;
    
    /**
     * 
     */
    protected void handleResponse(SocketAddress src, 
            ResponseMessage response) throws IOException {
        
        MessageId messageId = response.getMessageId();
        if (!factory.isFor(messageId, src)) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Wrong MessageId signature: " 
                        + src + ", " + messageId);
            }
            return;
        }<commit_msg>Making sure the same MssageId cannot be sent multiple times<commit_after>package com.ardverk.dht.io;

import java.io.Closeable;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;

import com.ardverk.collection.FixedSizeHashSet;
import com.ardverk.dht.io.transport.Transport;
import com.ardverk.dht.io.transport.TransportListener;
import com.ardverk.dht.message.Message;
import com.ardverk.dht.message.MessageCodec;
import com.ardverk.dht.message.MessageFactory;
import com.ardverk.dht.message.MessageId;
import com.ardverk.dht.message.RequestMessage;
import com.ardverk.dht.message.ResponseMessage;
import com.ardverk.logging.LoggerUtils;
import com.ardverk.utils.ExecutorUtils;

/**
 * 
 */
public abstract class MessageDispatcher implements Closeable {
    
    private static final Logger LOG 
        = LoggerUtils.getLogger(MessageDispatcher.class);
    
    private final TransportListener listener 
            = new TransportListener() {
        @Override
        public void received(SocketAddress src, 
                byte[] message) throws IOException {
            MessageDispatcher.this.handleMessage(src, message);
        }
    };
    
    private final Set<MessageId> history 
        = new FixedSizeHashSet<MessageId>(512);
    
    private final MessageEntityManager entityManager 
        = new MessageEntityManager();
    
    private final Transport transport;
    
    private final MessageFactory factory;
    
    private final MessageCodec codec;
    
    /**
     * 
     */
    public MessageDispatcher(Transport transport, 
            MessageFactory factory, MessageCodec codec) {
        
        if (transport == null) {
            throw new NullPointerException("transport");
        }
        
        if (factory == null) {
            throw new NullPointerException("factory");
        }
        
        if (codec == null) {
            throw new NullPointerException("codec");
        }
        
        this.transport = transport;
        this.factory = factory;
        this.codec = codec;
        
        transport.addTransportListener(listener);
    }
    
    /**
     * 
     */
    public Transport getTransport() {
        return transport;
    }
    
    /**
     * 
     */
    public MessageFactory getMessageFactory() {
        return factory;
    }
    
    /**
     * 
     */
    public MessageCodec getMessageCodec() {
        return codec;
    }
    
    @Override
    public void close() {
        transport.removeTransportListener(listener);
        entityManager.close();
    }
    
    /**
     * 
     */
    public void send(ResponseMessage message) throws IOException {
        byte[] data = codec.encode(message);
        transport.send(message.getAddress(), data);
    }
    
    /**
     * 
     */
    public void send(MessageCallback callback, 
            RequestMessage message, long timeout, 
            TimeUnit unit) throws IOException {
        
        byte[] data = codec.encode(message);
        
        entityManager.add(callback, message, timeout, unit);
        transport.send(message.getAddress(), data);
    }
    
    /**
     * 
     */
    private void handleMessage(SocketAddress src, 
            byte[] data) throws IOException {
        Message message = codec.decode(data);
        handleMessage(src, message);
    }
    
    /**
     * 
     */
    protected void handleMessage(SocketAddress src, 
            Message message) throws IOException {
        
        if (message instanceof RequestMessage) {
            handleRequest(src, (RequestMessage)message);
        } else {
            handleResponse(src, (ResponseMessage)message);
        }
    }
    
    /**
     * 
     */
    protected abstract void handleRequest(SocketAddress src, 
            RequestMessage message) throws IOException;
    
    /**
     * 
     */
    protected void handleResponse(SocketAddress src, 
            ResponseMessage response) throws IOException {
        
        MessageId messageId = response.getMessageId();
        if (!history.add(messageId)) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Multiple respones: " 
                        + src + ", " + messageId);
            }
            return;
        }
        
        if (!factory.isFor(messageId, src)) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Wrong MessageId signature: " 
                        + src + ", " + messageId);
            }
            return;
        }
{'fraction_non_alphanumeric': 0.09342522699923422, 'fraction_numerical': 0.0005469861065528935, 'mean_word_length': 2.05752508361204, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '418428', 'n_tokens_mistral': 2417, 'n_tokens_neox': 2347, 'n_words': 503}