diff options
Diffstat (limited to 'qpid/java/common/src/main/java/org/apache/mina/common')
3 files changed, 1045 insertions, 0 deletions
diff --git a/qpid/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java b/qpid/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java new file mode 100644 index 0000000000..0c311b6645 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java @@ -0,0 +1,467 @@ +package org.apache.mina.common; + +import org.apache.mina.common.ByteBuffer; + +import java.nio.*; + +/* +* +* 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. +* +*/ +public class FixedSizeByteBufferAllocator implements ByteBufferAllocator +{ + + + private static final int MINIMUM_CAPACITY = 1; + + public FixedSizeByteBufferAllocator () + { + } + + public ByteBuffer allocate( int capacity, boolean direct ) + { + java.nio.ByteBuffer nioBuffer; + if( direct ) + { + nioBuffer = java.nio.ByteBuffer.allocateDirect( capacity ); + } + else + { + nioBuffer = java.nio.ByteBuffer.allocate( capacity ); + } + return new FixedSizeByteBuffer( nioBuffer ); + } + + public ByteBuffer wrap( java.nio.ByteBuffer nioBuffer ) + { + return new FixedSizeByteBuffer( nioBuffer ); + } + + public void dispose() + { + } + + + + private static final class FixedSizeByteBuffer extends ByteBuffer + { + private java.nio.ByteBuffer buf; + private int mark = -1; + + + protected FixedSizeByteBuffer( java.nio.ByteBuffer buf ) + { + this.buf = buf; + buf.order( ByteOrder.BIG_ENDIAN ); + } + + public synchronized void acquire() + { + } + + public void release() + { + } + + public java.nio.ByteBuffer buf() + { + return buf; + } + + public boolean isPooled() + { + return false; + } + + public void setPooled( boolean pooled ) + { + } + + public ByteBuffer duplicate() { + return new FixedSizeByteBuffer( this.buf.duplicate() ); + } + + public ByteBuffer slice() { + return new FixedSizeByteBuffer( this.buf.slice() ); + } + + public ByteBuffer asReadOnlyBuffer() { + return new FixedSizeByteBuffer( this.buf.asReadOnlyBuffer() ); + } + + public byte[] array() + { + return buf.array(); + } + + public int arrayOffset() + { + return buf.arrayOffset(); + } + + public boolean isDirect() + { + return buf.isDirect(); + } + + public boolean isReadOnly() + { + return buf.isReadOnly(); + } + + public int capacity() + { + return buf.capacity(); + } + + public ByteBuffer capacity( int newCapacity ) + { + if( newCapacity > capacity() ) + { + throw new IllegalArgumentException(); + } + + return this; + } + + + + public boolean isAutoExpand() + { + return false; + } + + public ByteBuffer setAutoExpand( boolean autoExpand ) + { + if(autoExpand) throw new IllegalArgumentException(); + else return this; + } + + public ByteBuffer expand( int pos, int expectedRemaining ) + { + int end = pos + expectedRemaining; + if( end > capacity() ) + { + // The buffer needs expansion. + capacity( end ); + } + + if( end > limit() ) + { + // We call limit() directly to prevent StackOverflowError + buf.limit( end ); + } + return this; + } + + public int position() + { + return buf.position(); + } + + public ByteBuffer position( int newPosition ) + { + + buf.position( newPosition ); + if( mark > newPosition ) + { + mark = -1; + } + return this; + } + + public int limit() + { + return buf.limit(); + } + + public ByteBuffer limit( int newLimit ) + { + buf.limit( newLimit ); + if( mark > newLimit ) + { + mark = -1; + } + return this; + } + + public ByteBuffer mark() + { + buf.mark(); + mark = position(); + return this; + } + + public int markValue() + { + return mark; + } + + public ByteBuffer reset() + { + buf.reset(); + return this; + } + + public ByteBuffer clear() + { + buf.clear(); + mark = -1; + return this; + } + + public ByteBuffer flip() + { + buf.flip(); + mark = -1; + return this; + } + + public ByteBuffer rewind() + { + buf.rewind(); + mark = -1; + return this; + } + + public byte get() + { + return buf.get(); + } + + public ByteBuffer put( byte b ) + { + buf.put( b ); + return this; + } + + public byte get( int index ) + { + return buf.get( index ); + } + + public ByteBuffer put( int index, byte b ) + { + buf.put( index, b ); + return this; + } + + public ByteBuffer get( byte[] dst, int offset, int length ) + { + buf.get( dst, offset, length ); + return this; + } + + public ByteBuffer put( java.nio.ByteBuffer src ) + { + buf.put( src ); + return this; + } + + public ByteBuffer put( byte[] src, int offset, int length ) + { + buf.put( src, offset, length ); + return this; + } + + public ByteBuffer compact() + { + buf.compact(); + mark = -1; + return this; + } + + public ByteOrder order() + { + return buf.order(); + } + + public ByteBuffer order( ByteOrder bo ) + { + buf.order( bo ); + return this; + } + + public char getChar() + { + return buf.getChar(); + } + + public ByteBuffer putChar( char value ) + { + buf.putChar( value ); + return this; + } + + public char getChar( int index ) + { + return buf.getChar( index ); + } + + public ByteBuffer putChar( int index, char value ) + { + buf.putChar( index, value ); + return this; + } + + public CharBuffer asCharBuffer() + { + return buf.asCharBuffer(); + } + + public short getShort() + { + return buf.getShort(); + } + + public ByteBuffer putShort( short value ) + { + buf.putShort( value ); + return this; + } + + public short getShort( int index ) + { + return buf.getShort( index ); + } + + public ByteBuffer putShort( int index, short value ) + { + buf.putShort( index, value ); + return this; + } + + public ShortBuffer asShortBuffer() + { + return buf.asShortBuffer(); + } + + public int getInt() + { + return buf.getInt(); + } + + public ByteBuffer putInt( int value ) + { + buf.putInt( value ); + return this; + } + + public int getInt( int index ) + { + return buf.getInt( index ); + } + + public ByteBuffer putInt( int index, int value ) + { + buf.putInt( index, value ); + return this; + } + + public IntBuffer asIntBuffer() + { + return buf.asIntBuffer(); + } + + public long getLong() + { + return buf.getLong(); + } + + public ByteBuffer putLong( long value ) + { + buf.putLong( value ); + return this; + } + + public long getLong( int index ) + { + return buf.getLong( index ); + } + + public ByteBuffer putLong( int index, long value ) + { + buf.putLong( index, value ); + return this; + } + + public LongBuffer asLongBuffer() + { + return buf.asLongBuffer(); + } + + public float getFloat() + { + return buf.getFloat(); + } + + public ByteBuffer putFloat( float value ) + { + buf.putFloat( value ); + return this; + } + + public float getFloat( int index ) + { + return buf.getFloat( index ); + } + + public ByteBuffer putFloat( int index, float value ) + { + buf.putFloat( index, value ); + return this; + } + + public FloatBuffer asFloatBuffer() + { + return buf.asFloatBuffer(); + } + + public double getDouble() + { + return buf.getDouble(); + } + + public ByteBuffer putDouble( double value ) + { + buf.putDouble( value ); + return this; + } + + public double getDouble( int index ) + { + return buf.getDouble( index ); + } + + public ByteBuffer putDouble( int index, double value ) + { + buf.putDouble( index, value ); + return this; + } + + public DoubleBuffer asDoubleBuffer() + { + return buf.asDoubleBuffer(); + } + + + } + + +} diff --git a/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java b/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java new file mode 100644 index 0000000000..4fd28c4eb5 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/mina/common/support/DefaultIoFuture.java @@ -0,0 +1,227 @@ +/* + * 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.common.support; + +import org.apache.mina.common.IoFuture; +import org.apache.mina.common.IoSession; +import org.apache.mina.common.IoFutureListener; + +import java.util.List; +import java.util.ArrayList; +import java.util.Iterator; + +/** + * A default implementation of {@link org.apache.mina.common.IoFuture}. + * + * @author The Apache Directory Project (mina-dev@directory.apache.org) + */ +public class DefaultIoFuture implements IoFuture +{ + private final IoSession session; + private final Object lock; + private List listeners; + private Object result; + private boolean ready; + + + /** + * Creates a new instance. + * + * @param session an {@link IoSession} which is associated with this future + */ + public DefaultIoFuture( IoSession session ) + { + this.session = session; + this.lock = this; + } + + /** + * Creates a new instance which uses the specified object as a lock. + */ + public DefaultIoFuture( IoSession session, Object lock ) + { + if( lock == null ) + { + throw new NullPointerException( "lock" ); + } + this.session = session; + this.lock = lock; + } + + public IoSession getSession() + { + return session; + } + + public Object getLock() + { + return lock; + } + + public void join() + { + synchronized( lock ) + { + while( !ready ) + { + try + { + lock.wait(); + } + catch( InterruptedException e ) + { + } + } + } + } + + public boolean join( long timeoutInMillis ) + { + long startTime = ( timeoutInMillis <= 0 ) ? 0 : System + .currentTimeMillis(); + long waitTime = timeoutInMillis; + + synchronized( lock ) + { + if( ready ) + { + return ready; + } + else if( waitTime <= 0 ) + { + return ready; + } + + for( ;; ) + { + try + { + lock.wait( waitTime ); + } + catch( InterruptedException e ) + { + } + + if( ready ) + return true; + else + { + waitTime = timeoutInMillis - ( System.currentTimeMillis() - startTime ); + if( waitTime <= 0 ) + { + return ready; + } + } + } + } + } + + public boolean isReady() + { + synchronized( lock ) + { + return ready; + } + } + + /** + * Sets the result of the asynchronous operation, and mark it as finished. + */ + protected void setValue( Object newValue ) + { + synchronized( lock ) + { + // Allow only once. + if( ready ) + { + return; + } + + result = newValue; + ready = true; + lock.notifyAll(); + + notifyListeners(); + } + } + + /** + * Returns the result of the asynchronous operation. + */ + protected Object getValue() + { + synchronized( lock ) + { + return result; + } + } + + public void addListener( IoFutureListener listener ) + { + if( listener == null ) + { + throw new NullPointerException( "listener" ); + } + + synchronized( lock ) + { + if(listeners == null) + { + listeners = new ArrayList(); + } + listeners.add( listener ); + if( ready ) + { + listener.operationComplete( this ); + } + } + } + + public void removeListener( IoFutureListener listener ) + { + if( listener == null ) + { + throw new NullPointerException( "listener" ); + } + + synchronized( lock ) + { + listeners.remove( listener ); + } + } + + private void notifyListeners() + { + synchronized( lock ) + { + + if(listeners != null) + { + + for( Iterator i = listeners.iterator(); i.hasNext(); ) { + ( ( IoFutureListener ) i.next() ).operationComplete( this ); + } + } + } + } +} + + + diff --git a/qpid/java/common/src/main/java/org/apache/mina/common/support/IoServiceListenerSupport.java b/qpid/java/common/src/main/java/org/apache/mina/common/support/IoServiceListenerSupport.java new file mode 100644 index 0000000000..5723ffbaa9 --- /dev/null +++ b/qpid/java/common/src/main/java/org/apache/mina/common/support/IoServiceListenerSupport.java @@ -0,0 +1,351 @@ +/* + * 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.common.support; + +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; + +import org.apache.mina.common.IoAcceptorConfig; +import org.apache.mina.common.IoConnector; +import org.apache.mina.common.IoFuture; +import org.apache.mina.common.IoFutureListener; +import org.apache.mina.common.IoHandler; +import org.apache.mina.common.IoService; +import org.apache.mina.common.IoServiceConfig; +import org.apache.mina.common.IoServiceListener; +import org.apache.mina.common.IoSession; +import org.apache.mina.util.IdentityHashSet; + +/** + * A helper which provides addition and removal of {@link IoServiceListener}s and firing + * events. + * + * @author The Apache Directory Project (mina-dev@directory.apache.org) + * @version $Rev: 446526 $, $Date: 2006-09-15 01:44:11 -0400 (Fri, 15 Sep 2006) $ + */ +public class IoServiceListenerSupport +{ + /** + * A list of {@link IoServiceListener}s. + */ + private final List listeners = new ArrayList(); + + /** + * Tracks managed <tt>serviceAddress</tt>es. + */ + private final Set managedServiceAddresses = new HashSet(); + + /** + * Tracks managed sesssions with <tt>serviceAddress</tt> as a key. + */ + private final Map managedSessions = new HashMap(); + + /** + * Creates a new instance. + */ + public IoServiceListenerSupport() + { + } + + /** + * Adds a new listener. + */ + public void add( IoServiceListener listener ) + { + synchronized( listeners ) + { + listeners.add( listener ); + } + } + + /** + * Removes an existing listener. + */ + public void remove( IoServiceListener listener ) + { + synchronized( listeners ) + { + listeners.remove( listener ); + } + } + + public Set getManagedServiceAddresses() + { + return Collections.unmodifiableSet( managedServiceAddresses ); + } + + public boolean isManaged( SocketAddress serviceAddress ) + { + synchronized( managedServiceAddresses ) + { + return managedServiceAddresses.contains( serviceAddress ); + } + } + + public Set getManagedSessions( SocketAddress serviceAddress ) + { + Set sessions; + synchronized( managedSessions ) + { + sessions = ( Set ) managedSessions.get( serviceAddress ); + if( sessions == null ) + { + sessions = new IdentityHashSet(); + } + } + + synchronized( sessions ) + { + return new IdentityHashSet( sessions ); + } + } + + /** + * Calls {@link IoServiceListener#serviceActivated(IoService, SocketAddress, IoHandler, IoServiceConfig)} + * for all registered listeners. + */ + public void fireServiceActivated( + IoService service, SocketAddress serviceAddress, + IoHandler handler, IoServiceConfig config ) + { + synchronized( managedServiceAddresses ) + { + if( !managedServiceAddresses.add( serviceAddress ) ) + { + return; + } + } + + synchronized( listeners ) + { + for( Iterator i = listeners.iterator(); i.hasNext(); ) + { + ( ( IoServiceListener ) i.next() ).serviceActivated( + service, serviceAddress, handler, config ); + } + } + } + + /** + * Calls {@link IoServiceListener#serviceDeactivated(IoService, SocketAddress, IoHandler, IoServiceConfig)} + * for all registered listeners. + */ + public synchronized void fireServiceDeactivated( + IoService service, SocketAddress serviceAddress, + IoHandler handler, IoServiceConfig config ) + { + synchronized( managedServiceAddresses ) + { + if( !managedServiceAddresses.remove( serviceAddress ) ) + { + return; + } + } + + try + { + synchronized( listeners ) + { + for( Iterator i = listeners.iterator(); i.hasNext(); ) + { + ( ( IoServiceListener ) i.next() ).serviceDeactivated( + service, serviceAddress, handler, config ); + } + } + } + finally + { + disconnectSessions( serviceAddress, config ); + } + } + + + /** + * Calls {@link IoServiceListener#sessionCreated(IoSession)} for all registered listeners. + */ + public void fireSessionCreated( IoSession session ) + { + SocketAddress serviceAddress = session.getServiceAddress(); + + // Get the session set. + boolean firstSession = false; + Set sessions; + synchronized( managedSessions ) + { + sessions = ( Set ) managedSessions.get( serviceAddress ); + if( sessions == null ) + { + sessions = new IdentityHashSet(); + managedSessions.put( serviceAddress, sessions ); + firstSession = true; + } + } + + // If already registered, ignore. + synchronized( sessions ) + { + if ( !sessions.add( session ) ) + { + return; + } + } + + // If the first connector session, fire a virtual service activation event. + if( session.getService() instanceof IoConnector && firstSession ) + { + fireServiceActivated( + session.getService(), session.getServiceAddress(), + session.getHandler(), session.getServiceConfig() ); + } + + // Fire session events. + session.getFilterChain().fireSessionCreated( session ); + session.getFilterChain().fireSessionOpened( session); + + // Fire listener events. + synchronized( listeners ) + { + for( Iterator i = listeners.iterator(); i.hasNext(); ) + { + ( ( IoServiceListener ) i.next() ).sessionCreated( session ); + } + } + } + + /** + * Calls {@link IoServiceListener#sessionDestroyed(IoSession)} for all registered listeners. + */ + public void fireSessionDestroyed( IoSession session ) + { + SocketAddress serviceAddress = session.getServiceAddress(); + + // Get the session set. + Set sessions; + boolean lastSession = false; + synchronized( managedSessions ) + { + sessions = ( Set ) managedSessions.get( serviceAddress ); + // Ignore if unknown. + if( sessions == null ) + { + return; + } + + // Try to remove the remaining empty seession set after removal. + synchronized( sessions ) + { + sessions.remove( session ); + if( sessions.isEmpty() ) + { + managedSessions.remove( serviceAddress ); + lastSession = true; + } + } + } + + // Fire session events. + session.getFilterChain().fireSessionClosed( session ); + + // Fire listener events. + try + { + synchronized( listeners ) + { + for( Iterator i = listeners.iterator(); i.hasNext(); ) + { + ( ( IoServiceListener ) i.next() ).sessionDestroyed( session ); + } + } + } + finally + { + // Fire a virtual service deactivation event for the last session of the connector. + //TODO double-check that this is *STILL* the last session. May not be the case + if( session.getService() instanceof IoConnector && lastSession ) + { + fireServiceDeactivated( + session.getService(), session.getServiceAddress(), + session.getHandler(), session.getServiceConfig() ); + } + } + } + + private void disconnectSessions( SocketAddress serviceAddress, IoServiceConfig config ) + { + if( !( config instanceof IoAcceptorConfig ) ) + { + return; + } + + if( !( ( IoAcceptorConfig ) config ).isDisconnectOnUnbind() ) + { + return; + } + + Set sessions; + synchronized( managedSessions ) + { + sessions = ( Set ) managedSessions.get( serviceAddress ); + } + + if( sessions == null ) + { + return; + } + + Set sessionsCopy; + + // Create a copy to avoid ConcurrentModificationException + synchronized( sessions ) + { + sessionsCopy = new IdentityHashSet( sessions ); + } + + final CountDownLatch latch = new CountDownLatch(sessionsCopy.size()); + + for( Iterator i = sessionsCopy.iterator(); i.hasNext(); ) + { + ( ( IoSession ) i.next() ).close().addListener( new IoFutureListener() + { + public void operationComplete( IoFuture future ) + { + latch.countDown(); + } + } ); + } + + try + { + latch.await(); + } + catch( InterruptedException ie ) + { + // Ignored + } + } +} |