jicarilla-sandbox/platform/plumbing/impl/src/java/org/jicarilla/plumbing BarrierProcessor.java,NONE,1.1 Connector.java,NONE,1.1 DefaultStage.java,NONE,1.1 NoopSink.java,NONE,1.1 PostProcessor.java,NONE,1.1 PreProcessor.java,NONE,1.1 SimpleAlternator.java,NONE,1.1 SimpleCollector.java,NONE,1.1 SimpleMulticaster.java,NONE,1.1 SimpleScreener.java,NONE,1.1 SimpleSink.java,NONE,1.1 SimpleSource.java,NONE,1.1 SimpleStage.java,NONE,1.1
Leo Simons <[email protected]>
| Newsgroups | gmane.comp.java.jicarilla.cvs |
|---|---|
| Message-ID | <[email protected]> |
Update of /cvsroot/jicarilla/jicarilla-sandbox/platform/plumbing/impl/src/java/org/jicarilla/plumbing
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv30323/plumbing/impl/src/java/org/jicarilla/plumbing
Added Files:
BarrierProcessor.java Connector.java DefaultStage.java
NoopSink.java PostProcessor.java PreProcessor.java
SimpleAlternator.java SimpleCollector.java
SimpleMulticaster.java SimpleScreener.java SimpleSink.java
SimpleSource.java SimpleStage.java
Log Message:
s/o.j.f.p/o.j.p/
--- NEW FILE: SimpleSource.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Takable;
import org.jicarilla.plumbing.Source;
import org.jicarilla.framework.Assert;
/**
* A basic {@link Source} implementation backed by a {@link Takable}.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleSource.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleSource implements Source
{
/** the backing {@link Takable}. */
protected final Takable m_t;
/**
* Create a new instance backed by the provided <code>Takable</code>.
*
* @param t the takable to use as the backing implementation
*/
public SimpleSource( final Takable t )
{
Assert.assertNotNull( t );
m_t = t;
}
/**
* See {@link org.jicarilla.plumbing.Stage#take()}.
*
* @return the message retrieved
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object take() throws InterruptedException
{
return m_t.take();
}
/**
* See {@link org.jicarilla.plumbing.Stage#poll(long)}.
*
* @param l how long to try and fetch before giving up
* @return the message retrieved, or null if none was retrieved within
* the specified time interval
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object poll( final long l ) throws InterruptedException
{
return m_t.poll( l );
}
}
--- NEW FILE: NoopSink.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import org.jicarilla.plumbing.Sink;
/**
* <p>A sink which simply discards any message you send to it (similar to
* <code>/dev/null</code> on unix systems). This can be a useful end to
* a pipeline where all messages have already been handled and are of no
* further use, for example.</p>
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: NoopSink.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class NoopSink implements Sink
{
/**
* Discards a message.
*
* @param o the message that is discarded
*/
public void put( final Object o ) {}
/**
* Discard a message.
*
* @param o the message that is discarded
* @param l an ignored parameter specifying how long we should take in
* discarding the message
* @return true to indicate the message was discarded.
*/
public boolean offer( final Object o, final long l )
{
return true;
}
}
--- NEW FILE: SimpleStage.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Channel;
import org.jicarilla.plumbing.Stage;
import org.jicarilla.framework.Assert;
/**
* A basic implementation of a {@link org.jicarilla.plumbing.Stage} backed by a {@link Channel}.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleStage.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleStage implements Stage
{
/** The backing {@link Channel}. */
protected Channel m_channel;
/**
* Create a new instance backed by the provided channel.
*
* @param channel the channel to use as backend
*/
public SimpleStage( final Channel channel )
{
Assert.assertNotNull( channel );
m_channel = channel;
}
/**
* See {@link Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
getChannel().put( o );
}
/**
* See {@link Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l ) throws InterruptedException
{
return getChannel().offer( o, l );
}
/**
* See {@link Stage#take()}.
*
* @return the message retrieved
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object take() throws InterruptedException
{
return getChannel().take();
}
/**
* See {@link org.jicarilla.plumbing.Stage#poll(long)}.
*
* @param l how long to try and fetch before giving up
* @return the message retrieved, or null if none was retrieved within
* the specified time interval
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object poll( final long l ) throws InterruptedException
{
return getChannel().poll( l );
}
/**
* See {@link org.jicarilla.plumbing.Stage#peek()}.
*
* @return the message retrieved
*/
public Object peek()
{
return getChannel().peek();
}
/**
* Get the backing channel.
*
* @return the backing channel
*/
protected Channel getChannel()
{
return m_channel;
}
}
--- NEW FILE: SimpleMulticaster.java ---
/* ====================================================================
* The Apache Software License, Version 1.1
*
* Copyright (c) 1997-2003 The Apache Software Foundation. All rights
* reserved.
*
* 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. The end-user documentation included with the redistribution,
* if any, must include the following acknowledgment:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowledgment may appear in the software
* itself, if and wherever such third-party acknowledgments
* normally appear.
*
* 4. The names "Avalon", and "Apache Software Foundation"
* must not be used to endorse or promote products derived from this
* software without prior written permission. For written
* permission, please contact [email protected].
*
* 5. Products derived from this software may not be called "Apache",
* nor may "Apache" appear in their name, without prior written
* permission of the Apache Software Foundation.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED 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 APACHE SOFTWARE FOUNDATION OR
* ITS 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*/
package org.jicarilla.plumbing;
import org.jicarilla.plumbing.Multicaster;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.framework.Assert;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;
/**
* A {@link Multicaster} which clones {@link Cloneable} messages before sending
* them to all its sinks, but makes no attempt to do so for objects that do not
* implement Cloneable.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleMulticaster.java,v 1.4 2003/11/18 16:09:11 lsimons Exp
* $
*/
public class SimpleMulticaster implements Multicaster
{
/** the referenced sinks. */
protected final List m_sinks;
/** synchronization point for {@link m_sinks}. */
protected final Object m_mutex = new Object();
/**
* Create a new instance backed by an {@link ArrayList}.
*/
public SimpleMulticaster()
{
m_sinks = new ArrayList();
}
/**
* Add a sink.
*
* @param sink the sink to add
*/
public void addSink( final Sink sink )
{
Assert.assertNotNull( sink );
synchronized( m_mutex )
{
m_sinks.add( sink );
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
synchronized( m_mutex )
{
for( int i = 0; i < m_sinks.size(); i++ )
{
final Object puttableObject = tryClone( o );
((Sink)m_sinks.get( i )).put( puttableObject );
}
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l ) throws InterruptedException
{
boolean accept = false;
synchronized( m_mutex )
{
for( int i = 0; i < m_sinks.size(); i++ )
{
final Object puttableObject = tryClone( o );
final boolean accepted = ((Sink)m_sinks.get( i )).offer(
puttableObject, l );
accept = (accept || accepted);
}
}
return accept;
}
/**
* Try and create a clone of an object. If the object is not {@link
* Cloneable} or if the cloning fails, don't clone.
*
* @param o the object to clone
* @return the new clone, or the object itself if the object wasn't cloned
*/
protected Object tryClone( final Object o )
{
if( o instanceof Cloneable )
{
try
{
final Object clone = o.getClass().getMethod( "clone", new Class[0] )
.invoke( o, new Object[0] );
return clone;
}
catch( NoSuchMethodException nsme ) {}
catch( IllegalAccessException iae ) {}
catch( InvocationTargetException ite ) {}
}
return o;
}
}
--- NEW FILE: PreProcessor.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Channel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.plumbing.Stage;
import org.jicarilla.plumbing.DefaultStage;
import org.jicarilla.framework.Assert;
/**
* A Preprocessor runs a series of plumbing over a message before enqueueing it:
*
* <pre>
* +-----------------------------------------+
* |Preprocessor |
* put() |+------+ +------+ +------+ +---------+ take()
* ------>|Stage1|-->|Stage2|-->|Stage3|-->| Channel |----->
* |+------+ +------+ +------+ +---------+
* +-----------------------------------------+
* </pre>
*
* Use a Preprocessor when it does not make sense to put any kind of
* (asynchronous) buffering between plumbing and it is acceptable for put() to
* take a long time.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: PreProcessor.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class PreProcessor extends DefaultStage
{
/** The stages to run before returning from {@link #put(Object)}. */
protected final List m_stages;
/**
* The mutex to synchronize on for operations involving
* {@link m_stages}.
*/
protected final Object m_mutex = new Object();
/**
* Create a new instance.
*
* @param channel the channel to pass to {@link DefaultStage}
* @param errorHandler the error handler to pass to {@link DefaultStage}
*/
public PreProcessor( final Channel channel, final Sink errorHandler )
{
super( channel, errorHandler );
m_stages = new ArrayList();
}
/**
* Add a stage to the processor. It's appended at the end of the current
* set of stages.
*
* @param stage the stage to add
*/
public void addStage( final Stage stage )
{
Assert.assertNotNull( stage );
synchronized( m_mutex )
{
m_stages.add( stage );
}
}
/**
* See {@link Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
process( o );
super.put( o );
}
/**
* See {@link org.jicarilla.plumbing.Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l )
throws InterruptedException
{
process( o );
return super.offer( o, l );
}
/**
* Runs the message through the set of post processing stages.
*
* @param o the message to process
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
protected void process( final Object o ) throws InterruptedException
{
synchronized( m_mutex )
{
Object ob = o;
final Iterator it = m_stages.iterator();
while( it.hasNext() )
{
final Stage stage = (Stage)it.next();
stage.put( ob );
ob = stage.take();
}
}
}
}
--- NEW FILE: SimpleScreener.java ---
/* ====================================================================
* The Apache Software License, Version 1.1
*
* Copyright (c) 1997-2003 The Apache Software Foundation. All rights
* reserved.
*
* 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. The end-user documentation included with the redistribution,
* if any, must include the following acknowledgment:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowledgment may appear in the software
* itself, if and wherever such third-party acknowledgments
* normally appear.
*
* 4. The names "Avalon", and "Apache Software Foundation"
* must not be used to endorse or promote products derived from this
* software without prior written permission. For written
* permission, please contact [email protected].
*
* 5. Products derived from this software may not be called "Apache",
* nor may "Apache" appear in their name, without prior written
* permission of the Apache Software Foundation.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED 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 APACHE SOFTWARE FOUNDATION OR
* ITS 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*/
package org.jicarilla.plumbing;
import org.jicarilla.framework.Assert;
import org.jicarilla.framework.Selector;
import org.jicarilla.framework.SelectorSwitch;
import org.jicarilla.framework.Switch;
import java.util.Iterator;
import java.util.List;
/**
* A screener which tries sinks in the order they were added, or falls back on
* the sink that was added last when none of the sinks will accept a message.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleScreener.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleScreener implements Screener
{
protected final Switch m_switch;
/** synchronization point for {@link m_switch}. */
protected final Object m_mutex = new Object();
/**
* Create a new instance backed by a {@link SelectorSwitch}.
*/
public SimpleScreener()
{
m_switch = new SelectorSwitch();
}
/**
* Add a sink.
*
* @param sink the sink to add
* @param selector the selector that determines whether this sink will
* receive a particular message
*/
public void addSink( final Selector selector, final Sink sink )
{
Assert.assertNotNull( selector );
Assert.assertNotNull( sink );
synchronized( m_mutex )
{
m_switch.put( selector, sink );
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
final Sink sink = selectSink( o );
sink.put( o );
}
/**
* See {@link org.jicarilla.plumbing.Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l )
throws InterruptedException
{
final Sink sink = selectSink( o );
return sink.offer( o, l );
}
/**
* Select a sink from all referenced sinks based on the provided selection
* criterium.
*
* @param o the selection criterium
* @return the selected sink
* @throws IllegalStateException if no sink can be found for this
* message
*/
protected Sink selectSink( final Object o )
{
synchronized( m_mutex )
{
Iterator it = m_switch.keySet().iterator();
while( it.hasNext() )
{
Selector selector = (Selector)it.next();
final boolean selected = selector.select( o );
if( selected )
return (Sink)m_switch.get( selector );
}
List switchList = m_switch.entryList();
Switch.Entry entry =
(Switch.Entry)switchList.get( m_switch.size()-1 );
return (Sink)entry.getValue();
}
}
}
--- NEW FILE: DefaultStage.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Channel;
import org.jicarilla.plumbing.Sink;
/**
* A basic {@link org.jicarilla.plumbing.Stage} implementation which allows for the redirection of
* exceptions to a special handler. In a diagram:
*
* <pre>
*
* <<flow in case of exception>>
* ^
* | ErrorHandler
* | ^
* | /-------------|-------------\
* | | | |
* | Stage1 -----> Stage2 -----> Stage3 -->
* |
* ---------------------------------------->
* <<normal program flow>>
* </pre>
*
* this is a common pattern which allows you to keep your regular pipelining
* free of constant checks for a consistent state.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: DefaultStage.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class DefaultStage extends SimpleStage
{
/** The errorhandler to send errors to. */
protected final Sink m_errorHandler;
/**
* Create a new instance.
*
* @param channel the channel to pass on to {@link SimpleStage}
* @param errorHandler the recipient of all errors
*/
public DefaultStage( final Channel channel,
final Sink errorHandler )
{
super( channel );
m_errorHandler = errorHandler;
}
/**
* Sends an error to the error handler, if it exists. If no handler exists,
* nothing happens.
*
* @param o the error to redirect
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
protected void handleError( final Object o ) throws InterruptedException
{
if( m_errorHandler != null )
{
m_errorHandler.put( o );
}
}
}
--- NEW FILE: PostProcessor.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Channel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.plumbing.Stage;
import org.jicarilla.plumbing.DefaultStage;
import org.jicarilla.framework.Assert;
/**
* <p>A Postprocessor runs a message through several stages <i>before</i>
* dequeueing it. In a diagram:</p>
*
* <pre>
* +-------------------------------------------+
* |Postprocessor |
* put() |+---------+ +------+ +------+ +------+ take()
* ------>|Channel |-->|Stage1|--->|Stage2|-->|Stage3|----->
* |+---------+ +------+ +------+ +------+
* +-------------------------------------------+
* </pre>
*
* <p>Use a Postprocessor when it does not make sense to put any kind of
* (asynchronous) buffering between plumbing and it is acceptable for
* {@link #take()} to take a long time (no pun intended).</p>
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: PostProcessor.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class PostProcessor extends DefaultStage
{
/** The stages to run before returning from {@link #take()}. */
protected final List m_stages;
/**
* The mutex to synchronize on for operations involving
* {@link m_stages}.
*/
protected final Object m_mutex = new Object();
/**
* Create a new instance.
*
* @param channel the channel to pass to {@link DefaultStage}
* @param errorHandler the error handler to pass to {@link org.jicarilla.plumbing.DefaultStage}
*/
public PostProcessor( final Channel channel, final Sink errorHandler )
{
super( channel, errorHandler );
m_stages = new ArrayList();
}
/**
* Add a stage to the processor. It's appended at the end of the current
* set of stages.
*
* @param stage the stage to add
*/
public void addStage( final Stage stage )
{
Assert.assertNotNull( stage );
synchronized( m_mutex )
{
m_stages.add( stage );
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#take()}.
*
* @return the message retrieved
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object take() throws InterruptedException
{
final Object o = super.take();
process( o );
return o;
}
/**
* See {@link org.jicarilla.plumbing.Stage#poll(long)}.
*
* @param pollingTime how long to try and fetch before giving up
* @return the message retrieved, or null if none was retrieved within
* the specified time interval
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object poll( final long pollingTime ) throws InterruptedException
{
final Object o = super.poll( pollingTime );
process( o );
return o;
}
/**
* Runs the message through the set of post processing stages.
*
* @param o the message to process
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
protected void process( final Object o ) throws InterruptedException
{
synchronized( m_mutex )
{
Object ob = o;
final Iterator it = m_stages.iterator();
while( it.hasNext() )
{
final Stage stage = (Stage)it.next();
stage.put( ob );
ob = stage.take();
}
}
}
}
--- NEW FILE: SimpleAlternator.java ---
/* ====================================================================
* The Apache Software License, Version 1.1
*
* Copyright (c) 1997-2003 The Apache Software Foundation. All rights
* reserved.
*
* 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. The end-user documentation included with the redistribution,
* if any, must include the following acknowledgment:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowledgment may appear in the software
* itself, if and wherever such third-party acknowledgments
* normally appear.
*
* 4. The names "Avalon", and "Apache Software Foundation"
* must not be used to endorse or promote products derived from this
* software without prior written permission. For written
* permission, please contact [email protected].
*
* 5. Products derived from this software may not be called "Apache",
* nor may "Apache" appear in their name, without prior written
* permission of the Apache Software Foundation.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED 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 APACHE SOFTWARE FOUNDATION OR
* ITS 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*/
package org.jicarilla.plumbing;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.plumbing.Alternator;
import org.jicarilla.framework.Assert;
import java.util.ArrayList;
import java.util.List;
/**
* A list-backed {@link org.jicarilla.plumbing.Alternator} which uses the priority as a weight. In
* other words, a sink with priority <i>2*n</i> has twice the chance of being
* selected compared to a sink with priority <i>n</i>).
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleAlternator.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleAlternator implements Alternator
{
/** the sinks to alternate through. */
protected final List m_sinks;
/** synchronization point for {@link m_sinks}. */
protected final Object m_mutex = new Object();
/**
* Create a new instance backed by an {@link ArrayList}.
*/
public SimpleAlternator()
{
m_sinks = new ArrayList();
}
/**
* Add a sink with the default priority.
*
* @param sink the sink to add
*/
public void addSink( final Sink sink )
{
addSink( sink, DEFAULT_PRIORITY );
}
/**
* Add a sink with the specified priority.
*
* @param sink the sink to add
* @param priority the weight of the sink in determining how often
* it receives a message
*/
public void addSink( final Sink sink, final int priority )
{
Assert.assertNotNull( sink );
Assert.assertTrue( priority >= MIN_PRIORITY );
Assert.assertTrue( priority <= MAX_PRIORITY );
synchronized( m_mutex )
{
for( int i = 0; i < priority; i++ )
{
m_sinks.add( sink );
}
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
final Sink sink = randomSink();
sink.put( o );
}
/**
* See {@link org.jicarilla.plumbing.Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l ) throws InterruptedException
{
final Sink sink = randomSink();
return sink.offer( o, l );
}
/**
* Select a random sink from all referenced sinks.
*
* @return the selected sink
*/
protected Sink randomSink()
{
synchronized( m_mutex )
{
final int selected = (int)(Math.random() * m_sinks.size());
return (Sink)m_sinks.get( selected );
}
}
}
--- NEW FILE: BarrierProcessor.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Channel;
/**
* <p>A Barrier Processor runs am enqueued message through
* several stages, <i>in its own thread</i>, before dequeueing it.
* In a diagram:</p>
*
* <pre>
* +-----------------------------------------------+
* |Barrierprocessor +------------+ |
* | ...................| Worker |... |
* | . . +------------+ . |
* | . . . . |
* put() |+---------+ +------+ +------+ +---------+ | take()
* ------>|Channel |->|Stage1|->|Stage2|->|Channel |------->
* |+---------+ +------+ +------+ +---------+ |
* +-----------------------------------------------+
* </pre>
*
* <p>Use a barrier processor if you need to shield stages that are
* not thread safe from receiving messages from multiple concurrent
* threads. Note a barrier processor may create a big bottleneck in
* your system. If this happens, you should consider redesigning
* your application so that you can either use multiple
* singlethreaded stages of the same type, or change those stages to
* be thread-safe. The former is usually the more performant option,
* but the latter may be easier (by using synchronization in the
* appropriate places in your code).</p>
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: BarrierProcessor.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class BarrierProcessor extends DefaultStage
{
public BarrierProcessor( Channel channel, Sink errorHandler )
{
super( channel, errorHandler );
}
}
--- NEW FILE: SimpleSink.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Puttable;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.framework.Assert;
/**
* A basic {@link Sink} implementation backed by a {@link Puttable}.
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleSink.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleSink implements Sink
{
/** the backing {@link Puttable}. */
protected final Puttable m_p;
/**
* Create a new instance based on the provided <code>Puttable</code>.
*
* @param p the <code>Puttable</code> to use as the backing implementation
*/
public SimpleSink( final Puttable p )
{
Assert.assertNotNull( p );
m_p = p;
}
/**
* See {@link org.jicarilla.plumbing.Stage#put(Object)}.
*
* @param o the object to add
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public void put( final Object o ) throws InterruptedException
{
m_p.put( o );
}
/**
* See {@link org.jicarilla.plumbing.Stage#offer(Object,long)}.
*
* @param o the object to add
* @param l how long to try adding the object before returning
* @return true if the object was added, false if it wasn't
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public boolean offer( final Object o, final long l ) throws InterruptedException
{
return m_p.offer( o, l );
}
}
--- NEW FILE: Connector.java ---
/* ====================================================================
The Jicarilla Software License
Copyright (c) 2003 Leo Simons.
All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
==================================================================== */
package org.jicarilla.plumbing;
import EDU.oswego.cs.dl.util.concurrent.Executor;
import org.jicarilla.plumbing.Sink;
import org.jicarilla.plumbing.Source;
import org.jicarilla.framework.AbstractExecutable;
import org.jicarilla.framework.Assert;
/**
* <p>Sets up a connection between a source and a sink that will do nothing but
* continuously move messages from source to sink. In other words, use a
* <code>Connector</code> to creates a zero-length pipe/channel between a
* source and a sink. This movement is done by a supplied {@link Executor},
* so will happen asynchronously.</p>
*
* <p>Note that this class is {@link org.jicarilla.framework.Active}, and extends from
* {@link org.jicarilla.framework.AbstractExecutable}. As such, you should not override the
* {@link #initialize()} nor the {@link #dispose()} methods, but only ever
* override {@link #doInitialize()} and/or {@link #doDispose()}. When you do,
* make sure to call the overridden methods as well.</p>
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: Connector.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class Connector extends AbstractExecutable
{
/** The source to retrieve messages from. */
protected final Source m_source;
/** The sink to send messages to. */
protected final Sink m_sink;
/**
* Create a new instance.
*
* @param source the source to retrieve messages from
* @param sink the sink to send messages to
* @param executor the executor to pass on to {@link AbstractExecutable}
*/
public Connector( final Source source, final Sink sink, final Executor executor )
{
super( executor );
Assert.assertNotNull( source );
Assert.assertNotNull( sink );
m_source = source;
m_sink = sink;
}
/**
* Get a message from the source and send it to the sink.
*
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
protected void work() throws InterruptedException
{
final Object msg = m_source.take();
m_sink.put( msg );
}
}
--- NEW FILE: SimpleCollector.java ---
/* ====================================================================
* The Apache Software License, Version 1.1
*
* Copyright (c) 1997-2003 The Apache Software Foundation. All rights
* reserved.
*
* 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. The end-user documentation included with the redistribution,
* if any, must include the following acknowledgment:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowledgment may appear in the software
* itself, if and wherever such third-party acknowledgments
* normally appear.
*
* 4. The names "Avalon", and "Apache Software Foundation"
* must not be used to endorse or promote products derived from this
* software without prior written permission. For written
* permission, please contact [email protected].
*
* 5. Products derived from this software may not be called "Apache",
* nor may "Apache" appear in their name, without prior written
* permission of the Apache Software Foundation.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED 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 APACHE SOFTWARE FOUNDATION OR
* ITS 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*/
package org.jicarilla.plumbing;
import org.jicarilla.plumbing.Collector;
import org.jicarilla.plumbing.Source;
import org.jicarilla.framework.Assert;
import java.util.ArrayList;
import java.util.List;
/**
* A collector which tries to gather messages from a random subset of its
* sources (the subset is of fixed size though).
*
* @author <a href="lsimons at jicarilla dot org">Leo Simons</a>
* @version $Id: SimpleCollector.java,v 1.1 2004/03/22 21:20:13 lsimons Exp $
*/
public class SimpleCollector implements Collector
{
/** how long to poll a source when doing a {@link #take()}. */
public final static int POLLING_TIME = 20;
/** how often to retry polling when doing a {@link #take()}. */
public final static int RETRIES = 5;
/**
* what to split the polling time by when doing a {@link #poll(long)}.
*/
public final static long POLLING_TIME_DIVIDE = 5;
/** the sources to collect from. */
protected final List m_sources;
/** synchronization point for {@link m_sources}. */
protected final Object m_mutex = new Object();
/**
* Create a new instance backed by an {@link ArrayList}.
*/
public SimpleCollector()
{
m_sources = new ArrayList();
}
/**
* Add a source to the processor.
*
* @param source the source to add
*/
public void addSource( final Source source )
{
Assert.assertNotNull( source );
synchronized( m_mutex )
{
m_sources.add( source );
}
}
/**
* See {@link org.jicarilla.plumbing.Stage#take()}.
*
* @return the message retrieved
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object take() throws InterruptedException
{
for( int i = 0; i < RETRIES; i++ )
{
final Source source = selectSource();
final Object result = source.poll( POLLING_TIME );
if( result != null )
{
return result;
}
}
final Source source = selectSource();
return source.take();
}
/**
* See {@link org.jicarilla.plumbing.Stage#poll(long)}.
*
* @param time how long to try and fetch before giving up
* @return the message retrieved, or null if none was retrieved within
* the specified time interval
* @throws InterruptedException if the current thread has been
* {@link java.lang.Thread#interrupt()}ed
*/
public Object poll( final long time ) throws InterruptedException
{
final long pollingTime = (time / POLLING_TIME_DIVIDE) + 1;
for( int i = 0; i < POLLING_TIME_DIVIDE; i++ )
{
final Source source = selectSource();
final Object result = source.poll( pollingTime );
if( result != null )
{
return result;
}
}
return null;
}
/**
* Select a random source from all referenced sources.
*
* @return the selected source
*/
protected Source selectSource()
{
synchronized( m_mutex )
{
final int selected = (int)(Math.random() * m_sources.size());
return (Source)m_sources.get( selected );
}
}
}
-------------------------------------------------------
This SF.Net email is sponsored by: IBM Linux Tutorials
Free Linux tutorial presented by Daniel Robbins, President and CEO of
GenToo technologies. Learn everything from fundamentals to system
administration.http://ads.osdn.com/?ad_id=1470&alloc_id=3638&op=click