Subversion Repositories SmartDukaan

/**

 * 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.Collections.Generic;

using System.Threading;

using Thrift.Collections;

using Thrift.Protocol;

using Thrift.Transport;

namespace Thrift.Server

{

	/// <summary>

	/// Server that uses C# threads (as opposed to the ThreadPool) when handling requests

	/// </summary>

	public class TThreadedServer : TServer

	{

		private const int DEFAULT_MAX_THREADS = 100;

		private volatile bool stop = false;

		private readonly int maxThreads;

		private Queue<TTransport> clientQueue;

		private THashSet<Thread> clientThreads;

		private object clientLock;

		private Thread workerThread;

		public TThreadedServer(TProcessor processor, TServerTransport serverTransport)

			: this(processor, serverTransport,

				 new TTransportFactory(), new TTransportFactory(),

				 new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),

				 DEFAULT_MAX_THREADS, DefaultLogDelegate)

		{

		}

		public TThreadedServer(TProcessor processor, TServerTransport serverTransport, LogDelegate logDelegate)

			: this(processor, serverTransport,

				 new TTransportFactory(), new TTransportFactory(),

				 new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),

				 DEFAULT_MAX_THREADS, logDelegate)

		{

		}

		public TThreadedServer(TProcessor processor,

								 TServerTransport serverTransport,

								 TTransportFactory transportFactory,

								 TProtocolFactory protocolFactory)

			: this(processor, serverTransport,

				 transportFactory, transportFactory,

				 protocolFactory, protocolFactory,

				 DEFAULT_MAX_THREADS, DefaultLogDelegate)

		{

		}

		public TThreadedServer(TProcessor processor,

								 TServerTransport serverTransport,

								 TTransportFactory inputTransportFactory,

								 TTransportFactory outputTransportFactory,

								 TProtocolFactory inputProtocolFactory,

								 TProtocolFactory outputProtocolFactory,

								 int maxThreads, LogDelegate logDel)

			: base(processor, serverTransport, inputTransportFactory, outputTransportFactory,

				  inputProtocolFactory, outputProtocolFactory, logDel)

		{

			this.maxThreads = maxThreads;

			clientQueue = new Queue<TTransport>();

			clientLock = new object();

			clientThreads = new THashSet<Thread>();

		}

		/// <summary>

		/// Use new Thread for each new client connection. block until numConnections < maxTHreads

		/// </summary>

		public override void Serve()

		{

			try

			{

				//start worker thread

				workerThread = new Thread(new ThreadStart(Execute));

				workerThread.Start();

				serverTransport.Listen();

			}

			catch (TTransportException ttx)

			{

				logDelegate("Error, could not listen on ServerTransport: " + ttx);

				return;

			}

			while (!stop)

			{

				int failureCount = 0;

				try

				{

					TTransport client = serverTransport.Accept();

					lock (clientLock)

					{

						clientQueue.Enqueue(client);

						Monitor.Pulse(clientLock);

					}

				}

				catch (TTransportException ttx)

				{

					if (stop)

					{

						logDelegate("TThreadPoolServer was shutting down, caught " + ttx);

					}

					else

					{

						++failureCount;

						logDelegate(ttx.ToString());

					}

				}

			}

			if (stop)

			{

				try

				{

					serverTransport.Close();

				}

				catch (TTransportException ttx)

				{

					logDelegate("TServeTransport failed on close: " + ttx.Message);

				}

				stop = false;

			}

		}

		/// <summary>

		/// Loops on processing a client forever

		/// threadContext will be a TTransport instance

		/// </summary>

		/// <param name="threadContext"></param>

		private void Execute()

		{

			while (!stop)

			{

				TTransport client;

				Thread t;

				lock (clientLock)

				{

					//don't dequeue if too many connections

					while (clientThreads.Count >= maxThreads)

					{

						Monitor.Wait(clientLock);

					}

					while (clientQueue.Count == 0)

					{

						Monitor.Wait(clientLock);

					}

					client = clientQueue.Dequeue();

					t = new Thread(new ParameterizedThreadStart(ClientWorker));

					clientThreads.Add(t);

				}

				//start processing requests from client on new thread

				t.Start(client);

			}

		}

		private void ClientWorker(Object context)

		{

			TTransport client = (TTransport)context;

			TTransport inputTransport = null;

			TTransport outputTransport = null;

			TProtocol inputProtocol = null;

			TProtocol outputProtocol = null;

			try

			{

				inputTransport = inputTransportFactory.GetTransport(client);

				outputTransport = outputTransportFactory.GetTransport(client);

				inputProtocol = inputProtocolFactory.GetProtocol(inputTransport);

				outputProtocol = outputProtocolFactory.GetProtocol(outputTransport);

				while (processor.Process(inputProtocol, outputProtocol))

				{

					//keep processing requests until client disconnects

				}

			}

			catch (TTransportException)

			{

			}

			catch (Exception x)

			{

				logDelegate("Error: " + x);

			}

			if (inputTransport != null)

			{

				inputTransport.Close();

			}

			if (outputTransport != null)

			{

				outputTransport.Close();

			}

			lock (clientLock)

			{

				clientThreads.Remove(Thread.CurrentThread);

				Monitor.Pulse(clientLock);

			}

			return;

		}

		public override void Stop()

		{

			stop = true;

			serverTransport.Close();

			//clean up all the threads myself

			workerThread.Abort();

			foreach (Thread t in clientThreads)

			{

				t.Abort();

			}

		}

	}

}