/home/tommi/tntnet/framework/common/tnt/job.h

00001 /*
00002  * Copyright (C) 2003-2005 Tommi Maekitalo
00003  * 
00004  * This library is free software; you can redistribute it and/or
00005  * modify it under the terms of the GNU Lesser General Public
00006  * License as published by the Free Software Foundation; either
00007  * version 2.1 of the License, or (at your option) any later version.
00008  * 
00009  * As a special exception, you may use this file as part of a free
00010  * software library without restriction. Specifically, if other files
00011  * instantiate templates or use macros or inline functions from this
00012  * file, or you compile this file and link it with other files to
00013  * produce an executable, this file does not by itself cause the
00014  * resulting executable to be covered by the GNU General Public
00015  * License. This exception does not however invalidate any other
00016  * reasons why the executable file might be covered by the GNU Library
00017  * General Public License.
00018  * 
00019  * This library is distributed in the hope that it will be useful,
00020  * but WITHOUT ANY WARRANTY; without even the implied warranty of
00021  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00022  * Lesser General Public License for more details.
00023  * 
00024  * You should have received a copy of the GNU Lesser General Public
00025  * License along with this library; if not, write to the Free Software
00026  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
00027  */
00028 
00029 
00030 #ifndef TNT_JOB_H
00031 #define TNT_JOB_H
00032 
00033 #include <deque>
00034 #include <tnt/httprequest.h>
00035 #include <tnt/httpparser.h>
00036 #include <cxxtools/mutex.h>
00037 #include <cxxtools/condition.h>
00038 #include <cxxtools/refcounted.h>
00039 #include <cxxtools/smartptr.h>
00040 
00066 namespace tnt
00067 {
00068   class Tntnet;
00069 
00071   class Job : public cxxtools::AtomicRefCounted
00072   {
00073       unsigned keepAliveCounter;
00074 
00075       HttpRequest request;
00076       HttpRequest::Parser parser;
00077       time_t lastAccessTime;
00078 
00079       static unsigned socket_read_timeout;
00080       static unsigned socket_write_timeout;
00081       static unsigned keepalive_max;
00082       static unsigned socket_buffer_size;
00083 
00084     public:
00085       explicit Job(Tntnet& app_, const SocketIf* socketIf_ = 0)
00086         : keepAliveCounter(keepalive_max),
00087           request(app_, socketIf_),
00088           parser(request),
00089           lastAccessTime(0)
00090         { }
00091       virtual ~Job();
00092 
00093       virtual std::iostream& getStream() = 0;
00094       virtual int getFd() const = 0;
00095       virtual void setRead() = 0;
00096       virtual void setWrite() = 0;
00097 
00098       HttpRequest& getRequest()         { return request; }
00099       HttpRequest::Parser& getParser()  { return parser; }
00100 
00101       unsigned decrementKeepAliveCounter()
00102         { return keepAliveCounter > 0 ? --keepAliveCounter : 0; }
00103       void clear();
00104       void touch()     { time(&lastAccessTime); }
00105       int msecToTimeout(time_t currentTime) const;
00106 
00107       static void setSocketReadTimeout(unsigned ms)     { socket_read_timeout = ms; }
00108       static void setSocketWriteTimeout(unsigned ms)    { socket_write_timeout = ms; }
00109       static void setKeepAliveMax(unsigned n)       { keepalive_max = n; }
00110       static void setSocketBufferSize(unsigned b)   { socket_buffer_size = b; }
00111 
00112       static unsigned getSocketReadTimeout()        { return socket_read_timeout; }
00113       static unsigned getSocketWriteTimeout()       { return socket_write_timeout; }
00114       static unsigned getKeepAliveTimeout();
00115       static unsigned getKeepAliveMax()       { return keepalive_max; }
00116       static unsigned getSocketBufferSize()   { return socket_buffer_size; }
00117   };
00118 
00120   class Jobqueue
00121   {
00122     public:
00123       typedef cxxtools::SmartPtr<Job> JobPtr;
00124 
00125       cxxtools::Condition noWaitThreads;
00126 
00127     private:
00128       std::deque<JobPtr> jobs;
00129       cxxtools::Mutex mutex;
00130       cxxtools::Condition notEmpty;
00131       cxxtools::Condition notFull;
00132       unsigned waitThreads;
00133       unsigned capacity;
00134 
00135     public:
00136       explicit Jobqueue(unsigned capacity_ = 1000)
00137         : waitThreads(0),
00138           capacity(capacity_)
00139         { }
00140 
00141       void put(JobPtr j, bool force = false);
00142       JobPtr get();
00143 
00144       void setCapacity(unsigned c)
00145         { capacity = c; }
00146       unsigned getCapacity() const
00147         { return capacity; }
00148       unsigned getWaitThreadCount() const
00149         { return waitThreads; }
00150       bool empty() const
00151         { return jobs.empty(); }
00152   };
00153 }
00154 
00155 #endif // TNT_JOB_H
00156