Started working on thorough testing

pom.xml

-<project xmlns="http://maven.apache.org/POM/4.0.0"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>beamline</groupId>
@@ -57,12 +56,38 @@
 		<dependency>
 			<groupId>com.github.beamline</groupId>
 			<artifactId>graphviz</artifactId>
-			<version>0.0.2</version>
+			<version>master-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>com.opencsv</groupId>
 			<artifactId>opencsv</artifactId>
 			<version>5.6</version>
 		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>5.8.2</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.platform</groupId>
+			<artifactId>junit-platform-suite-api</artifactId>
+			<version>1.8.2</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.hamcrest</groupId>
+			<artifactId>hamcrest</artifactId>
+			<version>2.2</version>
+		</dependency>
 	</dependencies>
+	
+	<build>
+		<plugins>
+			<plugin>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>2.22.2</version>
+			</plugin>
+		</plugins>
+	</build>
 </project>

src/main/java/beamline/exceptions/EventException.java

+package beamline.exceptions;
+
+public class EventException extends Exception {
+
+	private static final long serialVersionUID = 5835305478001040595L;
+	
+	public EventException(String message) {
+		super(message);
+	}
+}

src/main/java/beamline/filters/ExcludeOnCaseAttributeEqualityFilter.java

@@ -37,15 +37,6 @@ public class ExcludeOnCaseAttributeEqualityFilter<T extends XAttribute> implemen
 		this.attributeValues = new HashSet<T>(Arrays.asList(values));
 	}
 	
-	/**
-	 * Adds the value to the list of values to be considered for removal
-	 * 
-	 * @param value value
-	 */
-	public void addValue(T value) {
-		this.attributeValues.add(value);
-	}
-	
 	@Override
 	public boolean test(@NonNull XTrace t) throws Throwable {
 		return !attributeValues.contains(t.getAttributes().get(attributeName));

src/main/java/beamline/filters/RetainOnCaseAttributeEqualityFilter.java

@@ -37,15 +37,6 @@ public class RetainOnCaseAttributeEqualityFilter<T extends XAttribute> implement
 		this.attributeValues = new HashSet<T>(Arrays.asList(values));
 	}
 	
-	/**
-	 * Adds the value to the list of values to be considered for retention
-	 * 
-	 * @param value value
-	 */
-	public void addValue(T value) {
-		this.attributeValues.add(value);
-	}
-	
 	@Override
 	public boolean test(@NonNull XTrace t) throws Throwable {
 		return attributeValues.contains(t.getAttributes().get(attributeName));

src/main/java/beamline/sources/CSVLogSource.java

@@ -3,17 +3,17 @@ package beamline.sources;
 import java.io.Reader;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.LinkedList;
+import java.util.List;
 
-import org.deckfour.xes.extension.std.XConceptExtension;
-import org.deckfour.xes.factory.XFactory;
-import org.deckfour.xes.factory.XFactoryNaiveImpl;
-import org.deckfour.xes.model.XEvent;
+import org.apache.commons.lang3.tuple.Pair;
 import org.deckfour.xes.model.XTrace;
 
 import com.opencsv.CSVParser;
 import com.opencsv.CSVReader;
 import com.opencsv.CSVReaderBuilder;
 
+import beamline.utils.EventUtils;
 import io.reactivex.rxjava3.annotations.NonNull;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.core.ObservableEmitter;
@@ -27,7 +27,6 @@ import io.reactivex.rxjava3.core.ObservableOnSubscribe;
  */
 public class CSVLogSource implements XesSource {
 
-	private static XFactory xesFactory = new XFactoryNaiveImpl();
 	private CSVReader csvReader;
 	private String filename;
 	private int caseIdColumn;
@@ -80,16 +79,11 @@ public class CSVLogSource implements XesSource {
 			public void subscribe(@NonNull ObservableEmitter<@NonNull XTrace> emitter) throws Throwable {
 				String[] line;
 				while ((line = csvReader.readNext()) != null) {
-					XTrace eventWrapper = xesFactory.createTrace();
-					XEvent newEvent = xesFactory.createEvent();
-					XConceptExtension.instance().assignName(eventWrapper, line[caseIdColumn]);
-					XConceptExtension.instance().assignName(newEvent, line[activityNameColumn]);
+					List<Pair<String, String>> attributes = new LinkedList<Pair<String, String>>();
 					for (int i = 0; i < line.length; i++) {
-						String attributeName = "attribute_" + i;
-						newEvent.getAttributes().put(attributeName, xesFactory.createAttributeLiteral(attributeName, line[i], null));
+						attributes.add(Pair.of("attribute_" + i, line[i]));
 					}
-					eventWrapper.add(newEvent);
-					emitter.onNext(eventWrapper);
+					emitter.onNext(EventUtils.create(line[activityNameColumn], line[caseIdColumn], null, attributes));
 				}
 				emitter.onComplete();
 			}

src/main/java/beamline/sources/MQTTXesSource.java

 package beamline.sources;
 
-import java.util.Date;
 import java.util.UUID;
 
-import org.deckfour.xes.extension.std.XConceptExtension;
-import org.deckfour.xes.extension.std.XTimeExtension;
-import org.deckfour.xes.factory.XFactory;
-import org.deckfour.xes.factory.XFactoryNaiveImpl;
-import org.deckfour.xes.model.XEvent;
 import org.deckfour.xes.model.XTrace;
 import org.eclipse.paho.client.mqttv3.IMqttClient;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
@@ -16,6 +10,7 @@ import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
+import beamline.utils.EventUtils;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.subjects.PublishSubject;
 
@@ -37,7 +32,6 @@ import io.reactivex.rxjava3.subjects.PublishSubject;
  */
 public class MQTTXesSource implements XesSource {
 
-	private static XFactory xesFactory = new XFactoryNaiveImpl();
 	private String processName;
 	private String brokerHost;
 	private String topicBase;
@@ -77,14 +71,7 @@ public class MQTTXesSource implements XesSource {
 				String partBeforeActName = topic.substring(0, posLastSlash);
 				String activityName = topic.substring(posLastSlash + 1);
 				String caseId = partBeforeActName.substring(partBeforeActName.lastIndexOf("/") + 1);
-
-				XEvent event = xesFactory.createEvent();
-				XConceptExtension.instance().assignName(event, activityName);
-				XTimeExtension.instance().assignTimestamp(event, new Date());
-				XTrace eventWrapper = xesFactory.createTrace();
-				XConceptExtension.instance().assignName(eventWrapper, caseId);
-				eventWrapper.add(event);
-				ps.onNext(eventWrapper);
+				ps.onNext(EventUtils.create(activityName, caseId));
 			}
 			
 			@Override

src/main/java/beamline/tester/Tester.java

+package beamline.tester;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.extension.std.XTimeExtension;
+import org.deckfour.xes.in.XParser;
+import org.deckfour.xes.in.XesXmlParser;
+import org.deckfour.xes.model.XEvent;
+import org.deckfour.xes.model.XLog;
+import org.deckfour.xes.model.XTrace;
+
+import beamline.filters.ExcludeActivitiesFilter;
+import beamline.filters.RetainActivitiesFilter;
+import beamline.mappers.DirectlyFollowsRelation;
+import beamline.mappers.InfiniteSizeDirectlyFollowsMapper;
+import beamline.models.algorithms.HookEventProcessing;
+import beamline.models.algorithms.StreamMiningAlgorithm;
+import beamline.models.responses.GraphvizResponse;
+import beamline.sources.MQTTXesSource;
+import beamline.sources.XesLogSource;
+import beamline.sources.XesSource;
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableEmitter;
+import io.reactivex.rxjava3.core.ObservableOnSubscribe;
+import io.reactivex.rxjava3.core.ObservableSource;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.functions.Predicate;
+import jdk.jshell.execution.StreamingExecutionControl;
+
+public class Tester {
+
+	public static void main(String[] args) throws Exception {
+		System.out.println("start");
+		
+		XParser p = new XesXmlParser();
+		XLog l = p.parse(new File("C:\\Users\\andbur\\Desktop\\input.xes")).get(0);
+//		XesSource source = new XesLogSource(l);
+		XesSource source = new MQTTXesSource("tcp://broker.hivemq.com:1883", "pmcep", "test");
+		source.prepare();
+		
+//		DiscoveryMiner miner = new DiscoveryMiner();
+//		miner.setMinDependency(0.3);
+//		miner.setModelRefreshRate(1);
+//		miner.setOnAfterEvent(new HookEventProcessing() {
+//			@Override
+//			public void trigger() {
+//				if (miner.getProcessedEvents() % 100 == 0) {
+//					try {
+//						GraphvizResponse resp = miner.getLatestResponse();
+//						resp.generateDot().exportToSvg(new File("C:\\Users\\andbur\\Desktop\\output-" + miner.getProcessedEvents() + ".svg"));
+//					} catch (IOException e) { }
+//				}
+//			}
+//		});
+		
+		Observable<XTrace> obs = source.getObservable();
+		obs
+//		.filter(new RetainActivitiesFilter("A", "B", "C", "dummy-retain"))
+//		.filter(new ExcludeActivitiesFilter("A", "dummy-exclude"))
+//		.map(new DirectSuccessionMapper())
+//		.combine(new SlidingWindow(1000))
+//		.map(new Miner(1, 0.5))
+//		.subscribe(miner);
+//		.mapOptional(new Function<XTrace, Optional<Pair<String, String>>>() {
+//
+//			@Override
+//			public @NonNull Optional<Pair<String, String>> apply(@NonNull XTrace t) throws Throwable {
+//				// TODO Auto-generated method stub
+//				return null;
+//			}
+//		});
+//		.map(new Function<XTrace, DirectlyFollowRelation>() {
+//			Map<String, String> map = new HashMap<String, String>();
+//			
+//			@Override
+//			public @NonNull DirectlyFollowRelation apply(@NonNull XTrace t) throws Throwable {
+//				String caseId = XConceptExtension.instance().extractName(t);
+//				String act = XConceptExtension.instance().extractName(t.get(0));
+//				DirectlyFollowRelation toRet = new DirectlyFollowRelation();
+//				if (map.containsKey(caseId)) {
+//					String prevAct = map.get(caseId);
+//					toRet.first = prevAct;
+//					toRet.second = act;
+//					toRet.caseId = caseId;
+//				}
+//				map.put(caseId, act);
+//				
+//				return toRet;
+//			}
+//		})
+//		.filter(new Predicate<DirectlyFollowRelation>() {
+//			@Override
+//			public boolean test(@NonNull DirectlyFollowRelation t) throws Throwable {
+//				return t.first != null && t.second != null;
+//			}
+//		})
+		.flatMap(new InfiniteSizeDirectlyFollowsMapper())
+		.subscribe(new Consumer<DirectlyFollowsRelation>() {
+
+			@Override
+			public void accept(@NonNull DirectlyFollowsRelation t) throws Throwable {
+				System.out.println(
+						XConceptExtension.instance().extractName(t.getFirst()) +  " -> " +
+						XConceptExtension.instance().extractName(t.getSecond()) + " for case " + t.getCaseId());
+			}
+		});
+		
+		
+		
+//		.subscribe(new Consumer<XTrace>() {
+//			@Override
+//			public void accept(@NonNull XTrace t) throws Throwable {
+//				System.out.println(
+//					XConceptExtension.instance().extractName(t) + " - " +
+//					XConceptExtension.instance().extractName(t.get(0)) +  " - " +
+//					XTimeExtension.instance().extractTimestamp(t.get(0))
+//				);
+//			}
+//		});
+		
+//		miner.getLatestResponse().generateDot().exportToSvg(new File("C:\\Users\\andbur\\Desktop\\output.svg"));
+
+		System.out.println("done");
+	}
+
+}

src/main/java/beamline/tester/Tester2.java

+package beamline.tester;
+
+import java.io.IOException;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.model.XTrace;
+
+import beamline.sources.CSVLogSource;
+import beamline.sources.XesSource;
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.functions.Consumer;
+
+public class Tester2 {
+
+	public static void main(String[] args) throws Exception {
+		String f = "C:\\Users\\andbur\\OneDrive - Danmarks Tekniske Universitet\\uni\\publishing\\papers\\conferences\\2022-caise\\material\\data\\synthetic\\event streams\\sudden_time_noise0_500_pm_simple.csv";
+		XesSource source = new CSVLogSource(f, 1, 2);
+		source.prepare();
+		
+		Observable<XTrace> obs = source.getObservable();
+		obs.subscribe(new Consumer<XTrace>() {
+			@Override
+			public void accept(@NonNull XTrace t) throws Throwable {
+				System.out.println(
+					XConceptExtension.instance().extractName(t) + " - " +
+					XConceptExtension.instance().extractName(t.get(0)) +  " - " +
+							t.get(0).getAttributes()
+				);
+			}
+		});
+		
+		System.out.println("done");
+	}
+}

src/main/java/beamline/tester/Tester3.java

+package beamline.tester;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.factory.XFactory;
+import org.deckfour.xes.factory.XFactoryNaiveImpl;
+import org.deckfour.xes.model.XEvent;
+import org.deckfour.xes.model.XTrace;
+
+import beamline.mappers.DirectlyFollowsRelation;
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableSource;
+import io.reactivex.rxjava3.core.Observer;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.functions.Function;
+
+public class Tester3 {
+
+	public static void main(String[] args) {
+		
+		String fileName = "C:\\Users\\andbur\\Desktop\\log.txt";
+		XFactory factory = new XFactoryNaiveImpl();
+		
+		Observable<String> stramOfStrings = Observable.defer(() -> new ObservableSource<String>() {
+			@Override
+			public void subscribe(@NonNull Observer<? super @NonNull String> observer) {
+				try {
+					Files.lines(Paths.get(fileName)).forEach(observer::onNext);
+					observer.onComplete();
+				} catch (IOException e) {
+					observer.onError(e);
+				}
+			}
+		});
+		
+		
+		Observable<XTrace> streamOfXTraces = stramOfStrings.flatMap(new Function<String, ObservableSource<XTrace>>() {
+			@Override
+			public @NonNull ObservableSource<XTrace> apply(@NonNull String t) throws Throwable {
+				String caseId = t.substring(0, 3);
+				String activityName = t.substring(3);
+				
+				XTrace wrapper = factory.createTrace();
+				XEvent event = factory.createEvent();
+				
+				XConceptExtension.instance().assignName(wrapper, caseId);
+				XConceptExtension.instance().assignName(event, activityName);
+				
+				wrapper.add(event);
+				
+				return Observable.just(wrapper);
+			}
+		});
+		
+		
+		streamOfXTraces.subscribe(new Consumer<XTrace>() {
+			@Override
+			public void accept(@NonNull XTrace t) throws Throwable {
+				System.out.println(
+					XConceptExtension.instance().extractName(t) + " - " +
+					XConceptExtension.instance().extractName(t.get(0)) +  " - " +
+							t.get(0).getAttributes()
+				);
+			}
+		});
+	}
+}

src/main/java/beamline/utils/EventUtils.java

+package beamline.utils;
+
+import java.util.Collection;
+import java.util.Date;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.extension.std.XTimeExtension;
+import org.deckfour.xes.factory.XFactory;
+import org.deckfour.xes.factory.XFactoryNaiveImpl;
+import org.deckfour.xes.model.XEvent;
+import org.deckfour.xes.model.XTrace;
+import org.deckfour.xes.model.impl.XAttributeLiteralImpl;
+
+import beamline.exceptions.EventException;
+
+/**
+ * This class contains some utility methods useful to handle and process events.
+ * 
+ * @author Andrea Burattin
+ */
+public class EventUtils {
+
+	private static final XFactory xesFactory = new XFactoryNaiveImpl();
+	
+	/**
+	 * Creates a new {@link XTrace} referring to one event
+	 * 
+	 * @param activityName the name of the activity
+	 * @param caseId the identifier of the process instance
+	 * @param time the time when the event has happened
+	 * @param eventAttributes a collection of string attributes for the event
+	 * @return the new event
+	 * @throws EventException this exception is thrown is incomplete information
+	 * is provided
+	 */
+	public static XTrace create(String activityName, String caseId, Date time, Collection<Pair<String, String>> eventAttributes) throws EventException {
+		if (activityName == null || caseId == null) {
+			throw new EventException("Activity name or case id missing");
+		}
+		
+		XEvent event = xesFactory.createEvent();
+		XConceptExtension.instance().assignName(event, activityName);
+		if (time == null) {
+			XTimeExtension.instance().assignTimestamp(event, new Date());
+		} else {
+			XTimeExtension.instance().assignTimestamp(event, time);
+		}
+		if (eventAttributes != null) {
+			for(Pair<String, String> a : eventAttributes) {
+				event.getAttributes().put(a.getLeft(), new XAttributeLiteralImpl(a.getLeft(), a.getRight()));
+			}
+		}
+		XTrace eventWrapper = xesFactory.createTrace();
+		XConceptExtension.instance().assignName(eventWrapper, caseId);
+		eventWrapper.add(event);
+		return eventWrapper;
+	}
+	
+	/**
+	 * Creates a new {@link XTrace} referring to one event
+	 * 
+	 * @param activityName the name of the activity
+	 * @param caseId the identifier of the process instance
+	 * @param time the time when the event has happened
+	 * @return the new event
+	 * @throws EventException this exception is thrown is incomplete information
+	 * is provided
+	 */
+	public static XTrace create(String activityName, String caseId, Date time) throws EventException {
+		return create(activityName, caseId, time, null);
+	}
+	
+	/**
+	 * Creates a new {@link XTrace} referring to one event. The time of the
+	 * event is set to the current time
+	 * 
+	 * @param activityName the name of the activity
+	 * @param caseId the identifier of the process instance
+	 * @return the new event
+	 * @throws EventException this exception is thrown is incomplete information
+	 * is provided
+	 */
+	public static XTrace create(String activityName, String caseId) throws EventException {
+		return create(activityName, caseId, null, null);
+	}
+	
+	/**
+	 * Extracts the activity name
+	 * 
+	 * @param event the event
+	 * @return the activity name
+	 */
+	public static String getActivityName(XTrace event) {
+		return XConceptExtension.instance().extractName(event.get(0));
+	}
+	
+	/**
+	 * Extracts the case id
+	 * 
+	 * @param event the event
+	 * @return the case id
+	 */
+	public static String getCaseId(XTrace event) {
+		return XConceptExtension.instance().extractName(event);
+	}
+}

src/main/java/beamline/utils/package-info.java

+/**
+ * This package contains utility classes to be used throughout the framework.
+ */
+package beamline.utils;
\ No newline at end of file

src/test/java/beamline/tests/AlgorithmTests.java

+package beamline.tests;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItems;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import beamline.models.algorithms.StreamMiningAlgorithm;
+import io.reactivex.rxjava3.core.Observable;
+
+public class AlgorithmTests {
+
+	Observable<Integer> o = Observable.just(3, 7, 11, 13);
+	
+	@Test
+	public void test_result() {
+		StreamMiningAlgorithm<Integer, Integer> m = new StreamMiningAlgorithm<Integer, Integer>() {
+			public Integer product = 1;
+			
+			@Override
+			public Integer ingest(Integer event) {
+				product *= event;
+				setLatestResponse(-product);
+				return product;
+			}
+		};
+		
+		o.subscribe(m);
+		assertEquals(4, m.getProcessedEvents());
+		assertEquals(3003, m.getLatestResponse());
+	}
+	
+	@Test
+	public void test_hooks() {
+		StreamMiningAlgorithm<Integer, Integer> m = new StreamMiningAlgorithm<Integer, Integer>() {
+			public Integer product = 1;
+			
+			@Override
+			public Integer ingest(Integer event) {
+				product *= event;
+				setLatestResponse(-product);
+				return product;
+			}
+		};
+		
+		List<Integer> resultsBefore = new ArrayList<Integer>();
+		m.setOnBeforeEvent(() -> {
+			resultsBefore.add(m.getProcessedEvents());
+		});
+		
+		List<Integer> resultsAfter = new ArrayList<Integer>();
+		m.setOnAfterEvent(() -> {
+			resultsAfter.add(m.getProcessedEvents());
+		});
+		
+		o.subscribe(m);
+		
+		assertThat(resultsBefore, hasItems(0,1,2,3));
+		assertThat(resultsAfter, hasItems(1,2,3,4));
+	}
+}

src/test/java/beamline/tests/FiltersTests.java

+package beamline.tests;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItems;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.deckfour.xes.model.XTrace;
+import org.deckfour.xes.model.impl.XAttributeLiteralImpl;
+import org.junit.jupiter.api.Test;
+
+import beamline.exceptions.EventException;
+import beamline.filters.ExcludeActivitiesFilter;
+import beamline.filters.ExcludeOnCaseAttributeEqualityFilter;
+import beamline.filters.RetainActivitiesFilter;
+import beamline.filters.RetainOnCaseAttributeEqualityFilter;
+import beamline.utils.EventUtils;
+import io.reactivex.rxjava3.core.Observable;
+
+public class FiltersTests {
+
+	EventUtils e = new EventUtils();
+	
+	@Test
+	public void test_exclude_activities_on_name_filter() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new ExcludeActivitiesFilter("A"))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(3, results.size());
+		assertThat(results, hasItems("K","B","C"));
+	}
+	
+	@Test
+	public void test_retain_activities_on_name_filter() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new RetainActivitiesFilter("A","B"))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(3, results.size());
+		assertThat(results, hasItems("A","B","A"));
+	}
+	
+	@Test
+	public void test_retain_activities_on_case_attribute_filter_1() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new RetainOnCaseAttributeEqualityFilter<XAttributeLiteralImpl>(
+					"a1",
+					new XAttributeLiteralImpl("a1", "v1")))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(1, results.size());
+		assertThat(results, hasItems("A"));
+	}
+	
+	@Test
+	public void test_retain_activities_on_case_attribute_filter_2() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new RetainOnCaseAttributeEqualityFilter<XAttributeLiteralImpl>(
+					"a1",
+					new XAttributeLiteralImpl("a1", "v1"),
+					new XAttributeLiteralImpl("a1", "v4")))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(2, results.size());
+		assertThat(results, hasItems("A","C"));
+	}
+	
+	@Test
+	public void test_exclude_activities_on_case_attribute_filter_1() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new ExcludeOnCaseAttributeEqualityFilter<XAttributeLiteralImpl>(
+					"a1",
+					new XAttributeLiteralImpl("a1", "v1")))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(4, results.size());
+		assertThat(results, hasItems("K","B","A","C"));
+	}
+	
+	@Test
+	public void test_exclude_activities_on_case_attribute_filter_2() {
+		List<String> results = new ArrayList<String>();
+		generateObservableSameCaseId()
+			.filter(new ExcludeOnCaseAttributeEqualityFilter<XAttributeLiteralImpl>(
+					"a1",
+					new XAttributeLiteralImpl("a1", "v1"),
+					new XAttributeLiteralImpl("a1", "v4")))
+			.subscribe((t) -> results.add(EventUtils.getActivityName(t)));
+		assertEquals(3, results.size());
+		assertThat(results, hasItems("K","B","A"));
+	}
+	
+	/*
+	 * Generate a streams with these events:
+	 * - K
+	 * - A / trace attribute: (a1,v1)
+	 * - B
+	 * - A
+	 * - C / trace attribute: (a1,v4)
+	 */
+	private Observable<XTrace> generateObservableSameCaseId() {
+		XTrace[] events = null;
+		try {
+			events = new XTrace[] {
+				EventUtils.create("K", "c"),
+				EventUtils.create("A", "c"),
+				EventUtils.create("B", "c"),
+				EventUtils.create("A", "c"),
+				EventUtils.create("C", "c")
+			};
+		} catch (EventException e) {
+			e.printStackTrace();
+		}
+		events[1].getAttributes().put("a1", new XAttributeLiteralImpl("a1", "v1"));
+		events[2].get(0).getAttributes().put("a2", new XAttributeLiteralImpl("a2", "v3"));
+		events[3].get(0).getAttributes().put("a2", new XAttributeLiteralImpl("a2", "v2"));
+		events[4].getAttributes().put("a1", new XAttributeLiteralImpl("a1", "v4"));
+		return Observable.fromArray(events);
+	}
+}

src/test/java/beamline/tests/UtilsTests.java

+package beamline.tests;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.sql.Date;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.extension.std.XTimeExtension;
+import org.deckfour.xes.model.XTrace;
+import org.junit.jupiter.api.Test;
+
+import beamline.exceptions.EventException;
+import beamline.utils.EventUtils;
+
+public class UtilsTests {
+
+	@Test
+	public void test_create_event() {
+		String activityName = UUID.randomUUID().toString();
+		String caseId = UUID.randomUUID().toString();
+		XTrace t = null;
+		try {
+			t = EventUtils.create(activityName, caseId);
+		} catch (EventException e) { }
+		
+		assertNotNull(t);
+		assertEquals(XConceptExtension.instance().extractName(t), caseId);
+		assertEquals(XConceptExtension.instance().extractName(t.get(0)), activityName);
+	}
+	
+	@Test
+	public void test_create_event_time() {
+		String activityName = UUID.randomUUID().toString();
+		String caseId = UUID.randomUUID().toString();
+		Date date = Date.valueOf("1996-01-23");
+		XTrace t = null;
+		try {
+			t = EventUtils.create(activityName, caseId, date);
+		} catch (EventException e) { }
+		
+		assertNotNull(t);
+		assertEquals(XConceptExtension.instance().extractName(t), caseId);
+		assertEquals(XConceptExtension.instance().extractName(t.get(0)), activityName);
+		assertEquals(XTimeExtension.instance().extractTimestamp(t.get(0)), date);
+	}
+	
+	@Test
+	public void test_create_event_attributes() {
+		String activityName = UUID.randomUUID().toString();
+		String caseId = UUID.randomUUID().toString();
+		Date date = Date.valueOf("1996-01-23");
+		List<Pair<String, String>> attributes = new LinkedList<Pair<String, String>>();
+		Map<String, String> values = new HashMap<String, String>();
+		for (int i = 0; i < 10; i++) {
+			String attributeName = "attr-" + i;
+			String attributeValue = UUID.randomUUID().toString();
+			values.put(attributeName, attributeValue);
+			attributes.add(Pair.of(attributeName, attributeValue));
+		}
+		XTrace t = null;
+		try {
+			t = EventUtils.create(activityName, caseId, date, attributes);
+		} catch (EventException e) { }
+		
+		assertNotNull(t);
+		assertEquals(XConceptExtension.instance().extractName(t), caseId);
+		assertEquals(XConceptExtension.instance().extractName(t.get(0)), activityName);
+		assertEquals(XTimeExtension.instance().extractTimestamp(t.get(0)), date);
+		for(String name : t.get(0).getAttributes().keySet()) {
+			if (name.startsWith("attr-")) {
+				assertEquals(t.get(0).getAttributes().get(name).toString(), values.get(name));
+			}
+		}
+	}
+	
+	@Test
+	public void test_no_activityname() {
+		assertThrows(EventException.class, () -> {
+			EventUtils.create(null, "");
+		});
+		assertThrows(EventException.class, () -> {
+			EventUtils.create("", null);
+		});
+	}
+	
+	@Test
+	public void test_extract_name_case() throws EventException {
+		String activityName = UUID.randomUUID().toString();
+		String caseId = UUID.randomUUID().toString();
+		XTrace t = EventUtils.create(activityName, caseId);
+		assertEquals(activityName, EventUtils.getActivityName(t));
+		assertEquals(caseId, EventUtils.getCaseId(t));
+	}
+}