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();
                        }
                }
        }
}