/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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.mina.filter.executor;
import java.util.ArrayList;
import java.util.List;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoFilterAdapter;
import org.apache.mina.common.IoFilterChain;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.ThreadModel;
import org.apache.mina.util.ByteBufferUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import edu.emory.mathcs.backport.java.util.concurrent.Executor;
import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
import edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor;
import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
/**
* A filter that forward events to {@link Executor} in
* <a href="http://dcl.mathcs.emory.edu/util/backport-util-concurrent/">backport-util-concurrent</a>.
* You can apply various thread model by inserting this filter to the {@link IoFilterChain}.
* This filter is usually inserted by {@link ThreadModel} automatically, so you don't need
* to add this filter in most cases.
* <p>
* Please note that this filter doesn't manage the life cycle of the underlying
* {@link Executor}. You have to destroy or stop it by yourself.
*
* <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
* @version $Rev: 350169 $, $Date: 2005-12-01 00:17:41 -0500 (Thu, 01 Dec 2005) $
*/
public class ExecutorFilter extends IoFilterAdapter
{
private static final Logger logger = LoggerFactory.getLogger( ExecutorFilter.class.getName() );
private final Executor executor;
/**
* Creates a new instace with the default thread pool implementation
* (<tt>new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue() )</tt>).
*/
public ExecutorFilter()
{
this( new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue() ) );
}
/**
* Creates a new instance with the specified <tt>executor</tt>.
*/
public ExecutorFilter( Executor executor )
{
if( executor == null )
{
throw new NullPointerException( "executor" );
}
this.executor = executor;
}
/**
* Returns the underlying {@link Executor} instance this filter uses.
*/
public Executor getExecutor()
{
return executor;
}
private void fireEvent( NextFilter nextFilter, IoSession session,
EventType type, Object data )
{
Event event = new Event( type, nextFilter, data );
SessionBuffer buf = SessionBuffer.getSessionBuffer( session );
synchronized( buf.eventQueue )
{
buf.eventQueue.add( event );
if( buf.processingCompleted )
{
buf.processingCompleted = false;
if ( logger.isDebugEnabled() ) {
logger.debug( "Launching thread for " + session.getRemoteAddress() );
}
executor.execute( new ProcessEventsRunnable( buf ) );
}
}
}
private static class SessionBuffer
{
private static final String KEY = SessionBuffer.class.getName() + ".KEY";
private static SessionBuffer getSessionBuffer( IoSession session )
{
synchronized( session )
{
SessionBuffer buf = (SessionBuffer)session.getAttribute( KEY );
if( buf == null )
{
buf = new SessionBuffer( session );
session.setAttribute( KEY, buf );
}
return buf;
}
}
private final IoSession session;
private final List eventQueue = new ArrayList();
private boolean processingCompleted = true;
private SessionBuffer( IoSession session )
{
this.session = session;
}
}
protected static class EventType
{
public static final EventType OPENED = new EventType( "OPENED" );
public static final EventType CLOSED = new EventType( "CLOSED" );
public static final EventType READ = new EventType( "READ" );
public static final EventType WRITTEN = new EventType( "WRITTEN" );
public static final EventType RECEIVED = new EventType( "RECEIVED" );
public static final EventType SENT = new EventType( "SENT" );
public static final EventType IDLE = new EventType( "IDLE" );
public static final EventType EXCEPTION = new EventType( "EXCEPTION" );
private final String value;
private EventType( String value )
{
this.value = value;
}
public String toString()
{
return value;
}
}
protected static class Event
{
private final EventType type;
private final NextFilter nextFilter;
private final Object data;
Event( EventType type, NextFilter nextFilter, Object data )
{
this.type = type;
this.nextFilter = nextFilter;
this.data = data;
}
public Object getData()
{
return data;
}
public NextFilter getNextFilter()
{
return nextFilter;
}
public EventType getType()
{
return type;
}
}
public void sessionCreated( NextFilter nextFilter, IoSession session )
{
nextFilter.sessionCreated( session );
}
public void sessionOpened( NextFilter nextFilter,
IoSession session )
{
fireEvent( nextFilter, session, EventType.OPENED, null );
}
public void sessionClosed( NextFilter nextFilter,
IoSession session )
{
fireEvent( nextFilter, session, EventType.CLOSED, null );
}
public void sessionIdle( NextFilter nextFilter,
IoSession session, IdleStatus status )
{
fireEvent( nextFilter, session, EventType.IDLE, status );
}
public void exceptionCaught( NextFilter nextFilter,
IoSession session, Throwable cause )
{
fireEvent( nextFilter, session, EventType.EXCEPTION, cause );
}
public void messageReceived( NextFilter nextFilter,
IoSession session, Object message )
{
ByteBufferUtil.acquireIfPossible( message );
fireEvent( nextFilter, session, EventType.RECEIVED, message );
}
public void messageSent( NextFilter nextFilter,
IoSession session, Object message )
{
ByteBufferUtil.acquireIfPossible( message );
fireEvent( nextFilter, session, EventType.SENT, message );
}
protected void processEvent( NextFilter nextFilter, IoSession session, EventType type, Object data )
{
if( type == EventType.RECEIVED )
{
nextFilter.messageReceived( session, data );
ByteBufferUtil.releaseIfPossible( data );
}
else if( type == EventType.SENT )
{
nextFilter.messageSent( session, data );
ByteBufferUtil.releaseIfPossible( data );
}
else if( type == EventType.EXCEPTION )
{
nextFilter.exceptionCaught( session, (Throwable)data );
}
else if( type == EventType.IDLE )
{
nextFilter.sessionIdle( session, (IdleStatus)data );
}
else if( type == EventType.OPENED )
{
nextFilter.sessionOpened( session );
}
else if( type == EventType.CLOSED )
{
nextFilter.sessionClosed( session );
}
}
public void filterWrite( NextFilter nextFilter, IoSession session, WriteRequest writeRequest )
{
nextFilter.filterWrite( session, writeRequest );
}
public void filterClose( NextFilter nextFilter, IoSession session ) throws Exception
{
nextFilter.filterClose( session );
}
private class ProcessEventsRunnable implements Runnable
{
private final SessionBuffer buffer;
ProcessEventsRunnable( SessionBuffer buffer )
{
this.buffer = buffer;
}
public void run()
{
while( true )
{
Event event;
synchronized( buffer.eventQueue )
{
if( buffer.eventQueue.isEmpty() )
{
buffer.processingCompleted = true;
break;
}
event = ( Event ) buffer.eventQueue.remove( 0 );
}
processEvent( event.getNextFilter(), buffer.session, event.getType(), event.getData() );
}
if ( logger.isDebugEnabled() ) {
logger.debug( "Exiting since queue is empty for " + buffer.session.getRemoteAddress() );
}
}
}
}