<commit_before><commit_msg>added first draft of DynamicDistributor<commit_after>package teetime.stage.basic.distributor;

import teetime.framework.InputPort;
import teetime.framework.OutputPort;
import teetime.framework.pipe.SpScPipeFactory;

public class DynamicDistributor<T> extends Distributor<T> {

	private static final SpScPipeFactory spScPipeFactory = new SpScPipeFactory();

	public enum DynamicPortAction {
		CREATE, REMOVE;
	}

	public static class DynamicPortActionContainer<T> {
		private final DynamicPortAction dynamicPortAction;
		private final InputPort<T> inputPort;

		public DynamicPortActionContainer(final DynamicPortAction dynamicPortAction, final InputPort<T> inputPort) {
			super();
			this.dynamicPortAction = dynamicPortAction;
			this.inputPort = inputPort;
		}

		public DynamicPortAction getDynamicPortAction() {
			return dynamicPortAction;
		}

		public InputPort<T> getInputPort() {
			return inputPort;
		}

	}

	@SuppressWarnings("rawtypes")
	private final InputPort<DynamicPortActionContainer> dynamicPortActionInputPort = createInputPort(DynamicPortActionContainer.class);

	@SuppressWarnings("unchecked")
	@Override
	protected void execute(final T element) {
		DynamicPortActionContainer<T> dynamicPortAction = dynamicPortActionInputPort.receive();
		switch (dynamicPortAction.dynamicPortAction) {
		case CREATE:
			OutputPort<T> newOutputPort = createOutputPort();
			InputPort<T> newInputPort = dynamicPortAction.inputPort;
			spScPipeFactory.create(newOutputPort, newInputPort);
			break;
		case REMOVE:
			// TODO implement "remove port at runtime"
			break;
		default:
			if (logger.isWarnEnabled()) {
				logger.warn("Unhandled switch case of " + DynamicPortAction.class.getName() + ": " + dynamicPortAction.dynamicPortAction);
			}
			break;
		}

		this.strategy.distribute((OutputPort<T>[]) this.getOutputPorts(), element);
	}
}
{'fraction_non_alphanumeric': 0.09013333333333333, 'fraction_numerical': 0.0, 'mean_word_length': 5.846715328467154, 'pattern_counts': {'":': 1, '<': 14, '<?xml version=': 0, '>': 14, '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': '419340', 'n_tokens_mistral': 604, 'n_tokens_neox': 536, 'n_words': 114}