/* * Copyright (c) 2003 - 2007 Israfil Consulting Services Corporation * Copyright (c) 2003 - 2007 Christian Edward Gruber * All Rights Reserved * * This software is licensed under the Berkeley Standard Distribution license, * (BSD license), as defined below: * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * 1. Redistributions of source code must retain the above copyright notice, this * list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * 3. Neither the name of Israfil Consulting Services nor the names of its contributors * may be used to endorse or promote products derived from this software without * specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. * IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, * OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY * OF SUCH DAMAGE. * * $Id: Pump.java 362 2007-01-21 06:37:33Z cgruber $ */ package net.israfil.foundation.concurrent; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.concurrent.TimeUnit; /** * A worker that repeats and draws items from a producer and feeds it to a consumer * * @author Christian Edward Gruber */ public class DistributingPump extends AbstractPump { protected ConcurrentProducer producer; protected Collection> consumers; private List> consumersArray = null; private int currentItem; public DistributingPump() {} public DistributingPump(String name, ConcurrentProducer producer, Collection> consumers, long timeout, TimeUnit timeoutUnit) { super(name,timeout,timeoutUnit); setProducer(producer); setConsumers(consumers); } @Override protected void validate() { super.validate(); if (getProducer() == null) throw new IllegalStateException("consumer cannot be null."); if (getConsumers() == null || getConsumers().size() == 0) throw new IllegalStateException("producers cannot be null."); } @Override public boolean continueProcessing() { return isRunning(); } @Override public long iterationDelay() { return 0; } public ConcurrentProducer getProducer() { return producer; } public void setProducer(ConcurrentProducer producer) { this.producer = producer; } public Collection> getConsumers() { return consumers; } public void setConsumers(Collection> consumers) { this.consumers = consumers; } @Override protected boolean init() throws InterruptedException { boolean init = super.init() && this.getProducer() != null && this.getConsumers() != null && this.getConsumers().size() > 0; if (!init) return init; // fail fast. consumersArray = new ArrayList>(consumers); // get consistent list. return init; } @Override protected void process() throws InterruptedException { ConcurrentConsumer consumer = consumersArray.get(currentItem++); if (currentItem >= consumersArray.size()) currentItem = 0; consumer.consume(producer.produce()); } }