/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.aries.osgi.functional.internal;

import org.apache.aries.osgi.functional.Event;

import java.util.Deque;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author Carlos Sierra Andrés
 */
class Tuple<T> implements Event<T> {

	public T t;
	private Deque<Runnable> _closingHandlers = new LinkedList<>();
	private DoublyLinkedList<Tuple<?>> _relatedTuples =
		new DoublyLinkedList<>();

	private Tuple(T t) {
		this(t, new LinkedList<>());
	}

	private Tuple(T t, Deque<Runnable> closingHandlers) {
		this.t = t;
		_closingHandlers = closingHandlers;
	}

	public DoublyLinkedList.Node<Tuple<?>> addRelatedTuple(Tuple<?> tuple) {
		DoublyLinkedList.Node<Tuple<?>> tupleNode = _relatedTuples.addLast(
			tuple);

		tuple.onTermination(tupleNode::remove);

		return tupleNode;
	}

	public static <T> Tuple<T> create(T t) {
		return new Tuple<>(t);
	}

	@Override
	public T getContent() {
		return t;
	}

	@Override
	public int hashCode() {
		return t.hashCode();
	}

	@Override
	public boolean equals(Object o) {
		if (this == o) return true;
		if (o == null || getClass() != o.getClass()) return false;

		Tuple<?> tuple = (Tuple<?>) o;

		return t.equals(tuple.t);
	}

	public <S> Tuple<S> map(Function<? super T, ? extends S> fun) {
		return new Tuple<>(fun.apply(t), _closingHandlers);
	}

	public void onTermination(Runnable terminator) {
		_closingHandlers.push(terminator);
	}

	public void terminate() {
		Iterator<Tuple<?>> iterator = _relatedTuples.iterator();

		while (iterator.hasNext()) {
			Tuple<?> next = iterator.next();

			iterator.remove();

			try {
				next.terminate();
			}
			catch (Exception e) {
				e.printStackTrace();
			}
		}

		while (!_closingHandlers.isEmpty()) {
			Runnable pop = _closingHandlers.pop();

			try {
				pop.run();
			}
			catch (Exception e) {
			}
		}
	}

}

@@ -1286,16 +1286,74 @@
 ist%3C%3E();
+%0A%09private AtomicBoolean closed = new AtomicBoolean(false);
 %0A%0A%09priva
@@ -1526,39 +1526,12 @@
 lic 
-DoublyLinkedList.Node%3CTuple%3C?%3E%3E
+void
  add
@@ -1561,16 +1561,54 @@
 uple) %7B%0A
+%09%09if (closed.get()) %7B%0A%09%09%09return;%0A%09%09%7D%0A%0A
 %09%09Doubly
@@ -1726,29 +1726,8 @@
 ve);
-%0A%0A%09%09return tupleNode;
 %0A%09%7D%0A
@@ -1965,157 +1965,79 @@
 %7B%0A%09%09
-if (this == o) return true;%0A%09%09if (o == null %7C%7C getClass() != o.getClass()) return false;%0A%0A%09%09Tuple%3C?%3E tuple = (Tuple%3C?%3E) o;%0A%0A%09%09return t.equals(tuple.t
+return this == o;%0A%09%7D%0A%0A%09public boolean isClosed() %7B%0A%09%09return closed.get(
 );%0A%09
@@ -2212,16 +2212,54 @@
 ator) %7B%0A
+%09%09if (closed.get()) %7B%0A%09%09%09return;%0A%09%09%7D%0A%0A
 %09%09_closi
@@ -2318,16 +2318,76 @@
 ate() %7B%0A
+%09%09if (!closed.compareAndSet(false, true)) %7B%0A%09%09%09return;%0A%09%09%7D%0A%0A
 %09%09Iterat

Add closed status to the tuple