<commit_before>/**
 * 
 */
package de.unikoblenz.west.koldfish.dam.impl;

import java.util.LinkedList;
import java.util.List;

import javax.jms.JMSException;

import org.apache.jena.iri.IRI;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import de.unikoblenz.west.koldfish.dam.DataAccessModule;
import de.unikoblenz.west.koldfish.dam.DataAccessModuleException;
import de.unikoblenz.west.koldfish.dam.DataAccessModuleListener;
import de.unikoblenz.west.koldfish.dam.DerefEncodedMessage;
import de.unikoblenz.west.koldfish.dam.DerefMessage;
import de.unikoblenz.west.koldfish.dam.DerefResponse;
import de.unikoblenz.west.koldfish.fluid.ConnectionManager;

/**
 * DataAccessModule implementation for JMS
 * 
 * @author lkastler
 */
public class JmsDataAccessModule implements DataAccessModule {

  private static final Logger log = LogManager.getLogger(JmsDataAccessModule.class);

  private final List<DataAccessModuleListener> listeners =
      new LinkedList<DataAccessModuleListener>();

  /**
   * creates a new JmsDataAccessModule object with the given DataAccessModuleListener
   * 
   * @throws Exception triggered if connection to JMS could not be created.
   */
  public JmsDataAccessModule() throws Exception {
    ConnectionManager.get().createTopic("dam.data", msg -> {
      if (msg instanceof DerefResponse) {
        DerefResponse derefResp = (DerefResponse) msg;
        log.debug("message: {}", derefResp);

        for (DataAccessModuleListener listener : listeners) {
          listener.onDerefResponse(derefResp);
        }

      }
    }).createTopic("dam.errors", msg -> {
      log.warn(msg);
    });
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#start()
   */
  @Override
  public void start() throws Exception {
    ConnectionManager.get().start();
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#isStarted()
   */
  @Override
  public boolean isStarted() {
    return ConnectionManager.get().isConnectionActive();
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.DataAccessModule#deref(org.apache.jena.iri.IRI)
   */
  @Override
  public void deref(IRI iri) throws DataAccessModuleException {
    try {
      ConnectionManager.get().sendToQueue("dam.deref", new DerefMessage(iri.toString()));
    } catch (JMSException e) {
      throw new DataAccessModuleException(e);
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.DataAccessModule#deref(long)
   */
  @Override
  public void deref(long compressedIri) throws DataAccessModuleException {
    try {
      ConnectionManager.get().sendToQueue("dam.deref", new DerefEncodedMessage(compressedIri));
    } catch (JMSException e) {
      throw new DataAccessModuleException(e);
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#stop()
   */
  @Override
  public void stop() throws Exception {
    ConnectionManager.get().close();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * de.unikoblenz.west.koldfish.dam.DataAccessModule#addListener(de.unikoblenz.west.koldfish.dam
   * .DataAccessModuleListener)
   */
  @Override
  public void addListener(DataAccessModuleListener listener) {
    throw new UnsupportedOperationException("implement addListener");
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * de.unikoblenz.west.koldfish.dam.DataAccessModule#removeListener(de.unikoblenz.west.koldfish
   * .dam.DataAccessModuleListener)
   */
  @Override
  public void removeListener(DataAccessModuleListener listener) {
    throw new UnsupportedOperationException("implement removeListener");
  }
}<commit_msg>implemented add and remove listener<commit_after>/**
 * 
 */
package de.unikoblenz.west.koldfish.dam.impl;

import java.util.LinkedList;
import java.util.List;

import javax.jms.JMSException;

import org.apache.jena.iri.IRI;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import de.unikoblenz.west.koldfish.dam.DataAccessModule;
import de.unikoblenz.west.koldfish.dam.DataAccessModuleException;
import de.unikoblenz.west.koldfish.dam.DataAccessModuleListener;
import de.unikoblenz.west.koldfish.dam.DerefEncodedMessage;
import de.unikoblenz.west.koldfish.dam.DerefMessage;
import de.unikoblenz.west.koldfish.dam.DerefResponse;
import de.unikoblenz.west.koldfish.fluid.ConnectionManager;

/**
 * DataAccessModule implementation for JMS
 * 
 * @author lkastler
 */
public class JmsDataAccessModule implements DataAccessModule {

  private static final Logger log = LogManager.getLogger(JmsDataAccessModule.class);

  private final List<DataAccessModuleListener> listeners =
      new LinkedList<DataAccessModuleListener>();

  /**
   * creates a new JmsDataAccessModule object with the given DataAccessModuleListener
   * 
   * @throws Exception triggered if connection to JMS could not be created.
   */
  public JmsDataAccessModule() throws Exception {
    ConnectionManager.get().createTopic("dam.data", msg -> {
      if (msg instanceof DerefResponse) {
        DerefResponse derefResp = (DerefResponse) msg;
        log.debug("message: {}", derefResp);

        for (DataAccessModuleListener listener : listeners) {
          listener.onDerefResponse(derefResp);
        }

      }
    }).createTopic("dam.errors", msg -> {
      log.warn(msg);
    });
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#start()
   */
  @Override
  public void start() throws Exception {
    ConnectionManager.get().start();
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#isStarted()
   */
  @Override
  public boolean isStarted() {
    return ConnectionManager.get().isConnectionActive();
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.DataAccessModule#deref(org.apache.jena.iri.IRI)
   */
  @Override
  public void deref(IRI iri) throws DataAccessModuleException {
    try {
      ConnectionManager.get().sendToQueue("dam.deref", new DerefMessage(iri.toString()));
    } catch (JMSException e) {
      throw new DataAccessModuleException(e);
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.DataAccessModule#deref(long)
   */
  @Override
  public void deref(long compressedIri) throws DataAccessModuleException {
    try {
      ConnectionManager.get().sendToQueue("dam.deref", new DerefEncodedMessage(compressedIri));
    } catch (JMSException e) {
      throw new DataAccessModuleException(e);
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see de.unikoblenz.west.koldfish.dam.LifeCycle#stop()
   */
  @Override
  public void stop() throws Exception {
    ConnectionManager.get().close();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * de.unikoblenz.west.koldfish.dam.DataAccessModule#addListener(de.unikoblenz.west.koldfish.dam
   * .DataAccessModuleListener)
   */
  @Override
  public void addListener(DataAccessModuleListener listener) {
    listeners.add(listener);
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * de.unikoblenz.west.koldfish.dam.DataAccessModule#removeListener(de.unikoblenz.west.koldfish
   * .dam.DataAccessModuleListener)
   */
  @Override
  public void removeListener(DataAccessModuleListener listener) {
    listeners.remove(listener);
  }
}
{'fraction_non_alphanumeric': 0.11985834922364479, 'fraction_numerical': 0.0005448106782892945, 'mean_word_length': 3.6211453744493394, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, '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': '22718', 'n_tokens_mistral': 2544, 'n_tokens_neox': 2452, 'n_words': 380}