/* * 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 CollectingPump extends AbstractPump { protected Collection> producers = null; protected ConcurrentConsumer consumer = null; private List> producersArray = null; private int currentItem; public CollectingPump() {} public CollectingPump(String name, Collection> producers, ConcurrentConsumer consumer, long timeout, TimeUnit timeoutUnit) { super(name,timeout,timeoutUnit); setProducers(producers); setConsumer(consumer); } @Override protected void validate() { super.validate(); if (getConsumer() == null) throw new IllegalStateException("consumer cannot be null."); if (getProducers() == null || getProducers().size() == 0) throw new IllegalStateException("producers cannot be null."); } @Override public boolean continueProcessing() { return isRunning(); } @Override public long iterationDelay() { return 0; } public Collection> getProducers() { return producers; } public void setProducers(Collection> producers) { if (this.isRunning()) throw new IllegalStateException("Cannot set producers while this pump is not stopped."); this.producers = producers; } public ConcurrentConsumer getConsumer() { return consumer; } public void setConsumer(ConcurrentConsumer consumer) { if (!this.isStopped()) throw new IllegalStateException("Cannot set consumer while this pump is not stopped."); this.consumer = consumer; } @Override protected boolean init() throws InterruptedException { boolean init = super.init() && this.getConsumer() != null && this.getProducers() != null && this.getProducers().size() > 0; if (!init) return init; // fail fast. producersArray = new ArrayList>(producers); // get consistent list. return init; } @Override protected void process() throws InterruptedException { ConcurrentProducer producer = producersArray.get(currentItem++); if (currentItem >= producersArray.size()) currentItem = 0; consumer.consume(producer.produce()); } }