/**
*
* Copyright 2004 Protique Ltd
*
* Licensed 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.activeio.adapter;
import java.io.IOException;
import java.io.InterruptedIOException;
import org.activeio.AsynchChannel;
import org.activeio.AsynchChannelListener;
import org.activeio.Channel;
import org.activeio.ChannelFactory;
import org.activeio.Packet;
import org.activeio.Service;
import org.activeio.SynchChannel;
import org.activeio.packet.EOSPacket;
import EDU.oswego.cs.dl.util.concurrent.Executor;
import EDU.oswego.cs.dl.util.concurrent.Latch;
import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
/**
* Adapts a {@see org.activeio.SynchChannel} so that it provides an
* {@see org.activeio.AsynchChannel} interface. When this channel
* is started, a background thread is used to poll the {@see org.activeio.SynchChannel}
* for packets comming up the channel which are then delivered to the
* {@see org.activeio.ChannelConsumer}.
*
* @version $Revision$
*/
final public class SynchToAsynchChannelAdapter implements AsynchChannel, Runnable {
private final SynchronizedBoolean running = new SynchronizedBoolean(false);
private final SynchChannel synchChannel;
private final Executor executor;
private AsynchChannelListener channelListener;
private Latch doneLatch;
static public AsynchChannel adapt(Channel channel) {
return adapt(channel, ChannelFactory.DEFAULT_EXECUTOR);
}
static public AsynchChannel adapt(Channel channel, Executor executor) {
// It might not need adapting
if( channel instanceof AsynchChannel ) {
return (AsynchChannel) channel;
}
// Can we just just undo the adaptor
if( channel.getClass() == SynchToAsynchChannelAdapter.class ) {
return ((AsynchToSynchChannelAdapter)channel).getAsynchChannel();
}
return new SynchToAsynchChannelAdapter((SynchChannel) channel, executor);
}
/**
* @deprecated {@see #adapt(SynchChannel)}
*/
public SynchToAsynchChannelAdapter(SynchChannel synchChannel) {
this(synchChannel, ChannelFactory.DEFAULT_EXECUTOR);
}
/**
* @deprecated {@see #adapt(SynchChannel, Executor)}
*/
public SynchToAsynchChannelAdapter(SynchChannel synchChannel, Executor executor) {
this.synchChannel = synchChannel;
this.executor = executor;
}
synchronized public void start() throws IOException {
if (running.commit(false, true)) {
if (channelListener == null)
throw new IllegalStateException("UpPacketListener must be set before object can be started.");
synchChannel.start();
try {
doneLatch = new Latch();
executor.execute(this);
} catch (InterruptedException e) {
throw new InterruptedIOException(e.getMessage());
}
}
}
synchronized public void stop(long timeout) throws IOException {
if (running.commit(true, false)) {
try {
if( timeout == NO_WAIT_TIMEOUT ) {
synchChannel.stop(NO_WAIT_TIMEOUT);
} else if( timeout == WAIT_FOREVER_TIMEOUT ) {
doneLatch.acquire();
synchChannel.stop(WAIT_FOREVER_TIMEOUT);
} else {
long start = System.currentTimeMillis();
if( doneLatch.attempt(timeout) ) {
timeout -= (System.currentTimeMillis() - start);
} else {
timeout=0;
}
if( timeout <= 0 ) {
synchChannel.stop(NO_WAIT_TIMEOUT);
} else {
synchChannel.stop(timeout);
}
}
} catch (IOException e) {
throw e;
} catch (Throwable e) {
throw (IOException)new IOException("stop failed: " + e.getMessage()).initCause(e);
}
}
}
/**
* reads packets from a Socket
*/
public void run() {
// Change the thread name.
String oldName = Thread.currentThread().getName();
Thread.currentThread().setName( synchChannel.toString() );
try {
while (running.get()) {
try {
Packet packet = synchChannel.read(500);
if( packet==null )
continue;
if( packet == EOSPacket.EOS_PACKET ) {
channelListener.onPacket(packet);
return;
}
if( packet.hasRemaining() ) {
channelListener.onPacket(packet);
}
} catch (IOException e) {
channelListener.onPacketError(e);
} catch (Throwable e) {
channelListener.onPacketError((IOException)new IOException("Unexpected Error: "+e).initCause(e));
}
}
} finally {
if( doneLatch!=null )
doneLatch.release();
Thread.currentThread().setName(oldName);
}
}
/**
* @see org.activeio.AsynchChannel#setAsynchChannelListener(org.activeio.UpPacketListener)
*/
public void setAsynchChannelListener(AsynchChannelListener channelListener) {
if (running.get())
throw new IllegalStateException("Cannot change the UpPacketListener while the object is running.");
this.channelListener = channelListener;
}
/**
* @see org.activeio.Channel#write(org.activeio.channel.Packet)
*/
public void write(org.activeio.Packet packet) throws IOException {
synchChannel.write(packet);
}
/**
* @see org.activeio.Channel#flush()
*/
public void flush() throws IOException {
synchChannel.flush();
}
/**
* @see org.activeio.Disposable#dispose()
*/
public void dispose() {
try {
stop(Service.NO_WAIT_TIMEOUT);
} catch ( IOException ignore) {
}
synchChannel.dispose();
}
public AsynchChannelListener getAsynchChannelListener() {
return channelListener;
}
public Object narrow(Class target) {
if( target.isAssignableFrom(getClass()) ) {
return this;
}
return synchChannel.narrow(target);
}
public SynchChannel getSynchChannel() {
return synchChannel;
}
public String toString() {
return synchChannel.toString();
}
}