src/main/csharp/Transport/InactivityMonitor.cs - activemq-nms-openwire - Git at Google

 /*
  * 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.
  */

 using System;
 using System.Threading;
 using Apache.NMS.ActiveMQ.Commands;
 using Apache.NMS.ActiveMQ.Threads;
 using Apache.NMS.Util;

 namespace Apache.NMS.ActiveMQ.Transport
 {
 	/// <summary>
 	/// This class make sure that the connection is still alive,
 	/// by monitoring the reception of commands from the peer of
 	/// the transport.
 	/// </summary>
 	public class InactivityMonitor : TransportFilter
 	{
 		private readonly Atomic<bool> monitorStarted = new Atomic<bool>(false);

 		private readonly Atomic<bool> commandSent = new Atomic<bool>(false);
 		private readonly Atomic<bool> commandReceived = new Atomic<bool>(false);

 		private readonly Atomic<bool> failed = new Atomic<bool>(false);
 		private readonly Atomic<bool> inRead = new Atomic<bool>(false);
 		private readonly Atomic<bool> inWrite = new Atomic<bool>(false);

 		private CompositeTaskRunner asyncTasks;
 		private AsyncSignalReadErrorkTask asyncErrorTask;
 		private AsyncWriteTask asyncWriteTask;

 		private readonly Mutex monitor = new Mutex();

 		private Timer connectionCheckTimer;

 		private DateTime lastReadCheckTime;

 		private static int id = 0;
 		private readonly int instanceId = 0;
 		private bool disposing = false;

 		private long readCheckTime;
 		public long ReadCheckTime
 		{
 			get { return this.readCheckTime; }
 			set { this.readCheckTime = value; }
 		}

 		private long writeCheckTime;
 		public long WriteCheckTime
 		{
 			get { return this.writeCheckTime; }
 			set { this.writeCheckTime = value; }
 		}

 		private long initialDelayTime;
 		public long InitialDelayTime
 		{
 			get { return this.initialDelayTime; }
 			set { this.initialDelayTime = value; }
 		}

 		private readonly Atomic<bool> keepAliveResponseRequired = new Atomic<bool>(false);
 		public bool KeepAliveResponseRequired
 		{
 			get { return this.keepAliveResponseRequired.Value; }
 			set { keepAliveResponseRequired.Value = value; }
 		}

 		// Local and remote Wire Format Information
 		private WireFormatInfo localWireFormatInfo;
 		private WireFormatInfo remoteWireFormatInfo;

 		/// <summary>
 		/// Constructor or the Inactivity Monitor
 		/// </summary>
 		/// <param name="next"></param>
 		public InactivityMonitor(ITransport next)
 			: base(next)
 		{
 			this.instanceId = ++id;
 			Tracer.DebugFormat("Creating Inactivity Monitor: {0}", instanceId);
 		}

 		~InactivityMonitor()
 		{
 			Dispose(false);
 		}

 		protected override void Dispose(bool disposing)
 		{
 			if(disposing)
 			{
 				// get rid of unmanaged stuff
 			}

 			lock(monitor)
 			{
 				this.localWireFormatInfo = null;
 				this.remoteWireFormatInfo = null;
 				this.disposing = true;
 				StopMonitorThreads();
 			}

 			base.Dispose(disposing);
 		}

 		public void CheckConnection(object state)
 		{
 			// First see if we have written or can write.
 			WriteCheck();

 			// Now check is we've read anything, if not then we send
 			// a new KeepAlive with response required.
 			ReadCheck();
 		}

 		#region WriteCheck Related
 		/// <summary>
 		/// Check the write to the broker
 		/// </summary>
 		public void WriteCheck()
 		{
 			if(this.inWrite.Value || this.failed.Value)
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}]: is in write or already failed.", instanceId);
 				return;
 			}

 			CompositeTaskRunner taskRunner = this.asyncTasks;

 			if(!commandSent.Value)
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}]: No Message sent since last write check. Sending a KeepAliveInfo.", instanceId);
 				if(null != this.asyncWriteTask)
 				{
 					this.asyncWriteTask.IsPending = true;
 				}

 				if (this.monitorStarted.Value && taskRunner != null)
 				{
 					taskRunner.Wakeup();
 				}
 			}
 			else
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}]: Message sent since last write check. Resetting flag.", instanceId);
 			}

 			commandSent.Value = false;
 		}
 		#endregion

 		#region ReadCheck Related
 		public void ReadCheck()
 		{
 			DateTime now = DateTime.Now;
 			TimeSpan elapsed = now - this.lastReadCheckTime;
 			CompositeTaskRunner taskRunner = this.asyncTasks;

 			if(!AllowReadCheck(elapsed))
 			{
 				Tracer.Debug("InactivityMonitor[" + instanceId + "]: A read check is not currently allowed.");
 				return;
 			}

 			this.lastReadCheckTime = now;

 			if(this.inRead.Value || this.failed.Value)
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}]: A receive is in progress or already failed.", instanceId);
 				return;
 			}

 			if(!commandReceived.Value)
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}]: No message received since last read check! Sending an InactivityException!", instanceId);
 				if(null != this.asyncErrorTask)
 				{
 					this.asyncErrorTask.IsPending = true;
 				}

 				if (this.monitorStarted.Value && taskRunner != null)
 				{
 					taskRunner.Wakeup();
 				}
 			}
 			else
 			{
 				commandReceived.Value = false;
 			}
 		}

 		/// <summary>
 		/// Checks if we should allow the read check(if less than 90% of the read
 		/// check time elapsed then we dont do the readcheck
 		/// </summary>
 		/// <param name="elapsed"></param>
 		/// <returns></returns>
 		public bool AllowReadCheck(TimeSpan elapsed)
 		{
 			return (elapsed.TotalMilliseconds > (readCheckTime * 9 / 10));
 		}
 		#endregion

 		public override void Stop()
 		{
 			StopMonitorThreads();
 			next.Stop();
 		}

 		protected override void OnCommand(ITransport sender, Command command)
 		{
 			commandReceived.Value = true;
 			inRead.Value = true;
 			try
 			{
 				if(command.IsKeepAliveInfo)
 				{
 					KeepAliveInfo info = command as KeepAliveInfo;
 					if(info.ResponseRequired)
 					{
 						try
 						{
 							info.ResponseRequired = false;
 							Oneway(info);
 						}
 						catch(IOException ex)
 						{
 							OnException(this, ex);
 						}
 					}
 				}
 				else if(command.IsWireFormatInfo)
 				{
 					lock(monitor)
 					{
 						remoteWireFormatInfo = command as WireFormatInfo;
 						try
 						{
 							StartMonitorThreads();
 						}
 						catch(IOException ex)
 						{
 							OnException(this, ex);
 						}
 					}
 				}
 				base.OnCommand(sender, command);
 			}
 			finally
 			{
 				inRead.Value = false;
 			}
 		}

 		public override void Oneway(Command command)
 		{
 			// Disable inactivity monitoring while processing a command.
 			// synchronize this method - its not synchronized
 			// further down the transport stack and gets called by more
 			// than one thread  by this class
 			lock(inWrite)
 			{
 				inWrite.Value = true;
 				try
 				{
 					if(failed.Value)
 					{
 						throw new IOException("Channel was inactive for too long: " + next.RemoteAddress.ToString());
 					}
 					if(command.IsWireFormatInfo)
 					{
 						lock(monitor)
 						{
 							localWireFormatInfo = command as WireFormatInfo;
 							StartMonitorThreads();
 						}
 					}
 					next.Oneway(command);
 				}
 				finally
 				{
 					commandSent.Value = true;
 					inWrite.Value = false;
 				}
 			}
 		}

 		protected override void OnException(ITransport sender, Exception command)
 		{
 			if(failed.CompareAndSet(false, true) && !this.disposing)
 			{
 				Tracer.DebugFormat("Exception received in the Inactivity Monitor: {0}", command.Message);
 				StopMonitorThreads();
 				base.OnException(sender, command);
 			}
 		}

 		private void StartMonitorThreads()
 		{
 			lock(monitor)
 			{
 				if(this.IsDisposed || this.disposing)
 				{
 					return;
 				}

 				if(monitorStarted.Value)
 				{
 					return;
 				}

 				if(localWireFormatInfo == null)
 				{
 					return;
 				}

 				if(remoteWireFormatInfo == null)
 				{
 					return;
 				}

 				readCheckTime =
 					Math.Min(
 						localWireFormatInfo.MaxInactivityDuration,
 						remoteWireFormatInfo.MaxInactivityDuration);
 				initialDelayTime = remoteWireFormatInfo.MaxInactivityDurationInitialDelay > 0 ?
 					Math.Min(localWireFormatInfo.MaxInactivityDurationInitialDelay,
 						     remoteWireFormatInfo.MaxInactivityDurationInitialDelay) :
                     localWireFormatInfo.MaxInactivityDurationInitialDelay;

 				if(readCheckTime > 0)
 				{
 					Tracer.DebugFormat("InactivityMonitor[{0}]: Read Check time interval: {1}",
 								   instanceId, readCheckTime);
 					Tracer.DebugFormat("InactivityMonitor[{0}]: Initial Delay time interval: {1}",
 									   instanceId, initialDelayTime);

 					monitorStarted.Value = true;
 					this.asyncTasks = new CompositeTaskRunner("InactivityMonitor[" + instanceId + "].Runner");

 					this.asyncErrorTask = new AsyncSignalReadErrorkTask(this, next.RemoteAddress);
 					this.asyncWriteTask = new AsyncWriteTask(this);

 					this.asyncTasks.AddTask(this.asyncErrorTask);
 					this.asyncTasks.AddTask(this.asyncWriteTask);

 					writeCheckTime = readCheckTime > 3 ? readCheckTime / 3 : readCheckTime;

 					Tracer.DebugFormat("InactivityMonitor[{0}]: Write Check time interval: {1}",
 									   instanceId, writeCheckTime);

 					this.connectionCheckTimer = new Timer(
 						new TimerCallback(CheckConnection),
 						null,
 						initialDelayTime,
 						writeCheckTime);
 				}
 			}
 		}

 		private void StopMonitorThreads()
 		{
 			lock(monitor)
 			{
 				if(monitorStarted.CompareAndSet(true, false))
 				{
 					AutoResetEvent shutdownEvent = new AutoResetEvent(false);

 					if(null != connectionCheckTimer)
 					{
 						// Attempt to wait for the Timer to shutdown, but don't wait
 						// forever, if they don't shutdown after a few seconds, just quit.
 						this.connectionCheckTimer.Dispose(shutdownEvent);
 						if(!shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(5000), false))
 						{
 							Tracer.WarnFormat("InactivityMonitor[{0}]: Timer Task didn't shutdown properly.", instanceId);
 						}

 						this.connectionCheckTimer = null;
 					}

 					if(null != this.asyncTasks)
 					{
 						this.asyncTasks.RemoveTask(this.asyncWriteTask);
 						this.asyncTasks.RemoveTask(this.asyncErrorTask);
 						this.asyncTasks.Shutdown();
 						this.asyncTasks = null;
 					}

 					this.asyncWriteTask = null;
 					this.asyncErrorTask = null;
 				}
 			}

 			Tracer.DebugFormat("InactivityMonitor[{0}]: Stopped Monitor Threads.", instanceId);
 		}

 		#region Async Tasks
 		// Task that fires when the TaskRunner is signaled by the ReadCheck Timer Task.
 		class AsyncSignalReadErrorkTask : CompositeTask
 		{
 			private readonly InactivityMonitor parent;
 			private readonly Uri remote;
 			private readonly Atomic<bool> pending = new Atomic<bool>(false);

 			public AsyncSignalReadErrorkTask(InactivityMonitor parent, Uri remote)
 			{
 				this.parent = parent;
 				this.remote = remote;
 			}

 			public bool IsPending
 			{
 				get { return this.pending.Value; }
 				set { this.pending.Value = value; }
 			}

 			public bool Iterate()
 			{
 				if(this.pending.CompareAndSet(true, false) && this.parent.monitorStarted.Value)
 				{
 					IOException ex = new IOException("Channel was inactive for too long: " + remote);
 					this.parent.OnException(parent, ex);
 				}

 				return this.pending.Value;
 			}
 		}

 		// Task that fires when the TaskRunner is signaled by the WriteCheck Timer Task.
 		class AsyncWriteTask : CompositeTask
 		{
 			private readonly InactivityMonitor parent;
 			private readonly Atomic<bool> pending = new Atomic<bool>(false);

 			public AsyncWriteTask(InactivityMonitor parent)
 			{
 				this.parent = parent;
 			}

 			public bool IsPending
 			{
 				get { return this.pending.Value; }
 				set { this.pending.Value = value; }
 			}

 			public bool Iterate()
 			{
 				Tracer.DebugFormat("InactivityMonitor[{0}] preparing for another Write Check", parent.instanceId);
 				if(this.pending.CompareAndSet(true, false) && this.parent.monitorStarted.Value)
 				{
 					try
 					{
 						Tracer.DebugFormat("InactivityMonitor[{0}] Write Check required sending KeepAlive.",
 										   parent.instanceId);
 						KeepAliveInfo info = new KeepAliveInfo();
 						info.ResponseRequired = this.parent.keepAliveResponseRequired.Value;
 						this.parent.Oneway(info);
 					}
 					catch(IOException e)
 					{
 						this.parent.OnException(parent, e);
 					}
 				}

 				return this.pending.Value;
 			}
 		}
 		#endregion
 	}

 }
	/*
	* 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.
	*/

	using System;
	using System.Threading;
	using Apache.NMS.ActiveMQ.Commands;
	using Apache.NMS.ActiveMQ.Threads;
	using Apache.NMS.Util;

	namespace Apache.NMS.ActiveMQ.Transport
	{
	/// <summary>
	/// This class make sure that the connection is still alive,
	/// by monitoring the reception of commands from the peer of
	/// the transport.
	/// </summary>
	public class InactivityMonitor : TransportFilter
	{
	private readonly Atomic<bool> monitorStarted = new Atomic<bool>(false);

	private readonly Atomic<bool> commandSent = new Atomic<bool>(false);
	private readonly Atomic<bool> commandReceived = new Atomic<bool>(false);

	private readonly Atomic<bool> failed = new Atomic<bool>(false);
	private readonly Atomic<bool> inRead = new Atomic<bool>(false);
	private readonly Atomic<bool> inWrite = new Atomic<bool>(false);

	private CompositeTaskRunner asyncTasks;
	private AsyncSignalReadErrorkTask asyncErrorTask;
	private AsyncWriteTask asyncWriteTask;

	private readonly Mutex monitor = new Mutex();

	private Timer connectionCheckTimer;

	private DateTime lastReadCheckTime;

	private static int id = 0;
	private readonly int instanceId = 0;
	private bool disposing = false;

	private long readCheckTime;
	public long ReadCheckTime
	{
	get { return this.readCheckTime; }
	set { this.readCheckTime = value; }
	}

	private long writeCheckTime;
	public long WriteCheckTime
	{
	get { return this.writeCheckTime; }
	set { this.writeCheckTime = value; }
	}

	private long initialDelayTime;
	public long InitialDelayTime
	{
	get { return this.initialDelayTime; }
	set { this.initialDelayTime = value; }
	}

	private readonly Atomic<bool> keepAliveResponseRequired = new Atomic<bool>(false);
	public bool KeepAliveResponseRequired
	{
	get { return this.keepAliveResponseRequired.Value; }
	set { keepAliveResponseRequired.Value = value; }
	}

	// Local and remote Wire Format Information
	private WireFormatInfo localWireFormatInfo;
	private WireFormatInfo remoteWireFormatInfo;

	/// <summary>
	/// Constructor or the Inactivity Monitor
	/// </summary>
	/// <param name="next"></param>
	public InactivityMonitor(ITransport next)
	: base(next)
	{
	this.instanceId = ++id;
	Tracer.DebugFormat("Creating Inactivity Monitor: {0}", instanceId);
	}

	~InactivityMonitor()
	{
	Dispose(false);
	}

	protected override void Dispose(bool disposing)
	{
	if(disposing)
	{
	// get rid of unmanaged stuff
	}

	lock(monitor)
	{
	this.localWireFormatInfo = null;
	this.remoteWireFormatInfo = null;
	this.disposing = true;
	StopMonitorThreads();
	}

	base.Dispose(disposing);
	}

	public void CheckConnection(object state)
	{
	// First see if we have written or can write.
	WriteCheck();

	// Now check is we've read anything, if not then we send
	// a new KeepAlive with response required.
	ReadCheck();
	}

	#region WriteCheck Related
	/// <summary>
	/// Check the write to the broker
	/// </summary>
	public void WriteCheck()
	{
	if(this.inWrite.Value \|\| this.failed.Value)
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: is in write or already failed.", instanceId);
	return;
	}

	CompositeTaskRunner taskRunner = this.asyncTasks;

	if(!commandSent.Value)
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: No Message sent since last write check. Sending a KeepAliveInfo.", instanceId);
	if(null != this.asyncWriteTask)
	{
	this.asyncWriteTask.IsPending = true;
	}

	if (this.monitorStarted.Value && taskRunner != null)
	{
	taskRunner.Wakeup();
	}
	}
	else
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: Message sent since last write check. Resetting flag.", instanceId);
	}

	commandSent.Value = false;
	}
	#endregion

	#region ReadCheck Related
	public void ReadCheck()
	{
	DateTime now = DateTime.Now;
	TimeSpan elapsed = now - this.lastReadCheckTime;
	CompositeTaskRunner taskRunner = this.asyncTasks;

	if(!AllowReadCheck(elapsed))
	{
	Tracer.Debug("InactivityMonitor[" + instanceId + "]: A read check is not currently allowed.");
	return;
	}

	this.lastReadCheckTime = now;

	if(this.inRead.Value \|\| this.failed.Value)
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: A receive is in progress or already failed.", instanceId);
	return;
	}

	if(!commandReceived.Value)
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: No message received since last read check! Sending an InactivityException!", instanceId);
	if(null != this.asyncErrorTask)
	{
	this.asyncErrorTask.IsPending = true;
	}

	if (this.monitorStarted.Value && taskRunner != null)
	{
	taskRunner.Wakeup();
	}
	}
	else
	{
	commandReceived.Value = false;
	}
	}

	/// <summary>
	/// Checks if we should allow the read check(if less than 90% of the read
	/// check time elapsed then we dont do the readcheck
	/// </summary>
	/// <param name="elapsed"></param>
	/// <returns></returns>
	public bool AllowReadCheck(TimeSpan elapsed)
	{
	return (elapsed.TotalMilliseconds > (readCheckTime * 9 / 10));
	}
	#endregion

	public override void Stop()
	{
	StopMonitorThreads();
	next.Stop();
	}

	protected override void OnCommand(ITransport sender, Command command)
	{
	commandReceived.Value = true;
	inRead.Value = true;
	try
	{
	if(command.IsKeepAliveInfo)
	{
	KeepAliveInfo info = command as KeepAliveInfo;
	if(info.ResponseRequired)
	{
	try
	{
	info.ResponseRequired = false;
	Oneway(info);
	}
	catch(IOException ex)
	{
	OnException(this, ex);
	}
	}
	}
	else if(command.IsWireFormatInfo)
	{
	lock(monitor)
	{
	remoteWireFormatInfo = command as WireFormatInfo;
	try
	{
	StartMonitorThreads();
	}
	catch(IOException ex)
	{
	OnException(this, ex);
	}
	}
	}
	base.OnCommand(sender, command);
	}
	finally
	{
	inRead.Value = false;
	}
	}

	public override void Oneway(Command command)
	{
	// Disable inactivity monitoring while processing a command.
	// synchronize this method - its not synchronized
	// further down the transport stack and gets called by more
	// than one thread by this class
	lock(inWrite)
	{
	inWrite.Value = true;
	try
	{
	if(failed.Value)
	{
	throw new IOException("Channel was inactive for too long: " + next.RemoteAddress.ToString());
	}
	if(command.IsWireFormatInfo)
	{
	lock(monitor)
	{
	localWireFormatInfo = command as WireFormatInfo;
	StartMonitorThreads();
	}
	}
	next.Oneway(command);
	}
	finally
	{
	commandSent.Value = true;
	inWrite.Value = false;
	}
	}
	}

	protected override void OnException(ITransport sender, Exception command)
	{
	if(failed.CompareAndSet(false, true) && !this.disposing)
	{
	Tracer.DebugFormat("Exception received in the Inactivity Monitor: {0}", command.Message);
	StopMonitorThreads();
	base.OnException(sender, command);
	}
	}

	private void StartMonitorThreads()
	{
	lock(monitor)
	{
	if(this.IsDisposed \|\| this.disposing)
	{
	return;
	}

	if(monitorStarted.Value)
	{
	return;
	}

	if(localWireFormatInfo == null)
	{
	return;
	}

	if(remoteWireFormatInfo == null)
	{
	return;
	}

	readCheckTime =
	Math.Min(
	localWireFormatInfo.MaxInactivityDuration,
	remoteWireFormatInfo.MaxInactivityDuration);
	initialDelayTime = remoteWireFormatInfo.MaxInactivityDurationInitialDelay > 0 ?
	Math.Min(localWireFormatInfo.MaxInactivityDurationInitialDelay,
	remoteWireFormatInfo.MaxInactivityDurationInitialDelay) :
	localWireFormatInfo.MaxInactivityDurationInitialDelay;

	if(readCheckTime > 0)
	{
	Tracer.DebugFormat("InactivityMonitor[{0}]: Read Check time interval: {1}",
	instanceId, readCheckTime);
	Tracer.DebugFormat("InactivityMonitor[{0}]: Initial Delay time interval: {1}",
	instanceId, initialDelayTime);

	monitorStarted.Value = true;
	this.asyncTasks = new CompositeTaskRunner("InactivityMonitor[" + instanceId + "].Runner");

	this.asyncErrorTask = new AsyncSignalReadErrorkTask(this, next.RemoteAddress);
	this.asyncWriteTask = new AsyncWriteTask(this);

	this.asyncTasks.AddTask(this.asyncErrorTask);
	this.asyncTasks.AddTask(this.asyncWriteTask);

	writeCheckTime = readCheckTime > 3 ? readCheckTime / 3 : readCheckTime;

	Tracer.DebugFormat("InactivityMonitor[{0}]: Write Check time interval: {1}",
	instanceId, writeCheckTime);

	this.connectionCheckTimer = new Timer(
	new TimerCallback(CheckConnection),
	null,
	initialDelayTime,
	writeCheckTime);
	}
	}
	}

	private void StopMonitorThreads()
	{
	lock(monitor)
	{
	if(monitorStarted.CompareAndSet(true, false))
	{
	AutoResetEvent shutdownEvent = new AutoResetEvent(false);

	if(null != connectionCheckTimer)
	{
	// Attempt to wait for the Timer to shutdown, but don't wait
	// forever, if they don't shutdown after a few seconds, just quit.
	this.connectionCheckTimer.Dispose(shutdownEvent);
	if(!shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(5000), false))
	{
	Tracer.WarnFormat("InactivityMonitor[{0}]: Timer Task didn't shutdown properly.", instanceId);
	}

	this.connectionCheckTimer = null;
	}

	if(null != this.asyncTasks)
	{
	this.asyncTasks.RemoveTask(this.asyncWriteTask);
	this.asyncTasks.RemoveTask(this.asyncErrorTask);
	this.asyncTasks.Shutdown();
	this.asyncTasks = null;
	}

	this.asyncWriteTask = null;
	this.asyncErrorTask = null;
	}
	}

	Tracer.DebugFormat("InactivityMonitor[{0}]: Stopped Monitor Threads.", instanceId);
	}

	#region Async Tasks
	// Task that fires when the TaskRunner is signaled by the ReadCheck Timer Task.
	class AsyncSignalReadErrorkTask : CompositeTask
	{
	private readonly InactivityMonitor parent;
	private readonly Uri remote;
	private readonly Atomic<bool> pending = new Atomic<bool>(false);

	public AsyncSignalReadErrorkTask(InactivityMonitor parent, Uri remote)
	{
	this.parent = parent;
	this.remote = remote;
	}

	public bool IsPending
	{
	get { return this.pending.Value; }
	set { this.pending.Value = value; }
	}

	public bool Iterate()
	{
	if(this.pending.CompareAndSet(true, false) && this.parent.monitorStarted.Value)
	{
	IOException ex = new IOException("Channel was inactive for too long: " + remote);
	this.parent.OnException(parent, ex);
	}

	return this.pending.Value;
	}
	}

	// Task that fires when the TaskRunner is signaled by the WriteCheck Timer Task.
	class AsyncWriteTask : CompositeTask
	{
	private readonly InactivityMonitor parent;
	private readonly Atomic<bool> pending = new Atomic<bool>(false);

	public AsyncWriteTask(InactivityMonitor parent)
	{
	this.parent = parent;
	}

	public bool IsPending
	{
	get { return this.pending.Value; }
	set { this.pending.Value = value; }
	}

	public bool Iterate()
	{
	Tracer.DebugFormat("InactivityMonitor[{0}] preparing for another Write Check", parent.instanceId);
	if(this.pending.CompareAndSet(true, false) && this.parent.monitorStarted.Value)
	{
	try
	{
	Tracer.DebugFormat("InactivityMonitor[{0}] Write Check required sending KeepAlive.",
	parent.instanceId);
	KeepAliveInfo info = new KeepAliveInfo();
	info.ResponseRequired = this.parent.keepAliveResponseRequired.Value;
	this.parent.Oneway(info);
	}
	catch(IOException e)
	{
	this.parent.OnException(parent, e);
	}
	}

	return this.pending.Value;
	}
	}
	#endregion
	}

	}