<commit_before>/**
 * Copyright 2014 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.dsys.snio.impl.handler;

import static net.dsys.snio.impl.handler.ExecutionType.DECOUPLED;
import static net.dsys.snio.impl.handler.ExecutionType.ZERO_COPY;
import static net.dsys.snio.impl.handler.HandlerType.MULTI_THREADED;
import static net.dsys.snio.impl.handler.HandlerType.SINGLE_THREADED;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import net.dsys.commons.api.exception.Bug;
import net.dsys.commons.api.lang.Cleaner;
import net.dsys.commons.api.lang.Copier;
import net.dsys.commons.api.lang.Factory;
import net.dsys.commons.api.lang.Interruptible;
import net.dsys.commons.impl.builder.Mandatory;
import net.dsys.commons.impl.builder.OptionGroup;
import net.dsys.commons.impl.builder.Optional;
import net.dsys.commons.impl.lang.ByteBufferCopier;
import net.dsys.commons.impl.lang.ByteBufferFactory;
import net.dsys.commons.impl.lang.DirectByteBufferFactory;
import net.dsys.snio.api.buffer.MessageBufferConsumer;
import net.dsys.snio.api.buffer.MessageBufferProducer;
import net.dsys.snio.api.channel.AcceptListener;
import net.dsys.snio.api.handler.MessageConsumer;
import net.dsys.snio.api.handler.MessageConsumerFactory;
import net.dsys.snio.api.handler.MessageHandler;
import net.dsys.snio.api.handler.MessageProducer;

/**
 * @author <NAME>
 */
public final class MessageHandlers {

	private MessageHandlers() {
		// no instantiation
		return;
	}

	public static <T> Interruptible syncConsumer(final MessageBufferConsumer<T> in, final MessageConsumer<T> consumer) {
		return new ConsumerThread<>(in, consumer);
	}

	public static <T> Interruptible asyncConsumer(final MessageBufferConsumer<T> in, final MessageConsumer<T> consumer,
			final T holder, final Copier<T> copier, final Cleaner<T> cleaner) {
		return new ConsumerThread<>(in, consumer, holder, copier, cleaner);
	}

	public static <T> Interruptible syncProducer(final MessageBufferProducer<T> out,
			final MessageProducer<T> producer) {
		return new ProducerThread<>(out, producer);
	}

	public static <T> Interruptible asyncProducer(final MessageBufferProducer<T> out,
			final MessageProducer<T> producer, final T holder, final Copier<T> copier, final Cleaner<T> cleaner) {
		return new ProducerThread<>(out, producer, holder, copier, cleaner);
	}

	public static HandlerBuilder buildHandler() {
		return new HandlerBuilder();
	}

	/**
	 * @author <NAME>
	 */
	public static final class HandlerBuilder {

		private HandlerType handlerType;
		private ExecutionType threadType;
		private ExecutorService executor;
		private MessageConsumer<ByteBuffer> consumer;
		private MessageConsumerFactory<ByteBuffer> consumerFactory;
		private AcceptListener<ByteBuffer> delegate;
		private int length;
		private boolean useDirectBuffer;

		HandlerBuilder() {
			this.handlerType = null;
			this.threadType = ZERO_COPY;
			this.consumer = null;
			this.consumerFactory = null;
			this.delegate = null;
			this.length = 0;
			this.useDirectBuffer = false;
		}

		@Optional(defaultValue = "Executors.newCachedThreadPool()", restrictions = "executor != null")
		public void setExecutor(final ExecutorService executor) {
			if (executor == null) {
				throw new NullPointerException("executor == null");
			}
			this.executor = executor;
		}

		@Mandatory(restrictions = "consumer != null")
		@OptionGroup(name = "consumer", seeAlso = "useManyConsumers(factory)")
		public HandlerBuilder useSingleConsumer(final MessageConsumer<ByteBuffer> consumer) {
			if (consumer == null) {
				throw new NullPointerException("consumer == null");
			}
			this.handlerType = SINGLE_THREADED;
			this.consumer = consumer;
			this.consumerFactory = null;
			return this;
		}

		@Mandatory(restrictions = "factory != null")
		@OptionGroup(name = "consumer", seeAlso = "useSingleConsumer(consumer)")
		public HandlerBuilder useManyConsumers(final MessageConsumerFactory<ByteBuffer> factory) {
			if (factory == null) {
				throw new NullPointerException("consumerFactory == null");
			}
			this.handlerType = MULTI_THREADED;
			this.consumerFactory = factory;
			this.consumer = null;
			return this;
		}

		@Optional(defaultValue = "useZeroCopyProcessing()", restrictions = "none")
		@OptionGroup(name = "execution", seeAlso = "useDecoupledProcessing(length)")
		public HandlerBuilder useZeroCopyProcessing() {
			this.threadType = ZERO_COPY;
			this.length = 0;
			return this;
		}

		@Optional(defaultValue = "useZeroCopyProcessing()", restrictions = "messageLength > 0")
		@OptionGroup(name = "execution", seeAlso = "useZeroCopyProcessing()")
		public HandlerBuilder useDecoupledProcessing(final int messageLength) {
			this.threadType = DECOUPLED;
			this.length = messageLength;
			return this;
		}

		@Optional(defaultValue = "useHeapBuffer()")
		@OptionGroup(name = "bufferType", seeAlso = "useHeapBuffer()")
		public HandlerBuilder useDirectBuffer() {
			this.useDirectBuffer = true;
			return this;
		}

		@Optional(defaultValue = "useHeapBuffer()")
		@OptionGroup(name = "bufferType", seeAlso = "useDirectBuffer()")
		public HandlerBuilder useHeapBuffer() {
			this.useDirectBuffer = false;
			return this;
		}

		@Optional(defaultValue = "null")
		public HandlerBuilder setDelegate(final AcceptListener<ByteBuffer> delegate) {
			this.delegate = delegate;
			return this;
		}

		public MessageHandler<ByteBuffer> build() {
			ExecutorService exec = executor;
			if (exec == null) {
				exec = Executors.newCachedThreadPool();
			}
			final ConsumerThreadFactory<ByteBuffer> threads;
			switch (threadType) {
			case DECOUPLED: {
				final Factory<ByteBuffer> factory;
				if (useDirectBuffer) {<commit_msg>Added name property to MessageHandler builder.<commit_after>/**
 * Copyright 2014 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.dsys.snio.impl.handler;

import static net.dsys.snio.impl.handler.ExecutionType.DECOUPLED;
import static net.dsys.snio.impl.handler.ExecutionType.ZERO_COPY;
import static net.dsys.snio.impl.handler.HandlerType.MULTI_THREADED;
import static net.dsys.snio.impl.handler.HandlerType.SINGLE_THREADED;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import net.dsys.commons.api.exception.Bug;
import net.dsys.commons.api.lang.Cleaner;
import net.dsys.commons.api.lang.Copier;
import net.dsys.commons.api.lang.Factory;
import net.dsys.commons.api.lang.Interruptible;
import net.dsys.commons.impl.builder.Mandatory;
import net.dsys.commons.impl.builder.OptionGroup;
import net.dsys.commons.impl.builder.Optional;
import net.dsys.commons.impl.lang.ByteBufferCopier;
import net.dsys.commons.impl.lang.ByteBufferFactory;
import net.dsys.commons.impl.lang.DaemonThreadFactory;
import net.dsys.commons.impl.lang.DirectByteBufferFactory;
import net.dsys.snio.api.buffer.MessageBufferConsumer;
import net.dsys.snio.api.buffer.MessageBufferProducer;
import net.dsys.snio.api.channel.AcceptListener;
import net.dsys.snio.api.handler.MessageConsumer;
import net.dsys.snio.api.handler.MessageConsumerFactory;
import net.dsys.snio.api.handler.MessageHandler;
import net.dsys.snio.api.handler.MessageProducer;

/**
 * @author <NAME>
 */
public final class MessageHandlers {

	private MessageHandlers() {
		// no instantiation
		return;
	}

	public static <T> Interruptible syncConsumer(final MessageBufferConsumer<T> in, final MessageConsumer<T> consumer) {
		return new ConsumerThread<>(in, consumer);
	}

	public static <T> Interruptible asyncConsumer(final MessageBufferConsumer<T> in, final MessageConsumer<T> consumer,
			final T holder, final Copier<T> copier, final Cleaner<T> cleaner) {
		return new ConsumerThread<>(in, consumer, holder, copier, cleaner);
	}

	public static <T> Interruptible syncProducer(final MessageBufferProducer<T> out,
			final MessageProducer<T> producer) {
		return new ProducerThread<>(out, producer);
	}

	public static <T> Interruptible asyncProducer(final MessageBufferProducer<T> out,
			final MessageProducer<T> producer, final T holder, final Copier<T> copier, final Cleaner<T> cleaner) {
		return new ProducerThread<>(out, producer, holder, copier, cleaner);
	}

	public static HandlerBuilder buildHandler() {
		return new HandlerBuilder();
	}

	/**
	 * @author <NAME>
	 */
	public static final class HandlerBuilder {

		private static AtomicInteger counter = new AtomicInteger();

		private String name;
		private HandlerType handlerType;
		private ExecutionType threadType;
		private ExecutorService executor;
		private MessageConsumer<ByteBuffer> consumer;
		private MessageConsumerFactory<ByteBuffer> consumerFactory;
		private AcceptListener<ByteBuffer> delegate;
		private int length;
		private boolean useDirectBuffer;

		HandlerBuilder() {
			this.name = "MessageHandler-" + counter.getAndIncrement();
			this.handlerType = null;
			this.threadType = ZERO_COPY;
			this.consumer = null;
			this.consumerFactory = null;
			this.delegate = null;
			this.length = 0;
			this.useDirectBuffer = false;
		}

		@Optional(defaultValue = "MessageHandler-#", restrictions = "name != null")
		@OptionGroup(name = "executor", seeAlso = "setExecutor(executor)")
		public void setName(final String name) {
			if (name == null) {
				throw new NullPointerException("name == null");
			}
			this.name = name;
		}

		@Optional(defaultValue = "Executors.newCachedThreadPool(new DaemonThreadFactory(name))",
				restrictions = "executor != null")
		@OptionGroup(name = "executor", seeAlso = "setName(name)")
		public void setExecutor(final ExecutorService executor) {
			if (executor == null) {
				throw new NullPointerException("executor == null");
			}
			this.executor = executor;
		}

		@Mandatory(restrictions = "consumer != null")
		@OptionGroup(name = "consumer", seeAlso = "useManyConsumers(factory)")
		public HandlerBuilder useSingleConsumer(final MessageConsumer<ByteBuffer> consumer) {
			if (consumer == null) {
				throw new NullPointerException("consumer == null");
			}
			this.handlerType = SINGLE_THREADED;
			this.consumer = consumer;
			this.consumerFactory = null;
			return this;
		}

		@Mandatory(restrictions = "factory != null")
		@OptionGroup(name = "consumer", seeAlso = "useSingleConsumer(consumer)")
		public HandlerBuilder useManyConsumers(final MessageConsumerFactory<ByteBuffer> factory) {
			if (factory == null) {
				throw new NullPointerException("consumerFactory == null");
			}
			this.handlerType = MULTI_THREADED;
			this.consumerFactory = factory;
			this.consumer = null;
			return this;
		}

		@Optional(defaultValue = "useZeroCopyProcessing()", restrictions = "none")
		@OptionGroup(name = "execution", seeAlso = "useDecoupledProcessing(length)")
		public HandlerBuilder useZeroCopyProcessing() {
			this.threadType = ZERO_COPY;
			this.length = 0;
			return this;
		}

		@Optional(defaultValue = "useZeroCopyProcessing()", restrictions = "messageLength > 0")
		@OptionGroup(name = "execution", seeAlso = "useZeroCopyProcessing()")
		public HandlerBuilder useDecoupledProcessing(final int messageLength) {
			this.threadType = DECOUPLED;
			this.length = messageLength;
			return this;
		}

		@Optional(defaultValue = "useHeapBuffer()")
		@OptionGroup(name = "bufferType", seeAlso = "useHeapBuffer()")
		public HandlerBuilder useDirectBuffer() {
			this.useDirectBuffer = true;
			return this;
		}

		@Optional(defaultValue = "useHeapBuffer()")
		@OptionGroup(name = "bufferType", seeAlso = "useDirectBuffer()")
		public HandlerBuilder useHeapBuffer() {
			this.useDirectBuffer = false;
			return this;
		}

		@Optional(defaultValue = "null")
		public HandlerBuilder setDelegate(final AcceptListener<ByteBuffer> delegate) {
			this.delegate = delegate;
			return this;
		}

		public MessageHandler<ByteBuffer> build() {
			ExecutorService exec = executor;
			if (exec == null) {
				exec = Executors.newCachedThreadPool(new DaemonThreadFactory(name));
			}
			final ConsumerThreadFactory<ByteBuffer> threads;
			switch (threadType) {
			case DECOUPLED: {
				final Factory<ByteBuffer> factory;
				if (useDirectBuffer) {
{'fraction_non_alphanumeric': 0.1079136690647482, 'fraction_numerical': 0.0016660355925785686, 'mean_word_length': 4.940620782726046, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6270821', 'n_tokens_mistral': 4372, 'n_tokens_neox': 3893, 'n_words': 954}