@Override
	public EventSet run(EventSetContext context, EventSet eventSetInUse, Object data) {

		if (eventSetInUse != null
				&& eventSetInUse.getStatus() == EventSetStatus.COLLECTION_IN_PROGRESS) {
			return eventSetInUse;
		}

		List<EventSet> correlatingEventSets = context.findCorrelatingEventSets(
				variableName, data, inboundCorrelationToken);
		
		Iterator<EventSet> iter = correlatingEventSets.iterator();
		while (iter.hasNext()) {
			EventSet eventSet = iter.next();
			if (eventSet.getStatus() != EventSetStatus.COLLECTION_IN_PROGRESS) {
				iter.remove();
			}
		}
		
		if (outboundCorrelationToken != null && outboundCorrelationToken.length() > 0) {
			context.addCorrelationToken(outboundCorrelationToken, correlatingEventSets);
			return eventSetInUse;
		}
		
		if (correlatingEventSets.size() != 1) {
			return eventSetInUse;
		}

		EventSet correlatingEventSet = correlatingEventSets.get(0);
		
		if (eventSetInUse != null
				&& eventSetInUse.getStatus() == EventSetStatus.NEW) {
			context.abortAndPublishEventSet(correlatingEventSet);
			return eventSetInUse;
		}

		return correlatingEventSet;

	}