// Path: camel-milo/src/main/java/org/apache/camel/component/milo/Messages.java
// public final class Messages {
// 	private Messages() {
// 	}
// 
// 	/**
// 	 * Fill a Message from a DataValue
// 	 *
// 	 * @param value
// 	 *            the value to feed from
// 	 * @param result
// 	 *            the result to feed to
// 	 */
// 	public static void fillFromDataValue(final DataValue value, final DefaultMessage result) {
// 		result.setBody(value);
// 		result.setFault(value.getStatusCode().isBad());
// 	}
// }
// 
// Path: camel-milo/src/main/java/org/apache/camel/component/milo/server/internal/CamelServerItem.java
// public class CamelServerItem {
// 	private static final Logger LOG = LoggerFactory.getLogger(CamelServerItem.class);
// 
// 	private UaObjectNode baseNode;
// 	private UaVariableNode item;
// 
// 	private DataValue value = new DataValue(StatusCode.BAD);
// 	private final Set<Consumer<DataValue>> listeners = new CopyOnWriteArraySet<>();
// 
// 	public CamelServerItem(final String itemId, final ServerNodeMap nodeManager, final UShort namespaceIndex,
// 			final UaObjectNode baseNode) {
// 
// 		this.baseNode = baseNode;
// 
// 		final NodeId nodeId = new NodeId(namespaceIndex, "items-" + itemId);
// 		final QualifiedName qname = new QualifiedName(namespaceIndex, itemId);
// 		final LocalizedText displayName = LocalizedText.english(itemId);
// 
// 		// create variable node
// 
// 		this.item = new UaVariableNode(nodeManager, nodeId, qname, displayName) {
// 
// 			@Override
// 			public DataValue getValue() {
// 				return getDataValue();
// 			}
// 
// 			@Override
// 			public synchronized void setValue(final DataValue value) {
// 				setDataValue(value);
// 			}
// 
// 		};
// 
// 		// item.setDataType();
// 		this.item.setAccessLevel(ubyte(AccessLevel.getMask(AccessLevel.READ_WRITE)));
// 		this.item.setUserAccessLevel(ubyte(AccessLevel.getMask(AccessLevel.READ_WRITE)));
// 
// 		baseNode.addComponent(this.item);
// 	}
// 
// 	public void dispose() {
// 		this.baseNode.removeComponent(this.item);
// 		this.listeners.clear();
// 	}
// 
// 	public void addWriteListener(final Consumer<DataValue> consumer) {
// 		this.listeners.add(consumer);
// 	}
// 
// 	public void removeWriteListener(final Consumer<DataValue> consumer) {
// 		this.listeners.remove(consumer);
// 	}
// 
// 	protected void setDataValue(final DataValue value) {
// 		LOG.debug("setValue -> {}", value);
// 		runThrough(this.listeners, c -> c.accept(value));
// 	}
// 
// 	/**
// 	 * Run through a list, aggregating errors
// 	 *
// 	 * <p>
// 	 * The consumer is called for each list item, regardless if the consumer did
// 	 * through an exception. All exceptions are caught and thrown in one
// 	 * RuntimeException. The first exception being wrapped directly while the
// 	 * latter ones, if any, are added as suppressed exceptions.
// 	 * </p>
// 	 *
// 	 * @param list
// 	 *            the list to run through
// 	 * @param consumer
// 	 *            the consumer processing list elements
// 	 */
// 	protected <T> void runThrough(final Collection<Consumer<T>> list, final Consumer<Consumer<T>> consumer) {
// 		LinkedList<Throwable> errors = null;
// 
// 		for (final Consumer<T> listener : list) {
// 			try {
// 				consumer.accept(listener);
// 			} catch (final Throwable e) {
// 				if (errors == null) {
// 					errors = new LinkedList<>();
// 				}
// 				errors.add(e);
// 			}
// 		}
// 
// 		if (errors == null || errors.isEmpty()) {
// 			return;
// 		}
// 
// 		final RuntimeException ex = new RuntimeException(errors.pollFirst());
// 		errors.forEach(ex::addSuppressed);
// 		throw ex;
// 	}
// 
// 	protected DataValue getDataValue() {
// 		return this.value;
// 	}
// 
// 	public void update(final Object value) {
// 		this.value = new DataValue(new Variant(value), StatusCode.GOOD, DateTime.now());
// 	}
// }


import java.util.function.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.milo.Messages;
import org.apache.camel.component.milo.server.internal.CamelServerItem;
import org.apache.camel.impl.DefaultConsumer;
import org.apache.camel.impl.DefaultMessage;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
/*
 * Copyright (C) 2016 Jens Reimann <jreimann@redhat.com>
 *
 * 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 org.apache.camel.component.milo.server;



class MiloServerConsumer extends DefaultConsumer {

	private final CamelServerItem item;
	private final Consumer<DataValue> writeHandler = this::performWrite;

	public MiloServerConsumer(final Endpoint endpoint, final Processor processor, final CamelServerItem item) {
		super(endpoint, processor);
		this.item = item;
	}

	@Override
	protected void doStart() throws Exception {
		super.doStart();

		this.item.addWriteListener(this.writeHandler);
	}

	@Override
	protected void doStop() throws Exception {
		this.item.removeWriteListener(this.writeHandler);

		super.doStop();
	}

// Path: camel-milo/src/main/java/org/apache/camel/component/milo/Messages.java
// public final class Messages {
// 	private Messages() {
// 	}
// 
// 	/**
// 	 * Fill a Message from a DataValue
// 	 *
// 	 * @param value
// 	 *            the value to feed from
// 	 * @param result
// 	 *            the result to feed to
// 	 */
// 	public static void fillFromDataValue(final DataValue value, final DefaultMessage result) {
// 		result.setBody(value);
// 		result.setFault(value.getStatusCode().isBad());
// 	}
// }
// 
// Path: camel-milo/src/main/java/org/apache/camel/component/milo/server/internal/CamelServerItem.java
// public class CamelServerItem {
// 	private static final Logger LOG = LoggerFactory.getLogger(CamelServerItem.class);
// 
// 	private UaObjectNode baseNode;
// 	private UaVariableNode item;
// 
// 	private DataValue value = new DataValue(StatusCode.BAD);
// 	private final Set<Consumer<DataValue>> listeners = new CopyOnWriteArraySet<>();
// 
// 	public CamelServerItem(final String itemId, final ServerNodeMap nodeManager, final UShort namespaceIndex,
// 			final UaObjectNode baseNode) {
// 
// 		this.baseNode = baseNode;
// 
// 		final NodeId nodeId = new NodeId(namespaceIndex, "items-" + itemId);
// 		final QualifiedName qname = new QualifiedName(namespaceIndex, itemId);
// 		final LocalizedText displayName = LocalizedText.english(itemId);
// 
// 		// create variable node
// 
// 		this.item = new UaVariableNode(nodeManager, nodeId, qname, displayName) {
// 
// 			@Override
// 			public DataValue getValue() {
// 				return getDataValue();
// 			}
// 
// 			@Override
// 			public synchronized void setValue(final DataValue value) {
// 				setDataValue(value);
// 			}
// 
// 		};
// 
// 		// item.setDataType();
// 		this.item.setAccessLevel(ubyte(AccessLevel.getMask(AccessLevel.READ_WRITE)));
// 		this.item.setUserAccessLevel(ubyte(AccessLevel.getMask(AccessLevel.READ_WRITE)));
// 
// 		baseNode.addComponent(this.item);
// 	}
// 
// 	public void dispose() {
// 		this.baseNode.removeComponent(this.item);
// 		this.listeners.clear();
// 	}
// 
// 	public void addWriteListener(final Consumer<DataValue> consumer) {
// 		this.listeners.add(consumer);
// 	}
// 
// 	public void removeWriteListener(final Consumer<DataValue> consumer) {
// 		this.listeners.remove(consumer);
// 	}
// 
// 	protected void setDataValue(final DataValue value) {
// 		LOG.debug("setValue -> {}", value);
// 		runThrough(this.listeners, c -> c.accept(value));
// 	}
// 
// 	/**
// 	 * Run through a list, aggregating errors
// 	 *
// 	 * <p>
// 	 * The consumer is called for each list item, regardless if the consumer did
// 	 * through an exception. All exceptions are caught and thrown in one
// 	 * RuntimeException. The first exception being wrapped directly while the
// 	 * latter ones, if any, are added as suppressed exceptions.
// 	 * </p>
// 	 *
// 	 * @param list
// 	 *            the list to run through
// 	 * @param consumer
// 	 *            the consumer processing list elements
// 	 */
// 	protected <T> void runThrough(final Collection<Consumer<T>> list, final Consumer<Consumer<T>> consumer) {
// 		LinkedList<Throwable> errors = null;
// 
// 		for (final Consumer<T> listener : list) {
// 			try {
// 				consumer.accept(listener);
// 			} catch (final Throwable e) {
// 				if (errors == null) {
// 					errors = new LinkedList<>();
// 				}
// 				errors.add(e);
// 			}
// 		}
// 
// 		if (errors == null || errors.isEmpty()) {
// 			return;
// 		}
// 
// 		final RuntimeException ex = new RuntimeException(errors.pollFirst());
// 		errors.forEach(ex::addSuppressed);
// 		throw ex;
// 	}
// 
// 	protected DataValue getDataValue() {
// 		return this.value;
// 	}
// 
// 	public void update(final Object value) {
// 		this.value = new DataValue(new Variant(value), StatusCode.GOOD, DateTime.now());
// 	}
// }


// Path: camel-milo/src/main/java/org/apache/camel/component/milo/server/MiloServerConsumer.java
import java.util.function.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.milo.Messages;
import org.apache.camel.component.milo.server.internal.CamelServerItem;
import org.apache.camel.impl.DefaultConsumer;
import org.apache.camel.impl.DefaultMessage;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;

/*
 * Copyright (C) 2016 Jens Reimann <jreimann@redhat.com>
 *
 * 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 org.apache.camel.component.milo.server;



class MiloServerConsumer extends DefaultConsumer {

	private final CamelServerItem item;
	private final Consumer<DataValue> writeHandler = this::performWrite;

	public MiloServerConsumer(final Endpoint endpoint, final Processor processor, final CamelServerItem item) {
		super(endpoint, processor);
		this.item = item;
	}

	@Override
	protected void doStart() throws Exception {
		super.doStart();

		this.item.addWriteListener(this.writeHandler);
	}

	@Override
	protected void doStop() throws Exception {
		this.item.removeWriteListener(this.writeHandler);

		super.doStop();
	}

	protected void performWrite(final DataValue value) {