Initial prototype of the system

src/main/java/discovery/beamline/view/graph/PMDotEndNode.java

+package discovery.beamline.view.graph;
+
+import beamline.graphviz.DotNode;
+
+public class PMDotEndNode extends DotNode {
+	public PMDotEndNode() {
+		super("", null);
+
+		setOption("shape", "circle");
+		setOption("style", "filled");
+		setOption("fillcolor", "#D8BBB9"); // #D8BBB9:#BC9F9D
+		setOption("gradientangle", "270");
+		setOption("color", "#614847");
+		setOption("height", "0.13");
+		setOption("width", "0.13");
+	}
+
+	@Override
+	public String toString() {
+		return "{ rank = \"sink\"; " + super.toString() + "}";
+	}
+}
\ No newline at end of file

src/main/java/discovery/beamline/view/graph/PMDotModel.java

+package discovery.beamline.view.graph;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import beamline.graphviz.Dot;
+import beamline.graphviz.DotNode;
+import discovery.beamline.miners.basic.ProcessMap;
+
+
+/**
+ *
+ * @author Andrea Burattin
+ */
+public class PMDotModel extends Dot {
+
+	private ProcessMap model;
+	private ColorPalette.Colors activityColor;
+
+	public PMDotModel(ProcessMap model, ColorPalette.Colors activityColor) {
+		this.model = model;
+		this.activityColor = activityColor;
+
+		realize();
+	}
+
+	private void realize() {
+//		setOption("rankdir", "LR");
+		setOption("ranksep", ".1");
+		setOption("fontsize", "9");
+		setOption("remincross", "true");
+		setOption("margin", "0.0,0.0");
+		setOption("outputorder", "edgesfirst");
+
+		Map<String, DotNode> activityToNode = new HashMap<String, DotNode>();
+		Map<String, String> nodeToActivity = new HashMap<String, String>();
+
+		Set<DotNode> startNodes = new HashSet<DotNode>();
+		Set<DotNode> endNodes = new HashSet<DotNode>();
+
+		// add all activities
+		for(String activity : model.getActivities()) {
+			DotNode node = addNodeIfNeeded(activity, activityToNode, nodeToActivity);
+			if (node instanceof PMDotNode) {
+				((PMDotNode) node).setColorWeight(model.getActivityValue(activity), activityColor);
+			}
+			if (model.isStartActivity(activity)) {
+				startNodes.add(node);
+			}
+			if (model.isEndActivity(activity)) {
+				endNodes.add(node);
+			}
+		}
+
+		// add all relations
+		for (Pair<String, String> relation : model.getRelations()) {
+			String sourceActivity = relation.getLeft();
+			String targetActivity = relation.getRight();
+
+			// adding source nodes
+			DotNode sourceNode = addNodeIfNeeded(sourceActivity, activityToNode, nodeToActivity);
+			// adding target nodes
+			DotNode targetNode = addNodeIfNeeded(targetActivity, activityToNode, nodeToActivity);
+
+			// adding relations
+			addRelation(sourceNode, targetNode, model.getRelationValue(relation));
+		}
+
+		// add relations from start and end
+		if (startNodes.size() > 0) {
+			PMDotStartNode start = new PMDotStartNode();
+			addNode(start);
+			for (DotNode n : startNodes) {
+				addRelation(start, n, null);
+			}
+		}
+		if (endNodes.size() > 0) {
+			PMDotEndNode end = new PMDotEndNode();
+			addNode(end);
+			for (DotNode n : endNodes) {
+				addRelation(n, end, null);
+			}
+		}
+	}
+
+	private void addRelation(DotNode sourceNode, DotNode targetNode, Double value) {
+		addEdge(new PMDotEdge(sourceNode, targetNode, (value == null? null : String.format("%.2g%n", value)), value));
+	}
+
+	private DotNode addNodeIfNeeded(String activity, Map<String, DotNode> activityToNode, Map<String, String> nodeToActivity) {
+		DotNode existingNode = activityToNode.get(activity);
+		if (existingNode == null) {
+//			if (model.isStartActivity(activity)) {
+//				PMCEPDotStartNode startNode = new PMCEPDotStartNode();
+//				addNode(startNode);
+//				activityToNode.put(activity, startNode);
+//				nodeToActivity.put(startNode.getId(), activity);
+//				return startNode;
+//			} else if (model.isEndActivity(activity)) {
+//				PMCEPDotEndNode endNode = new PMCEPDotEndNode();
+//				addNode(endNode);
+//				activityToNode.put(activity, endNode);
+//				nodeToActivity.put(endNode.getId(), activity);
+//				return endNode;
+//			} else {
+				PMDotNode newNode = new PMDotNode(activity.toString());
+				newNode.setColorWeight(model.getActivityValue(activity), activityColor);
+				newNode.setSecondLine(String.format("%.2g%n", model.getActivityValue(activity)));
+				addNode(newNode);
+				activityToNode.put(activity, newNode);
+				nodeToActivity.put(newNode.getId(), activity);
+				return newNode;
+//			}
+		} else {
+			return existingNode;
+		}
+	}
+}
\ No newline at end of file

src/main/java/discovery/beamline/view/graph/PMDotNode.java

+package discovery.beamline.view.graph;
+
+import java.awt.Color;
+
+import beamline.graphviz.DotNode;
+
+public class PMDotNode extends DotNode {
+
+	private String label;
+
+	public PMDotNode(String label) {
+		this(label, null, null, null);
+	}
+
+	public PMDotNode(String label, String secondLine, Double weight, ColorPalette.Colors activityColor) {
+		super(label, null);
+
+		this.label = label;
+
+		setOption("shape", "box");
+		setOption("fixedsize", "true");
+		setOption("height", "0.23");
+		setOption("width", "1.2");
+		setOption("style", "rounded,filled");
+		setOption("fontname", "Arial");
+
+		setSecondLine(secondLine);
+		setColorWeight(weight, activityColor);
+	}
+
+	public void setSecondLine(String secondLine) {
+		if (secondLine != null) {
+			setLabel("<<font point-size='22'>" + label + "</font> <br/><font point-size='16'>" + secondLine
+					+ "</font>>");
+		}
+	}
+
+	public void setColorWeight(Double weight, ColorPalette.Colors activityColor) {
+		if (weight == null) {
+			setOption("fillcolor", "#FDEFD8"); // #FDEFD8:#E1D3BC
+		} else {
+			Color backgroundColor = ColorPalette.getValue(activityColor, weight);
+			Color fontColor = ColorPalette.getFontColor(backgroundColor);
+			setOption("fillcolor", ColorPalette
+					.colorToString(backgroundColor)/* + ":" + ColorPalette.colorToString(backgroundColor.darker()) */);
+			setOption("fontcolor", ColorPalette.colorToString(fontColor));
+			setOption("fixedsize", "false");
+		}
+	}
+
+	public void setMovedIn() {
+		setOption("fillcolor", "white");
+	}
+
+	public void setMovedOut() {
+		setOption("fillcolor", "black");
+	}
+
+	@Override
+	public int hashCode() {
+		return getLabel().hashCode();
+	}
+
+	@Override
+	public boolean equals(Object object) {
+		return getLabel().equals(object);
+	}
+}
\ No newline at end of file

src/main/java/discovery/beamline/view/graph/PMDotStartNode.java

+package discovery.beamline.view.graph;
+
+import beamline.graphviz.DotNode;
+
+public class PMDotStartNode extends DotNode {
+	public PMDotStartNode() {
+		super("", null);
+
+		setOption("shape", "circle");
+		setOption("style", "filled");
+		setOption("fillcolor", "#CED6BD"); // #CED6BD:#B3BBA2
+		setOption("gradientangle", "270");
+		setOption("color", "#595F45");
+		setOption("height", "0.13");
+		setOption("width", "0.13");
+	}
+
+	@Override
+	public String toString() {
+		return "{ rank = \"source\"; " + super.toString() + " }";
+	}
+}
\ No newline at end of file

src/main/java/rx/beamline/Tester2.java

+package rx.beamline;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.HashSet;
+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 discovery.beamline.miners.basic.ProcessMap;
+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.functions.Consumer;
+import io.reactivex.rxjava3.functions.Function;
+import io.reactivex.rxjava3.functions.Predicate;
+import jdk.jshell.execution.StreamingExecutionControl;
+import rx.beamline.filters.ExcludeActivitiesFilter;
+import rx.beamline.filters.RetainActivitiesFilter;
+import rx.beamline.miners.DiscoveryMiner;
+import rx.beamline.miners.HookAroundEventProcessing;
+import rx.beamline.miners.StreamMiningAlgorithm;
+import rx.beamline.sources.XesLogSource;
+import rx.beamline.sources.XesSource;
+
+public class Tester2 {
+
+	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(broker, topic, process);
+		source.prepare();
+		
+		DiscoveryMiner miner = new DiscoveryMiner();
+		miner.setMinDependency(0.3);
+		miner.setModelRefreshRate(1);
+		miner.setOnAfterEvent(new HookAroundEventProcessing() {
+			@Override
+			public void run() {
+				System.out.println(miner.getProcessedEvents());
+			}
+		});
+		
+		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);
+//		.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"));
+		
+		
+//		Observable<String> stream = Observable.just("4444", "22", "333", "55555", "aaa", "bbb");
+//		
+//		stream
+//		.compose(new ObservableTransformer<String, Integer>() {
+//			@Override
+//			public @NonNull ObservableSource<@NonNull Integer> apply(@NonNull Observable<@NonNull String> upstream) {
+//				return upstream
+//						.filter(new Predicate<String>() {
+//							@Override
+//							public boolean test(@NonNull String t) throws Throwable {
+//								return t.length() > 0;
+//							}
+//						})
+//						.map(new Function<String, Integer>() {
+//							@Override
+//							public @NonNull Integer apply(@NonNull String t) throws Throwable {
+//								return t.length();
+//							}
+//						})
+//						;
+//			}
+//		})
+////		.filter(new Predicate<String>() {
+////			@Override
+////			public boolean test(@NonNull String t) throws Throwable {
+////				return t.length() > 0;
+////			}
+////		})
+////		.map(new Function<String, Integer>() {
+////			@Override
+////			public @NonNull Integer apply(@NonNull String t) throws Throwable {
+////				return t.length();
+////			}
+////		})
+////		.buffer(4)
+//		.subscribe(new Consumer<Integer>() {
+//			@Override
+//			public void accept(@NonNull Integer t) throws Throwable {
+//				System.out.println(t);
+//			}
+//		})
+//		;
+		System.out.println("done");
+	}
+
+}

src/main/java/rx/beamline/filters/ExcludeActivitiesFilter.java

+package rx.beamline.filters;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.model.XAttributeLiteral;
+import org.deckfour.xes.model.impl.XAttributeLiteralImpl;
+
+public class ExcludeActivitiesFilter extends ExcludeOnEventAttributeEqualityFilter<XAttributeLiteral> {
+
+	public ExcludeActivitiesFilter(String ...activities) {
+		super(XConceptExtension.KEY_NAME);
+		
+		for (String activity : activities) {
+			addValue(new XAttributeLiteralImpl(XConceptExtension.KEY_NAME, activity));
+		}
+	}
+}

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

+package rx.beamline.filters;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.deckfour.xes.model.XAttribute;
+import org.deckfour.xes.model.XTrace;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.functions.Predicate;
+
+public class ExcludeOnCaseAttributeEqualityFilter<T extends XAttribute> implements Predicate<XTrace> {
+
+	private String attributeName;
+	private Set<T> attributeValues;
+	
+	public ExcludeOnCaseAttributeEqualityFilter(String attributeName, T ...values) {
+		this.attributeName = attributeName;
+		this.attributeValues = new HashSet<T>(Arrays.asList(values));
+	}
+	
+	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/rx/beamline/filters/ExcludeOnEventAttributeEqualityFilter.java

+package rx.beamline.filters;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.deckfour.xes.model.XAttribute;
+import org.deckfour.xes.model.XTrace;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.functions.Predicate;
+
+public class ExcludeOnEventAttributeEqualityFilter<T extends XAttribute> implements Predicate<XTrace> {
+
+	private String attributeName;
+	private Set<T> attributeValues;
+	
+	public ExcludeOnEventAttributeEqualityFilter(String attributeName, T ...values) {
+		this.attributeName = attributeName;
+		this.attributeValues = new HashSet<T>(Arrays.asList(values));
+	}
+	
+	public void addValue(T value) {
+		this.attributeValues.add(value);
+	}
+	
+	@Override
+	public boolean test(@NonNull XTrace t) throws Throwable {
+		return !attributeValues.contains(t.get(0).getAttributes().get(attributeName));
+	}
+
+}

src/main/java/rx/beamline/filters/RetainActivitiesFilter.java

+package rx.beamline.filters;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.model.XAttributeLiteral;
+import org.deckfour.xes.model.impl.XAttributeLiteralImpl;
+
+public class RetainActivitiesFilter extends RetainOnEventAttributeEqualityFilter<XAttributeLiteral> {
+
+	public RetainActivitiesFilter(String ...activities) {
+		super(XConceptExtension.KEY_NAME);
+		
+		for (String activity : activities) {
+			addValue(new XAttributeLiteralImpl(XConceptExtension.KEY_NAME, activity));
+		}
+	}
+}

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

+package rx.beamline.filters;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.deckfour.xes.model.XAttribute;
+import org.deckfour.xes.model.XTrace;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.functions.Predicate;
+
+public class RetainOnCaseAttributeEqualityFilter<T extends XAttribute> implements Predicate<XTrace> {
+
+	private String attributeName;
+	private Set<T> attributeValues;
+	
+	public RetainOnCaseAttributeEqualityFilter(String attributeName, T ...values) {
+		this.attributeName = attributeName;
+		this.attributeValues = new HashSet<T>(Arrays.asList(values));
+	}
+	
+	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/rx/beamline/filters/RetainOnEventAttributeEqualityFilter.java

+package rx.beamline.filters;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.deckfour.xes.model.XAttribute;
+import org.deckfour.xes.model.XTrace;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.functions.Predicate;
+
+public class RetainOnEventAttributeEqualityFilter<T extends XAttribute> implements Predicate<XTrace> {
+
+	private String attributeName;
+	private Set<T> attributeValues;
+	
+	public RetainOnEventAttributeEqualityFilter(String attributeName, T ...values) {
+		this.attributeName = attributeName;
+		this.attributeValues = new HashSet<T>(Arrays.asList(values));
+	}
+	
+	public void addValue(T value) {
+		this.attributeValues.add(value);
+	}
+	
+	@Override
+	public boolean test(@NonNull XTrace t) throws Throwable {
+		return attributeValues.contains(t.get(0).getAttributes().get(attributeName));
+	}
+
+}

src/main/java/rx/beamline/miners/DiscoveryMiner.java

+package rx.beamline.miners;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.model.XTrace;
+
+import discovery.beamline.miners.basic.ProcessMap;
+
+public class DiscoveryMiner extends StreamMiningAlgorithm<XTrace, ProcessMap> {
+
+	private Map<String, String> latestActivityInCase = new HashMap<String, String>();
+	private Map<Pair<String, String>, Double> relations = new HashMap<Pair<String, String>, Double>();
+	private Map<String, Double> activities = new HashMap<String, Double>();
+	private Double maxActivityFreq = Double.MIN_VALUE;
+	private Double maxRelationsFreq = Double.MIN_VALUE;
+	private double minDependency = 1d;
+	private int modelRefreshRate = 0;
+	
+	public void setMinDependency(double minDependency) {
+		this.minDependency = minDependency;
+	}
+
+	public void setModelRefreshRate(int modelRefreshRate) {
+		this.modelRefreshRate = modelRefreshRate;
+	}
+	
+	@Override
+	public ProcessMap ingest(XTrace event) {
+		String caseID = XConceptExtension.instance().extractName(event);
+		String activityName = XConceptExtension.instance().extractName(event.get(0));
+		
+		Double activityFreq = 1d;
+		if (activities.containsKey(activityName)) {
+			activityFreq += activities.get(activityName);
+			maxActivityFreq = Math.max(maxActivityFreq, activityFreq);
+		}
+		activities.put(activityName, activityFreq);
+
+		if (latestActivityInCase.containsKey(caseID)) {
+			Pair<String, String> relation = new ImmutablePair<String, String>(latestActivityInCase.get(caseID), activityName);
+			Double relationFreq = 1d;
+			if (relations.containsKey(relation)) {
+				relationFreq += relations.get(relation);
+				maxRelationsFreq = Math.max(maxRelationsFreq, relationFreq);
+			}
+			relations.put(relation, relationFreq);
+		}
+		latestActivityInCase.put(caseID, activityName);
+
+		if (getProcessedEvents() % modelRefreshRate == 0) {
+			setLatestResponse(mine(minDependency));
+		}
+		
+		return getLatestResponse();
+	}
+	
+	public ProcessMap mine(double threshold) {
+		ProcessMap process = new ProcessMap();
+		for (String activity : activities.keySet()) {
+			process.addActivity(activity, activities.get(activity) / maxActivityFreq);
+		}
+		for (Pair<String, String> relation : relations.keySet()) {
+			double dependency = relations.get(relation) / maxRelationsFreq;
+			if (dependency >= threshold) {
+				process.addRelation(relation.getLeft(), relation.getRight(), dependency);
+			}
+		}
+		Set<String> toRemove = new HashSet<String>();
+		Set<String> selfLoopsToRemove = new HashSet<String>();
+		for (String activity : activities.keySet()) {
+			if (process.isStartActivity(activity) && process.isEndActivity(activity)) {
+				toRemove.add(activity);
+			}
+			if (process.isIsolatedNode(activity)) {
+				selfLoopsToRemove.add(activity);
+			}
+		}
+		for (String activity : toRemove) {
+			process.removeActivity(activity);
+		}
+
+		return process;
+	}
+}

src/main/java/rx/beamline/miners/HookAroundEventProcessing.java

+package rx.beamline.miners;
+
+public interface HookAroundEventProcessing {
+
+	public void run();
+}

src/main/java/rx/beamline/miners/StreamMiningAlgorithm.java

+package rx.beamline.miners;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.functions.Consumer;
+
+public abstract class StreamMiningAlgorithm<T, K> implements Consumer<T> {
+
+	private int processedEvents = 0;
+	private K latestResponse;
+	private HookAroundEventProcessing onBeforeEvent = null;
+	private HookAroundEventProcessing onAfterEvent = null;
+	
+	public abstract K ingest(T event);
+	
+	public void process(T event) {
+		this.processedEvents++;
+		latestResponse = ingest(event);
+	}
+	
+	public int getProcessedEvents() {
+		return processedEvents;
+	}
+	
+	public K getLatestResponse() {
+		return latestResponse;
+	}
+	
+	public void setOnBeforeEvent(HookAroundEventProcessing onBeforeEvent) {
+		this.onBeforeEvent = onBeforeEvent;
+	}
+	
+	public void setOnAfterEvent(HookAroundEventProcessing onAfterEvent) {
+		this.onAfterEvent = onAfterEvent;
+	}
+	
+	protected K setLatestResponse(K latestResponse) {
+		this.latestResponse = latestResponse;
+		return latestResponse;
+	}
+	
+	@Override
+	public void accept(@NonNull T t) throws Throwable {
+		if (onBeforeEvent != null) {
+			onBeforeEvent.run();
+		}
+		process(t);
+		if (onAfterEvent != null) {
+			onAfterEvent.run();
+		}
+	}
+}

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

+package rx.beamline.sources;
+
+import java.util.Date;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import org.deckfour.xes.extension.std.XConceptExtension;
+import org.deckfour.xes.extension.std.XTimeExtension;
+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;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableEmitter;
+import io.reactivex.rxjava3.core.ObservableOnSubscribe;
+
+public class MQTTXesSource implements XesSource {
+
+	private String processName;
+	private String brokerHost;
+	private String topicBase;
+	
+	public MQTTXesSource(String brokerHost, String topicBase, String processName) {
+		this.brokerHost = brokerHost;
+		this.topicBase = topicBase;
+		this.processName = processName;
+//		this.stream = Stream.generate(new Supplier<XTrace>() {
+//			@Override
+//			public XTrace get() {
+//				try {
+//					return queue.take();
+//				} catch (InterruptedException e) {
+//					e.printStackTrace();
+//				}
+//				return null;
+//			}
+//		});
+	}
+	
+	@Override
+	public Observable<XTrace> getObservable() {
+		return Observable.create(new ObservableOnSubscribe<XTrace>() {
+
+			@Override
+			public void subscribe(@NonNull ObservableEmitter<@NonNull XTrace> emitter) throws Throwable {
+				// TODO Auto-generated method stub
+				
+			}
+			
+		});
+	}
+
+	@Override
+	public void prepare() throws Exception {
+//		MqttConnectOptions options = new MqttConnectOptions();
+//		options.setCleanSession(true);
+//		options.setKeepAliveInterval(30);
+//
+//		IMqttClient myClient = new MqttClient(brokerHost, UUID.randomUUID().toString());
+//		myClient.setCallback(new MqttCallback() {
+//			
+//			@Override
+//			public void messageArrived(String topic, MqttMessage message) throws Exception {
+//				int posLastSlash = topic.lastIndexOf("/");
+//				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);
+//				
+//				queue.add(eventWrapper);
+//			}
+//			
+//			@Override
+//			public void deliveryComplete(IMqttDeliveryToken token) { }
+//			
+//			@Override
+//			public void connectionLost(Throwable cause) { }
+//		});
+//		myClient.connect(options);
+//		myClient.subscribe(topicBase + "/" + processName + "/#");
+	}
+
+}

src/main/java/rx/beamline/sources/Source.java

+package rx.beamline.sources;
+
+import io.reactivex.rxjava3.core.Observable;
+
+public interface Source<T> {
+
+	public Observable<T> getObservable();
+	
+	public void prepare() throws Exception;
+}

src/main/java/rx/beamline/sources/XesLogSource.java

+package rx.beamline.sources;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.deckfour.xes.extension.std.XTimeExtension;
+import org.deckfour.xes.factory.XFactory;
+import org.deckfour.xes.factory.XFactoryNaiveImpl;
+import org.deckfour.xes.in.XParser;
+import org.deckfour.xes.in.XesXmlGZIPParser;
+import org.deckfour.xes.in.XesXmlParser;
+import org.deckfour.xes.model.XAttributeMap;
+import org.deckfour.xes.model.XEvent;
+import org.deckfour.xes.model.XLog;
+import org.deckfour.xes.model.XTrace;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableEmitter;
+import io.reactivex.rxjava3.core.ObservableOnSubscribe;
+
+public class XesLogSource implements XesSource {
+
+private static XFactory xesFactory = new XFactoryNaiveImpl();
+	
+	private String fileName;
+	private XLog log;
+	private List<XTrace> events;
+	
+	public XesLogSource(String fileName) {
+		this.fileName = fileName;
+	}
+	
+	public XesLogSource(XLog log) {
+		this.log = log;
+	}
+	
+	public Observable<XTrace> getObservable() {
+		return Observable.create(new ObservableOnSubscribe<XTrace>() {
+			@Override
+			public void subscribe(@NonNull ObservableEmitter<@NonNull XTrace> emitter) throws Throwable {
+				for (XTrace wrapper : events) {
+					emitter.onNext(wrapper);
+				}
+			}
+		});
+	}
+	
+	public void prepare() throws Exception {
+		if (log == null) {
+			parseLog(fileName);
+		}
+		prepareStream();
+	}
+	
+	private void parseLog(String fileName) throws Exception {
+		XParser[] parsers = new XParser[] { new XesXmlGZIPParser(), new XesXmlParser() };
+		File file = new File(fileName);
+		for (XParser p : parsers) {
+			if (p.canParse(file)) {
+				log = p.parse(file).get(0);
+				return;
+			}
+		}
+		throw new Exception("XES file format not supported");
+	}
+	
+	private void prepareStream() {
+		if (log == null) {
+			return;
+		}
+		// populate all events
+		events = new LinkedList<XTrace>();
+		for (XTrace t : log) {
+			for (XEvent e : t) {
+				// create the wrapping trace
+				XTrace eventWrapper = xesFactory.createTrace();
+				XAttributeMap am = t.getAttributes();
+				for (String key : am.keySet()) {
+					eventWrapper.getAttributes().put(key, am.get(key));
+				}
+				// create the actual event
+				XEvent newEvent = xesFactory.createEvent();
+				XAttributeMap amEvent = e.getAttributes();
+				for (String key : amEvent.keySet()) {
+					newEvent.getAttributes().put(key, amEvent.get(key));
+				}
+				eventWrapper.add(newEvent);
+				events.add(eventWrapper);
+			}
+		}
+		
+		// sort events
+		Collections.sort(events, new Comparator<XTrace>() {
+			public int compare(XTrace o1, XTrace o2) {
+				XEvent e1 = o1.get(0);
+				XEvent e2 = o2.get(0);
+				Date d1 = XTimeExtension.instance().extractTimestamp(e1);
+				Date d2 = XTimeExtension.instance().extractTimestamp(e2);
+				if (d1 == null || d2 == null) {
+					return 0;
+				}
+				return d1.compareTo(d2);
+			}
+		});
+	}
+}

src/main/java/rx/beamline/sources/XesSource.java

+package rx.beamline.sources;
+
+import org.deckfour.xes.model.XTrace;
+
+public interface XesSource extends Source<XTrace> {
+
+}
\ No newline at end of file