plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/SignalProtocol.java - cdo/cdo - Git at Google

 /*
  * Copyright (c) 2006-2015 Eike Stepper (Berlin, Germany) and others.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
  * http://www.eclipse.org/legal/epl-v10.html
  *
  * Contributors:
  *    Eike Stepper - initial API and implementation
  *    Andre Dietisheim - maintenance
  */
 package org.eclipse.net4j.signal;

 import org.eclipse.net4j.buffer.BufferInputStream;
 import org.eclipse.net4j.buffer.IBuffer;
 import org.eclipse.net4j.buffer.IBufferProvider;
 import org.eclipse.net4j.channel.ChannelOutputStream;
 import org.eclipse.net4j.channel.IChannel;
 import org.eclipse.net4j.connector.IConnector;
 import org.eclipse.net4j.util.WrappedException;
 import org.eclipse.net4j.util.concurrent.ConcurrencyUtil;
 import org.eclipse.net4j.util.event.Event;
 import org.eclipse.net4j.util.event.IEvent;
 import org.eclipse.net4j.util.event.IListener;
 import org.eclipse.net4j.util.io.IORuntimeException;
 import org.eclipse.net4j.util.io.IStreamWrapper;
 import org.eclipse.net4j.util.io.StreamWrapperChain;
 import org.eclipse.net4j.util.lifecycle.LifecycleUtil;
 import org.eclipse.net4j.util.om.log.OMLogger;
 import org.eclipse.net4j.util.om.trace.ContextTracer;

 import org.eclipse.internal.net4j.bundle.OM;

 import org.eclipse.spi.net4j.InternalChannel;
 import org.eclipse.spi.net4j.Protocol;

 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.text.MessageFormat;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;

 /**
  * The default implementation of a {@link ISignalProtocol signal protocol}.
  * <p>
  * On the {@link org.eclipse.net4j.ILocationAware.Location#SERVER receiver side(s)} of protocol the
  * {@link #createSignalReactor(short) createSignalReactor()} method has to be overridden to
  * create appropriate peer instances for incoming {@link Signal signals}.
  *
  * @author Eike Stepper
  */
 public class SignalProtocol<INFRA_STRUCTURE> extends Protocol<INFRA_STRUCTURE>
     implements ISignalProtocol<INFRA_STRUCTURE>
 {
   /**
    * @since 2.0
    */
   public static final short SIGNAL_REMOTE_EXCEPTION = -1;

   /**
    * @since 2.0
    */
   public static final short SIGNAL_MONITOR_CANCELED = -2;

   /**
    * @since 2.0
    */
   public static final short SIGNAL_MONITOR_PROGRESS = -3;

   /**
    * @since 4.1
    */
   public static final short SIGNAL_SET_TIMEOUT = -4;

   private static final int MIN_CORRELATION_ID = 1;

   private static final int MAX_CORRELATION_ID = Integer.MAX_VALUE;

   private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG_SIGNAL, SignalProtocol.class);

   private static final ContextTracer STREAM_TRACER = new ContextTracer(OM.DEBUG_BUFFER_STREAM, SignalProtocol.class);

   private long timeout = DEFAULT_TIMEOUT;

   private IStreamWrapper streamWrapper;

   private Map<Integer, Signal> signals = new HashMap<Integer, Signal>();

   private int nextCorrelationID = MIN_CORRELATION_ID;

   private boolean failingOver;

   /**
    * @since 2.0
    */
   public SignalProtocol(String type)
   {
     super(type);
   }

   /**
    * @since 2.0
    */
   public long getTimeout()
   {
     return timeout;
   }

   /**
    * Equivalent to calling SignalProtocol.setTimeout(timeout, false).
    *
    * @since 2.0
    */
   public void setTimeout(long timeout)
   {
     setTimeout(timeout, false);
   }

   /**
    * Update the timeout used for signal end of stream waiting time.
    *
    * @param timeout the new timeout
    * @param useOldTimeoutToSendNewOne true to use the old timeout, false to use the new specified one to sent to server side the new specified timeout
    * @return true if the new specified timeout has correctly been sent
    *
    * NOTE: this second parameter is useful mostly for test to be able to set a to small timeout
    * @since 4.4
    */
   public boolean setTimeout(long timeout, boolean useOldTimeoutToSendNewOne)
   {
     boolean timeoutSent = false;
     long oldTimeout = this.timeout;
     if (!useOldTimeoutToSendNewOne)
     {
       handleSetTimeOut(timeout);
     }

     if (oldTimeout != timeout && isActive())
     {
       timeoutSent = sendSetTimeout();
     }

     if (timeoutSent && useOldTimeoutToSendNewOne)
     {
       handleSetTimeOut(timeout);
     }

     return timeoutSent;
   }

   public IStreamWrapper getStreamWrapper()
   {
     return streamWrapper;
   }

   public void setStreamWrapper(IStreamWrapper streamWrapper)
   {
     this.streamWrapper = streamWrapper;
   }

   public void addStreamWrapper(IStreamWrapper streamWrapper)
   {
     if (this.streamWrapper == null)
     {
       this.streamWrapper = streamWrapper;
     }
     else
     {
       this.streamWrapper = new StreamWrapperChain(streamWrapper, this.streamWrapper);
     }
   }

   /**
    * @since 2.0
    */
   public IChannel open(IConnector connector)
   {
     return connector.openChannel(this);
   }

   /**
    * @since 2.0
    */
   public void close()
   {
     LifecycleUtil.deactivate(this, OMLogger.Level.DEBUG);
   }

   public boolean waitForSignals(long timeout)
   {
     synchronized (signals)
     {
       while (!signals.isEmpty())
       {
         try
         {
           signals.wait(timeout);
         }
         catch (InterruptedException ex)
         {
           return false;
         }
       }
     }

     return true;
   }

   /**
    * Handles a given (incoming) buffer. Creates a signal to act upon the given buffer or uses a previously created
    * signal.
    */
   public void handleBuffer(IBuffer buffer)
   {
     ByteBuffer byteBuffer = buffer.getByteBuffer();
     int correlationID = byteBuffer.getInt();
     if (TRACER.isEnabled())
     {
       TRACER.trace("Received buffer for correlation " + correlationID); //$NON-NLS-1$
     }

     Signal signal;
     boolean newSignalScheduled = false;

     synchronized (signals)
     {
       if (correlationID > 0)
       {
         // Incoming indication
         signal = signals.get(-correlationID);
         if (signal == null)
         {
           short signalID = byteBuffer.getShort();
           if (TRACER.isEnabled())
           {
             TRACER.trace("Got signalID: " + signalID); //$NON-NLS-1$
           }

           signal = provideSignalReactor(signalID);
           if (signal != null)
           {
             signal.setCorrelationID(-correlationID);
             signal.setBufferInputStream(new SignalInputStream(getTimeout()));
             if (signal instanceof IndicationWithResponse)
             {
               signal.setBufferOutputStream(new SignalOutputStream(-correlationID, signalID, false));
             }

             signals.put(-correlationID, signal);
             getExecutorService().execute(signal);
             newSignalScheduled = true;
           }
         }
       }
       else
       {
         // Incoming confirmation
         signal = signals.get(-correlationID);
       }
     }

     if (signal != null) // Can be null after timeout
     {
       if (newSignalScheduled)
       {
         fireSignalScheduledEvent(signal);
       }

       BufferInputStream inputStream = signal.getBufferInputStream();
       inputStream.handleBuffer(buffer);
     }
     else
     {
       if (TRACER.isEnabled())
       {
         TRACER.trace("Discarding buffer"); //$NON-NLS-1$
       }

       buffer.release();
     }
   }

   @Override
   public String toString()
   {
     IChannel channel = getChannel();
     if (channel != null)
     {
       return MessageFormat.format("SignalProtocol[{0}, {1}, {2}]", channel.getID(), channel.getLocation(), getType()); //$NON-NLS-1$
     }

     return MessageFormat.format("SignalProtocol[{0}]", getType()); //$NON-NLS-1$
   }

   @Override
   protected void doBeforeDeactivate() throws Exception
   {
     synchronized (signals)
     {
       // Wait at most 10 seconds for running signals to finish
       int waitMillis = 10 * 1000;
       long stop = System.currentTimeMillis() + waitMillis;
       while (!signals.isEmpty() && System.currentTimeMillis() < stop)
       {
         signals.wait(1000L);
       }
     }
   }

   @Override
   protected void doDeactivate() throws Exception
   {
     try
     {
       synchronized (signals)
       {
         signals.clear();
       }

       IChannel channel = getChannel();
       if (channel != null)
       {
         channel.close();
         setChannel(null);
       }
     }
     finally
     {
       super.doDeactivate();
     }
   }

   @Override
   protected void handleChannelDeactivation()
   {
     if (!failingOver)
     {
       super.handleChannelDeactivation();
     }
   }

   protected final SignalReactor provideSignalReactor(short signalID)
   {
     if (!isActive())
     {
       return null;
     }

     switch (signalID)
     {
     case SIGNAL_REMOTE_EXCEPTION:
       return new RemoteExceptionIndication(this);

     case SIGNAL_MONITOR_CANCELED:
       return new MonitorCanceledIndication(this);

     case SIGNAL_MONITOR_PROGRESS:
       return new MonitorProgressIndication(this);

     case SIGNAL_SET_TIMEOUT:
       return new SetTimeoutIndication(this);

     default:
       SignalReactor signal = createSignalReactor(signalID);
       if (signal == null)
       {
         throw new IllegalArgumentException("Invalid signalID " + signalID); //$NON-NLS-1$
       }

       return signal;
     }
   }

   /**
    * Returns a new signal instance to serve the given signal ID or <code>null</code> if the signal ID is invalid/unknown
    * for this protocol.
    */
   protected SignalReactor createSignalReactor(short signalID)
   {
     return null;
   }

   /**
    * Returns <code>true</code> by default, override to change this behaviour.
    *
    * @since 4.1
    */
   protected boolean isSendingTimeoutChanges()
   {
     return true;
   }

   synchronized int getNextCorrelationID()
   {
     int correlationID = nextCorrelationID;
     if (nextCorrelationID == MAX_CORRELATION_ID)
     {
       if (TRACER.isEnabled())
       {
         TRACER.trace("Correlation ID wrap-around"); //$NON-NLS-1$
       }

       nextCorrelationID = MIN_CORRELATION_ID;
     }
     else
     {
       ++nextCorrelationID;
     }

     return correlationID;
   }

   InputStream wrapInputStream(InputStream in) throws IOException
   {
     if (streamWrapper != null)
     {
       in = streamWrapper.wrapInputStream(in);
     }

     return in;
   }

   OutputStream wrapOutputStream(OutputStream out) throws IOException
   {
     if (streamWrapper != null)
     {
       out = streamWrapper.wrapOutputStream(out);
     }

     return out;
   }

   void finishInputStream(InputStream in) throws IOException
   {
     if (streamWrapper != null)
     {
       streamWrapper.finishInputStream(in);
     }
   }

   void finishOutputStream(OutputStream out) throws IOException
   {
     if (streamWrapper != null)
     {
       streamWrapper.finishOutputStream(out);
     }
   }

   void startSignal(SignalActor signalActor, long timeout) throws Exception
   {
     checkArg(signalActor.getProtocol() == this, "Wrong protocol"); //$NON-NLS-1$
     short signalID = signalActor.getID();
     int correlationID = signalActor.getCorrelationID();
     signalActor.setBufferOutputStream(new SignalOutputStream(correlationID, signalID, true));
     if (signalActor instanceof RequestWithConfirmation<?>)
     {
       signalActor.setBufferInputStream(new SignalInputStream(timeout));
     }

     synchronized (signals)
     {
       signals.put(correlationID, signalActor);
     }

     fireSignalScheduledEvent(signalActor);
     signalActor.runSync();
   }

   void stopSignal(Signal signal, Exception exception)
   {
     int correlationID = signal.getCorrelationID();
     synchronized (signals)
     {
       signals.remove(correlationID);
       signals.notifyAll();
     }

     fireSignalFinishedEvent(signal, exception);
   }

   void handleRemoteException(int correlationID, Throwable t, boolean responding)
   {
     synchronized (signals)
     {
       Signal signal = signals.remove(correlationID);
       if (signal instanceof RequestWithConfirmation<?>)
       {
         RequestWithConfirmation<?> request = (RequestWithConfirmation<?>)signal;
         request.setRemoteException(t, responding);
       }

       signals.notifyAll();
     }
   }

   void handleMonitorProgress(int correlationID, double totalWork, double work)
   {
     synchronized (signals)
     {
       Signal signal = signals.get(correlationID);
       if (signal instanceof RequestWithMonitoring<?>)
       {
         RequestWithMonitoring<?> request = (RequestWithMonitoring<?>)signal;
         request.setMonitorProgress(totalWork, work);
       }
     }
   }

   void handleMonitorCanceled(int correlationID)
   {
     synchronized (signals)
     {
       Signal signal = signals.get(-correlationID);
       if (signal instanceof IndicationWithMonitoring)
       {
         IndicationWithMonitoring indication = (IndicationWithMonitoring)signal;
         indication.setMonitorCanceled();
       }
     }
   }

   void handleSetTimeOut(long timeout)
   {
     long oldTimeout = this.timeout;
     if (oldTimeout != timeout)
     {
       this.timeout = timeout;
       fireEvent(new TimeoutChangedEvent(this, oldTimeout, timeout));
     }
   }

   boolean sendSetTimeout()
   {
     boolean timeoutSent = false;
     if (isSendingTimeoutChanges())
     {
       try
       {
         timeoutSent = new SetTimeoutRequest(this, this.timeout).send();
       }
       catch (Exception ex)
       {
         throw WrappedException.wrap(ex);
       }
     }
     return timeoutSent;
   }

   private void fireSignalScheduledEvent(Signal signal)
   {
     IListener[] listeners = getListeners();
     if (listeners != null)
     {
       IEvent event = new SignalScheduledEvent<INFRA_STRUCTURE>(this, signal);
       fireEvent(event, listeners);
     }
   }

   private void fireSignalFinishedEvent(Signal signal, Exception exception)
   {
     IListener[] listeners = getListeners();
     if (listeners != null)
     {
       IEvent event = new SignalFinishedEvent<INFRA_STRUCTURE>(this, signal, exception);
       fireEvent(event, listeners);
     }
   }

   /**
    * An {@link IEvent event} fired from a {@link ISignalProtocol signal protocol} when the protocol {@link ISignalProtocol#setTimeout(long) timeout}
    * has been changed.
    *
    * @author Eike Stepper
    * @since 4.1
    */
   public static final class TimeoutChangedEvent extends Event
   {
     private static final long serialVersionUID = 1L;

     private long oldTimeout;

     private long newTimeout;

     private TimeoutChangedEvent(ISignalProtocol<?> source, long oldTimeout, long newTimeout)
     {
       super(source);
       this.oldTimeout = oldTimeout;
       this.newTimeout = newTimeout;
     }

     @Override
     public SignalProtocol<?> getSource()
     {
       return (SignalProtocol<?>)super.getSource();
     }

     public long getOldTimeout()
     {
       return oldTimeout;
     }

     public long getNewTimeout()
     {
       return newTimeout;
     }

     @Override
     public String toString()
     {
       return "TimeoutChangedEvent [oldTimeout=" + oldTimeout + ", newTimeout=" + newTimeout + ", source=" + source
           + "]";
     }

   }

   /**
    * @author Eike Stepper
    */
   class SignalInputStream extends BufferInputStream
   {
     private long timeout;

     public SignalInputStream(long timeout)
     {
       this.timeout = timeout;
     }

     @Override
     public long getMillisBeforeTimeout()
     {
       return timeout;
     }

     @Override
     public int read() throws IOException
     {
       if (isCCAM())
       {
         final InternalChannel channel = (InternalChannel)getChannel();

         ExecutorService executorService = channel.getReceiveExecutor();
         executorService.submit(new Runnable()
         {
           public void run()
           {
             ConcurrencyUtil.sleep(500);
             channel.close();
           }
         });
       }

       return super.read();
     }
   }

   /**
    * @author Eike Stepper
    */
   class SignalOutputStream extends ChannelOutputStream
   {
     public SignalOutputStream(final int correlationID, final short signalID, final boolean addSignalID)
     {
       super(getChannel(), new IBufferProvider()
       {
         private IBufferProvider delegate = getBufferProvider();

         private boolean firstBuffer = addSignalID;

         public short getBufferCapacity()
         {
           return delegate.getBufferCapacity();
         }

         public IBuffer provideBuffer()
         {
           IChannel channel = getChannel();
           if (channel == null)
           {
             throw new IORuntimeException("No channel for protocol " + SignalProtocol.this); //$NON-NLS-1$
           }

           IBuffer buffer = delegate.provideBuffer();
           ByteBuffer byteBuffer = buffer.startPutting(channel.getID());
           if (STREAM_TRACER.isEnabled())
           {
             STREAM_TRACER.trace("Providing buffer for correlation " + correlationID); //$NON-NLS-1$
           }

           byteBuffer.putInt(correlationID);
           if (firstBuffer)
           {
             if (SignalProtocol.TRACER.isEnabled())
             {
               STREAM_TRACER.trace("Put signal id " + signalID); //$NON-NLS-1$
             }

             byteBuffer.putShort(signalID);
           }

           firstBuffer = false;
           return buffer;
         }

         public void retainBuffer(IBuffer buffer)
         {
           delegate.retainBuffer(buffer);
         }
       });
     }
   }
 }
	/*
	* Copyright (c) 2006-2015 Eike Stepper (Berlin, Germany) and others.
	* All rights reserved. This program and the accompanying materials
	* are made available under the terms of the Eclipse Public License v1.0
	* which accompanies this distribution, and is available at
	* http://www.eclipse.org/legal/epl-v10.html
	*
	* Contributors:
	* Eike Stepper - initial API and implementation
	* Andre Dietisheim - maintenance
	*/
	package org.eclipse.net4j.signal;

	import org.eclipse.net4j.buffer.BufferInputStream;
	import org.eclipse.net4j.buffer.IBuffer;
	import org.eclipse.net4j.buffer.IBufferProvider;
	import org.eclipse.net4j.channel.ChannelOutputStream;
	import org.eclipse.net4j.channel.IChannel;
	import org.eclipse.net4j.connector.IConnector;
	import org.eclipse.net4j.util.WrappedException;
	import org.eclipse.net4j.util.concurrent.ConcurrencyUtil;
	import org.eclipse.net4j.util.event.Event;
	import org.eclipse.net4j.util.event.IEvent;
	import org.eclipse.net4j.util.event.IListener;
	import org.eclipse.net4j.util.io.IORuntimeException;
	import org.eclipse.net4j.util.io.IStreamWrapper;
	import org.eclipse.net4j.util.io.StreamWrapperChain;
	import org.eclipse.net4j.util.lifecycle.LifecycleUtil;
	import org.eclipse.net4j.util.om.log.OMLogger;
	import org.eclipse.net4j.util.om.trace.ContextTracer;

	import org.eclipse.internal.net4j.bundle.OM;

	import org.eclipse.spi.net4j.InternalChannel;
	import org.eclipse.spi.net4j.Protocol;

	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.nio.ByteBuffer;
	import java.text.MessageFormat;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.concurrent.ExecutorService;

	/**
	* The default implementation of a {@link ISignalProtocol signal protocol}.
	* <p>
	* On the {@link org.eclipse.net4j.ILocationAware.Location#SERVER receiver side(s)} of protocol the
	* {@link #createSignalReactor(short) createSignalReactor()} method has to be overridden to
	* create appropriate peer instances for incoming {@link Signal signals}.
	*
	* @author Eike Stepper
	*/
	public class SignalProtocol<INFRA_STRUCTURE> extends Protocol<INFRA_STRUCTURE>
	implements ISignalProtocol<INFRA_STRUCTURE>
	{
	/**
	* @since 2.0
	*/
	public static final short SIGNAL_REMOTE_EXCEPTION = -1;

	/**
	* @since 2.0
	*/
	public static final short SIGNAL_MONITOR_CANCELED = -2;

	/**
	* @since 2.0
	*/
	public static final short SIGNAL_MONITOR_PROGRESS = -3;

	/**
	* @since 4.1
	*/
	public static final short SIGNAL_SET_TIMEOUT = -4;

	private static final int MIN_CORRELATION_ID = 1;

	private static final int MAX_CORRELATION_ID = Integer.MAX_VALUE;

	private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG_SIGNAL, SignalProtocol.class);

	private static final ContextTracer STREAM_TRACER = new ContextTracer(OM.DEBUG_BUFFER_STREAM, SignalProtocol.class);

	private long timeout = DEFAULT_TIMEOUT;

	private IStreamWrapper streamWrapper;

	private Map<Integer, Signal> signals = new HashMap<Integer, Signal>();

	private int nextCorrelationID = MIN_CORRELATION_ID;

	private boolean failingOver;

	/**
	* @since 2.0
	*/
	public SignalProtocol(String type)
	{
	super(type);
	}

	/**
	* @since 2.0
	*/
	public long getTimeout()
	{
	return timeout;
	}

	/**
	* Equivalent to calling SignalProtocol.setTimeout(timeout, false).
	*
	* @since 2.0
	*/
	public void setTimeout(long timeout)
	{
	setTimeout(timeout, false);
	}

	/**
	* Update the timeout used for signal end of stream waiting time.
	*
	* @param timeout the new timeout
	* @param useOldTimeoutToSendNewOne true to use the old timeout, false to use the new specified one to sent to server side the new specified timeout
	* @return true if the new specified timeout has correctly been sent
	*
	* NOTE: this second parameter is useful mostly for test to be able to set a to small timeout
	* @since 4.4
	*/
	public boolean setTimeout(long timeout, boolean useOldTimeoutToSendNewOne)
	{
	boolean timeoutSent = false;
	long oldTimeout = this.timeout;
	if (!useOldTimeoutToSendNewOne)
	{
	handleSetTimeOut(timeout);
	}

	if (oldTimeout != timeout && isActive())
	{
	timeoutSent = sendSetTimeout();
	}

	if (timeoutSent && useOldTimeoutToSendNewOne)
	{
	handleSetTimeOut(timeout);
	}

	return timeoutSent;
	}

	public IStreamWrapper getStreamWrapper()
	{
	return streamWrapper;
	}

	public void setStreamWrapper(IStreamWrapper streamWrapper)
	{
	this.streamWrapper = streamWrapper;
	}

	public void addStreamWrapper(IStreamWrapper streamWrapper)
	{
	if (this.streamWrapper == null)
	{
	this.streamWrapper = streamWrapper;
	}
	else
	{
	this.streamWrapper = new StreamWrapperChain(streamWrapper, this.streamWrapper);
	}
	}

	/**
	* @since 2.0
	*/
	public IChannel open(IConnector connector)
	{
	return connector.openChannel(this);
	}

	/**
	* @since 2.0
	*/
	public void close()
	{
	LifecycleUtil.deactivate(this, OMLogger.Level.DEBUG);
	}

	public boolean waitForSignals(long timeout)
	{
	synchronized (signals)
	{
	while (!signals.isEmpty())
	{
	try
	{
	signals.wait(timeout);
	}
	catch (InterruptedException ex)
	{
	return false;
	}
	}
	}

	return true;
	}

	/**
	* Handles a given (incoming) buffer. Creates a signal to act upon the given buffer or uses a previously created
	* signal.
	*/
	public void handleBuffer(IBuffer buffer)
	{
	ByteBuffer byteBuffer = buffer.getByteBuffer();
	int correlationID = byteBuffer.getInt();
	if (TRACER.isEnabled())
	{
	TRACER.trace("Received buffer for correlation " + correlationID); //$NON-NLS-1$
	}

	Signal signal;
	boolean newSignalScheduled = false;

	synchronized (signals)
	{
	if (correlationID > 0)
	{
	// Incoming indication
	signal = signals.get(-correlationID);
	if (signal == null)
	{
	short signalID = byteBuffer.getShort();
	if (TRACER.isEnabled())
	{
	TRACER.trace("Got signalID: " + signalID); //$NON-NLS-1$
	}

	signal = provideSignalReactor(signalID);
	if (signal != null)
	{
	signal.setCorrelationID(-correlationID);
	signal.setBufferInputStream(new SignalInputStream(getTimeout()));
	if (signal instanceof IndicationWithResponse)
	{
	signal.setBufferOutputStream(new SignalOutputStream(-correlationID, signalID, false));
	}

	signals.put(-correlationID, signal);
	getExecutorService().execute(signal);
	newSignalScheduled = true;
	}
	}
	}
	else
	{
	// Incoming confirmation
	signal = signals.get(-correlationID);
	}
	}

	if (signal != null) // Can be null after timeout
	{
	if (newSignalScheduled)
	{
	fireSignalScheduledEvent(signal);
	}

	BufferInputStream inputStream = signal.getBufferInputStream();
	inputStream.handleBuffer(buffer);
	}
	else
	{
	if (TRACER.isEnabled())
	{
	TRACER.trace("Discarding buffer"); //$NON-NLS-1$
	}

	buffer.release();
	}
	}

	@Override
	public String toString()
	{
	IChannel channel = getChannel();
	if (channel != null)
	{
	return MessageFormat.format("SignalProtocol[{0}, {1}, {2}]", channel.getID(), channel.getLocation(), getType()); //$NON-NLS-1$
	}

	return MessageFormat.format("SignalProtocol[{0}]", getType()); //$NON-NLS-1$
	}

	@Override
	protected void doBeforeDeactivate() throws Exception
	{
	synchronized (signals)
	{
	// Wait at most 10 seconds for running signals to finish
	int waitMillis = 10 * 1000;
	long stop = System.currentTimeMillis() + waitMillis;
	while (!signals.isEmpty() && System.currentTimeMillis() < stop)
	{
	signals.wait(1000L);
	}
	}
	}

	@Override
	protected void doDeactivate() throws Exception
	{
	try
	{
	synchronized (signals)
	{
	signals.clear();
	}

	IChannel channel = getChannel();
	if (channel != null)
	{
	channel.close();
	setChannel(null);
	}
	}
	finally
	{
	super.doDeactivate();
	}
	}

	@Override
	protected void handleChannelDeactivation()
	{
	if (!failingOver)
	{
	super.handleChannelDeactivation();
	}
	}

	protected final SignalReactor provideSignalReactor(short signalID)
	{
	if (!isActive())
	{
	return null;
	}

	switch (signalID)
	{
	case SIGNAL_REMOTE_EXCEPTION:
	return new RemoteExceptionIndication(this);

	case SIGNAL_MONITOR_CANCELED:
	return new MonitorCanceledIndication(this);

	case SIGNAL_MONITOR_PROGRESS:
	return new MonitorProgressIndication(this);

	case SIGNAL_SET_TIMEOUT:
	return new SetTimeoutIndication(this);

	default:
	SignalReactor signal = createSignalReactor(signalID);
	if (signal == null)
	{
	throw new IllegalArgumentException("Invalid signalID " + signalID); //$NON-NLS-1$
	}

	return signal;
	}
	}

	/**
	* Returns a new signal instance to serve the given signal ID or <code>null</code> if the signal ID is invalid/unknown
	* for this protocol.
	*/
	protected SignalReactor createSignalReactor(short signalID)
	{
	return null;
	}

	/**
	* Returns <code>true</code> by default, override to change this behaviour.
	*
	* @since 4.1
	*/
	protected boolean isSendingTimeoutChanges()
	{
	return true;
	}

	synchronized int getNextCorrelationID()
	{
	int correlationID = nextCorrelationID;
	if (nextCorrelationID == MAX_CORRELATION_ID)
	{
	if (TRACER.isEnabled())
	{
	TRACER.trace("Correlation ID wrap-around"); //$NON-NLS-1$
	}

	nextCorrelationID = MIN_CORRELATION_ID;
	}
	else
	{
	++nextCorrelationID;
	}

	return correlationID;
	}

	InputStream wrapInputStream(InputStream in) throws IOException
	{
	if (streamWrapper != null)
	{
	in = streamWrapper.wrapInputStream(in);
	}

	return in;
	}

	OutputStream wrapOutputStream(OutputStream out) throws IOException
	{
	if (streamWrapper != null)
	{
	out = streamWrapper.wrapOutputStream(out);
	}

	return out;
	}

	void finishInputStream(InputStream in) throws IOException
	{
	if (streamWrapper != null)
	{
	streamWrapper.finishInputStream(in);
	}
	}

	void finishOutputStream(OutputStream out) throws IOException
	{
	if (streamWrapper != null)
	{
	streamWrapper.finishOutputStream(out);
	}
	}

	void startSignal(SignalActor signalActor, long timeout) throws Exception
	{
	checkArg(signalActor.getProtocol() == this, "Wrong protocol"); //$NON-NLS-1$
	short signalID = signalActor.getID();
	int correlationID = signalActor.getCorrelationID();
	signalActor.setBufferOutputStream(new SignalOutputStream(correlationID, signalID, true));
	if (signalActor instanceof RequestWithConfirmation<?>)
	{
	signalActor.setBufferInputStream(new SignalInputStream(timeout));
	}

	synchronized (signals)
	{
	signals.put(correlationID, signalActor);
	}

	fireSignalScheduledEvent(signalActor);
	signalActor.runSync();
	}

	void stopSignal(Signal signal, Exception exception)
	{
	int correlationID = signal.getCorrelationID();
	synchronized (signals)
	{
	signals.remove(correlationID);
	signals.notifyAll();
	}

	fireSignalFinishedEvent(signal, exception);
	}

	void handleRemoteException(int correlationID, Throwable t, boolean responding)
	{
	synchronized (signals)
	{
	Signal signal = signals.remove(correlationID);
	if (signal instanceof RequestWithConfirmation<?>)
	{
	RequestWithConfirmation<?> request = (RequestWithConfirmation<?>)signal;
	request.setRemoteException(t, responding);
	}

	signals.notifyAll();
	}
	}

	void handleMonitorProgress(int correlationID, double totalWork, double work)
	{
	synchronized (signals)
	{
	Signal signal = signals.get(correlationID);
	if (signal instanceof RequestWithMonitoring<?>)
	{
	RequestWithMonitoring<?> request = (RequestWithMonitoring<?>)signal;
	request.setMonitorProgress(totalWork, work);
	}
	}
	}

	void handleMonitorCanceled(int correlationID)
	{
	synchronized (signals)
	{
	Signal signal = signals.get(-correlationID);
	if (signal instanceof IndicationWithMonitoring)
	{
	IndicationWithMonitoring indication = (IndicationWithMonitoring)signal;
	indication.setMonitorCanceled();
	}
	}
	}

	void handleSetTimeOut(long timeout)
	{
	long oldTimeout = this.timeout;
	if (oldTimeout != timeout)
	{
	this.timeout = timeout;
	fireEvent(new TimeoutChangedEvent(this, oldTimeout, timeout));
	}
	}

	boolean sendSetTimeout()
	{
	boolean timeoutSent = false;
	if (isSendingTimeoutChanges())
	{
	try
	{
	timeoutSent = new SetTimeoutRequest(this, this.timeout).send();
	}
	catch (Exception ex)
	{
	throw WrappedException.wrap(ex);
	}
	}
	return timeoutSent;
	}

	private void fireSignalScheduledEvent(Signal signal)
	{
	IListener[] listeners = getListeners();
	if (listeners != null)
	{
	IEvent event = new SignalScheduledEvent<INFRA_STRUCTURE>(this, signal);
	fireEvent(event, listeners);
	}
	}

	private void fireSignalFinishedEvent(Signal signal, Exception exception)
	{
	IListener[] listeners = getListeners();
	if (listeners != null)
	{
	IEvent event = new SignalFinishedEvent<INFRA_STRUCTURE>(this, signal, exception);
	fireEvent(event, listeners);
	}
	}

	/**
	* An {@link IEvent event} fired from a {@link ISignalProtocol signal protocol} when the protocol {@link ISignalProtocol#setTimeout(long) timeout}
	* has been changed.
	*
	* @author Eike Stepper
	* @since 4.1
	*/
	public static final class TimeoutChangedEvent extends Event
	{
	private static final long serialVersionUID = 1L;

	private long oldTimeout;

	private long newTimeout;

	private TimeoutChangedEvent(ISignalProtocol<?> source, long oldTimeout, long newTimeout)
	{
	super(source);
	this.oldTimeout = oldTimeout;
	this.newTimeout = newTimeout;
	}

	@Override
	public SignalProtocol<?> getSource()
	{
	return (SignalProtocol<?>)super.getSource();
	}

	public long getOldTimeout()
	{
	return oldTimeout;
	}

	public long getNewTimeout()
	{
	return newTimeout;
	}

	@Override
	public String toString()
	{
	return "TimeoutChangedEvent [oldTimeout=" + oldTimeout + ", newTimeout=" + newTimeout + ", source=" + source
	+ "]";
	}

	}

	/**
	* @author Eike Stepper
	*/
	class SignalInputStream extends BufferInputStream
	{
	private long timeout;

	public SignalInputStream(long timeout)
	{
	this.timeout = timeout;
	}

	@Override
	public long getMillisBeforeTimeout()
	{
	return timeout;
	}

	@Override
	public int read() throws IOException
	{
	if (isCCAM())
	{
	final InternalChannel channel = (InternalChannel)getChannel();

	ExecutorService executorService = channel.getReceiveExecutor();
	executorService.submit(new Runnable()
	{
	public void run()
	{
	ConcurrencyUtil.sleep(500);
	channel.close();
	}
	});
	}

	return super.read();
	}
	}

	/**
	* @author Eike Stepper
	*/
	class SignalOutputStream extends ChannelOutputStream
	{
	public SignalOutputStream(final int correlationID, final short signalID, final boolean addSignalID)
	{
	super(getChannel(), new IBufferProvider()
	{
	private IBufferProvider delegate = getBufferProvider();

	private boolean firstBuffer = addSignalID;

	public short getBufferCapacity()
	{
	return delegate.getBufferCapacity();
	}

	public IBuffer provideBuffer()
	{
	IChannel channel = getChannel();
	if (channel == null)
	{
	throw new IORuntimeException("No channel for protocol " + SignalProtocol.this); //$NON-NLS-1$
	}

	IBuffer buffer = delegate.provideBuffer();
	ByteBuffer byteBuffer = buffer.startPutting(channel.getID());
	if (STREAM_TRACER.isEnabled())
	{
	STREAM_TRACER.trace("Providing buffer for correlation " + correlationID); //$NON-NLS-1$
	}

	byteBuffer.putInt(correlationID);
	if (firstBuffer)
	{
	if (SignalProtocol.TRACER.isEnabled())
	{
	STREAM_TRACER.trace("Put signal id " + signalID); //$NON-NLS-1$
	}

	byteBuffer.putShort(signalID);
	}

	firstBuffer = false;
	return buffer;
	}

	public void retainBuffer(IBuffer buffer)
	{
	delegate.retainBuffer(buffer);
	}
	});
	}
	}
	}