<commit_before>package org.safehaus.kiskis.mgmt.server.command;

import com.google.common.base.Strings;
import org.safehaus.kiskis.mgmt.shared.protocol.*;
import org.safehaus.kiskis.mgmt.shared.protocol.api.CommandManagerInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.CommandTransportInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.PersistenceInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.ResponseListener;
import org.safehaus.kiskis.mgmt.shared.protocol.api.ui.CommandListener;
import org.safehaus.kiskis.mgmt.shared.protocol.enums.ResponseType;
import org.safehaus.kiskis.mgmt.shared.protocol.enums.TaskStatus;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Created with IntelliJ IDEA. User: daralbaev Date: 11/7/13 Time: 11:16 PM
 */
public class CommandManager implements CommandManagerInterface, ResponseListener {

    private static final Logger LOG = Logger.getLogger(CommandManager.class.getName());
    private PersistenceInterface persistenceCommand;
    private CommandTransportInterface communicationService;
    private final Map<CommandListener, ReentrantLock> listeners = new ConcurrentHashMap<CommandListener, ReentrantLock>();
    private ExecutorService notifierExecService;
//    private CommandNotifier commandNotifier;

    @Override
    public void executeCommand(Command command) {
        try {
            if (persistenceCommand.saveCommand(command)) {
                communicationService.sendCommand(command);
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in executeCommand", ex);
        }
    }

    @Override
    public void onResponse(Response response) {
        switch (response.getType()) {
            case EXECUTE_TIMEOUTED:
            case EXECUTE_RESPONSE: {
                persistenceCommand.saveResponse(response);
//                commandNotifier.addResponse(response);
                notify(response);
                break;
            }
            case EXECUTE_RESPONSE_DONE: {
                persistenceCommand.saveResponse(response);
//                commandNotifier.addResponse(response);
                notify(response);
                break;
            }
            default: {
                break;
            }
        }
    }

    @Override
    public void addListener(CommandListener listener) {
        try {
            LOG.log(Level.INFO, "Adding module listener : {0}", listener.getName());
            listeners.put(listener, new ReentrantLock(true));
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in addListener", ex);
        }
    }

    @Override
    public void removeListener(CommandListener listener) {
        try {
            LOG.log(Level.INFO, "Removing module listener : {0}", listener.getName());
            listeners.remove(listener);
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in removeListener", ex);
        }
    }

    public void init() {
        try {
            if (communicationService != null) {
                notifierExecService = Executors.newFixedThreadPool(100);
//                exec = Executors.newSingleThreadExecutor();
//                commandNotifier = new CommandNotifier(listeners);
//                exec.execute(commandNotifier);
                communicationService.addListener(this);
            } else {
                throw new Exception("Missing communication service");
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in init", ex);
        }
    }

    private void notify(final Response response) {
        try {
            for (final Iterator<Entry<CommandListener, ReentrantLock>> it = listeners.entrySet().iterator(); it.hasNext();) {
                Entry<CommandListener, ReentrantLock> listenerEntry = it.next();
                if (listenerEntry != null && listenerEntry.getKey() != null
                        && listenerEntry.getValue() != null && listenerEntry.getKey().getName() != null) {
                    final CommandListener listener = listenerEntry.getKey();
                    final ReentrantLock lock = listenerEntry.getValue();
                    if (response != null && response.getSource() != null
                            && listener.getName().equalsIgnoreCase(response.getSource())) {
                        notifierExecService.execute(new ResponseNotifier(lock, listener, response));
                    }
                } else {
                    it.remove();
                }
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in notify", ex);
        }
    }

    public void destroy() {
        try {
            notifierExecService.shutdown();
            if (communicationService != null) {
                communicationService.removeListener(this);
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in destroy", ex);
        }
    }

    public void setPersistenceCommandService(PersistenceInterface persistenceCommand) {
        this.persistenceCommand = persistenceCommand;
    }

    public void setCommunicationService(CommandTransportInterface communicationService) {
        this.communicationService = communicationService;
    }
<commit_msg>Mongo in progress<commit_after>package org.safehaus.kiskis.mgmt.server.command;

import com.google.common.base.Strings;
import org.safehaus.kiskis.mgmt.shared.protocol.*;
import org.safehaus.kiskis.mgmt.shared.protocol.api.CommandManagerInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.CommandTransportInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.PersistenceInterface;
import org.safehaus.kiskis.mgmt.shared.protocol.api.ResponseListener;
import org.safehaus.kiskis.mgmt.shared.protocol.api.ui.CommandListener;
import org.safehaus.kiskis.mgmt.shared.protocol.enums.ResponseType;
import org.safehaus.kiskis.mgmt.shared.protocol.enums.TaskStatus;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Created with IntelliJ IDEA. User: daralbaev Date: 11/7/13 Time: 11:16 PM
 */
public class CommandManager implements CommandManagerInterface, ResponseListener {

    private static final Logger LOG = Logger.getLogger(CommandManager.class.getName());
    private PersistenceInterface persistenceCommand;
    private CommandTransportInterface communicationService;
    private final Map<CommandListener, ReentrantLock> listeners = new ConcurrentHashMap<CommandListener, ReentrantLock>();
    private ExecutorService notifierExecService;
//    private CommandNotifier commandNotifier;

    @Override
    public void executeCommand(Command command) {
        try {
            if (persistenceCommand.saveCommand(command)) {
                communicationService.sendCommand(command);
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in executeCommand", ex);
        }
    }

    @Override
    public void onResponse(Response response) {
        switch (response.getType()) {
            case EXECUTE_TIMEOUTED:
            case EXECUTE_RESPONSE: {
                persistenceCommand.saveResponse(response);
//                commandNotifier.addResponse(response);
                notify(response);
                break;
            }
            case EXECUTE_RESPONSE_DONE: {
                persistenceCommand.saveResponse(response);
//                commandNotifier.addResponse(response);
                notify(response);
                break;
            }
            default: {
                break;
            }
        }
    }

    @Override
    public void addListener(CommandListener listener) {
        try {
            LOG.log(Level.INFO, "Adding module listener : {0}", listener.getName());
            listeners.put(listener, new ReentrantLock(true));
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in addListener", ex);
        }
    }

    @Override
    public void removeListener(CommandListener listener) {
        try {
            LOG.log(Level.INFO, "Removing module listener : {0}", listener.getName());
            listeners.remove(listener);
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in removeListener", ex);
        }
    }

    public void init() {
        try {
            if (communicationService != null) {
                notifierExecService = Executors.newFixedThreadPool(100);
//                exec = Executors.newSingleThreadExecutor();
//                commandNotifier = new CommandNotifier(listeners);
//                exec.execute(commandNotifier);
                communicationService.addListener(this);
            } else {
                throw new Exception("Missing communication service");
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in init", ex);
        }
    }

    private void notify(final Response response) {
        try {
            for (final Iterator<Entry<CommandListener, ReentrantLock>> it = listeners.entrySet().iterator(); it.hasNext();) {
                Entry<CommandListener, ReentrantLock> listenerEntry = it.next();
                System.out.println("Iterating listener " + listenerEntry.getKey().getName());
                if (listenerEntry != null && listenerEntry.getKey() != null
                        && listenerEntry.getValue() != null && listenerEntry.getKey().getName() != null) {
                    final CommandListener listener = listenerEntry.getKey();
                    final ReentrantLock lock = listenerEntry.getValue();
                    System.out.println("Trying to notify listener " + listener.getName());
                    if (response != null && response.getSource() != null
                            && listener.getName().equalsIgnoreCase(response.getSource())) {
                        System.out.println("Notifying listener " + listener.getName() + " with " + response);
                        notifierExecService.execute(new ResponseNotifier(lock, listener, response));
                    }
                } else {
                    it.remove();
                }
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in notify", ex);
        }
    }

    public void destroy() {
        try {
            notifierExecService.shutdown();
            if (communicationService != null) {
                communicationService.removeListener(this);
            }
        } catch (Exception ex) {
            LOG.log(Level.SEVERE, "Error in destroy", ex);
        }
    }

    public void setPersistenceCommandService(PersistenceInterface persistenceCommand) {
        this.persistenceCommand = persistenceCommand;
    }

    public void setCommunicationService(CommandTransportInterface communicationService) {
        this.communicationService = communicationService;
    }

{'fraction_non_alphanumeric': 0.09999136516708401, 'fraction_numerical': 0.0024177532164752612, 'mean_word_length': 2.339677047289504, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '3147051', 'n_tokens_mistral': 2996, 'n_tokens_neox': 2905, 'n_words': 583}