//  Copyright 2005 The Apache Software Foundation
   //
   // 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.jph.channels.router;
  
  import org.jph.channels.DeadLetterOutputChannel;
  import org.jph.channels.MessageCloner;
  import org.jph.channels.OutputChannel;
  import org.jph.collections.predicate.Predicate;
  import org.jph.collections.predicate.TruePredicate;
  import org.jph.concurrent.unmodifiable.UnmodifiableLists;
  
  import java.util.ArrayList;
  import java.util.Iterator;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  /***
   * De StdRoutingOutputChannel is the default implementation of the RoutingOutputChannel.
   * <p/>
   * this implementation is not complete:
   * problems:
   * <li>
   * <ul>what to do if stopAfterFirst is false and not all receivers accept
   * the message, but some did. Should the message be send to the failOutputChannel?
   * <li>at the moment the failOutputChannel is never used.
   * </li>
   *
   * @author Peter Veentjer.
   */
  public final class StdRoutingOutputChannel<M> implements RoutingOutputChannel<M> {
  
  	//the entrylist always is immutable.
  	private List<OutRouterEntry<M>> _entryList;
  	//the channel where message are send to that can`t be delivered.
  	private final OutputChannel<M> _failOutputChannel;
  	//stopAfterFirst indicated if messages should be tried with other channels if
  	//a channel has accepted the message.
  	private final boolean _stopAfterFirst;
  
  	private static <M> List<OutRouterEntry<M>> makeList(OutRouterEntry<M> entry) {
  		if (entry == null) throw new NullPointerException();
  		List<OutRouterEntry<M>> list = new ArrayList<OutRouterEntry<M>>();
  		list.add(entry);
  		return list;
  	}
  
  	public StdRoutingOutputChannel(OutRouterEntry<M> entry, boolean stopAfterFirst) {
  		this(makeList(entry), stopAfterFirst);
  	}
  
  	public StdRoutingOutputChannel(List<OutRouterEntry<M>> entryList, boolean stopAfterFirst) {
  		this(entryList, new DeadLetterOutputChannel<M>(), stopAfterFirst);
  	}
  
  
  	public StdRoutingOutputChannel(List<OutRouterEntry<M>> entryList, OutputChannel<M> failOutputChannel,
  								   boolean stopAfterFirst) {
  		if (entryList == null || failOutputChannel == null) throw new NullPointerException();
  		_stopAfterFirst = stopAfterFirst;
  		_entryList = entryList;
  		_failOutputChannel = failOutputChannel;
  	}
  
  
  	/***
  	 * Returns the OutputChannel where all messages are send to that can`t
  	 * be delivered.
  	 */
  	public OutputChannel<M> getFailOutputChannel() {
  		return _failOutputChannel;
  	}
  
  	public boolean isStopAfterFirst() {
  		return _stopAfterFirst;
  	}
  
  	public MessageCloner<M> getMessageCloner() {
  		return null;  //To change body of implemented methods use File | Settings | File Templates.
  	}
  
  	public void add(OutRouterEntry<M> entry) {
  		if (entry == null)
  			throw new NullPointerException("entry can`t be null");
  
  		synchronized (this) {
  			_entryList = UnmodifiableLists.add(_entryList, entry);
  		}
  	}
 
 	public void subscribe(OutputChannel<M> outputChannel) {
 		subscribe(null, outputChannel);
 	}
 
 	public void subscribe(Predicate predicate, OutputChannel<M> outputChannel) {
 		if (outputChannel == null) throw new NullPointerException();
 
 		OutRouterEntry<M> entryOut = new OutRouterEntry<M>(
 				outputChannel,
 				predicate == null ? TruePredicate.INSTANCE : predicate);
 		add(entryOut);
 	}
 
 	public Iterator<OutRouterEntry<M>> entries() {
 		return _entryList.iterator();
 	}
 
 	public void put(M msg) throws InterruptedException {
 		if (msg == null) throw new NullPointerException();
 
 		for (OutRouterEntry<M> entryOut : _entryList) {
 			if (entryOut.getPredicate().evaluate(msg)) {
 				entryOut.getChannel().put(msg);
 				if (_stopAfterFirst)
 					return;
 			}
 		}
 	}
 
 	public boolean offer(M msg, long timeout, TimeUnit unit) throws InterruptedException {
 		if (msg == null || unit == null) throw new NullPointerException();
 
 		long timeoutNs = unit.convert(timeout, TimeUnit.NANOSECONDS);
 		long startNs = System.nanoTime();
 
 		for (OutRouterEntry<M> entryOut : _entryList) {
 			boolean accepted = entryOut.getPredicate().evaluate(msg);
 			if (accepted) {
 
 				boolean offered = entryOut.getChannel().offer(msg, timeoutNs, TimeUnit.NANOSECONDS);
 				if (offered && _stopAfterFirst)
 					return true;
 
 				//decrease the remaining timeout
 				long endNs = System.nanoTime();
 				long elapsedNs = endNs - startNs;
 				timeoutNs -= -elapsedNs;
 				timeoutNs = timeoutNs < 0 ? 0 : timeoutNs;
 				startNs = endNs;
 			}
 		}
 
 		//todo
 		return false;
 	}
 }