← Back to branch summary

~ubuntu-branches/ubuntu/trusty/vdr-plugin-live/trusty-proposed

« back to all changes in this revision

Viewing changes to httpd/tnt/job.h

  • Committer: Bazaar Package Importer
  • Author(s): Thomas Schmidt
  • Date: 2007-07-02 21:02:17 UTC
  • Revision ID: james.westby@ubuntu.com-20070702210217-d34t69xf1qosqgvc
Tags: upstream-0.1.0
Import upstream version 0.1.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
httpd/tnt/job.h
 
1
/* tnt/job.h
 
2
 * Copyright (C) 2003-2005 Tommi Maekitalo
 
3
 *
 
4
 * This program is free software; you can redistribute it and/or
 
5
 * modify it under the terms of the GNU General Public License as
 
6
 * published by the Free Software Foundation; either version 2 of the
 
7
 * License, or (at your option) any later version.
 
8
 *
 
9
 * This program is distributed in the hope that it will be useful, but
 
10
 * is provided AS IS, WITHOUT ANY WARRANTY; without even the implied
 
11
 * warranty of MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, and
 
12
 * NON-INFRINGEMENT.  See the GNU General Public License for more details.
 
13
 *
 
14
 * You should have received a copy of the GNU General Public License
 
15
 * along with this program; if not, write to the Free Software
 
16
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
 
17
 *
 
18
 */
 
19
 
 
20
#ifndef TNT_JOB_H
 
21
#define TNT_JOB_H
 
22
 
 
23
#ifdef HAVE_CONFIG_H
 
24
#       include <config.h>
 
25
#endif
 
26
#include <deque>
 
27
#include <cxxtools/thread.h>
 
28
#include <cxxtools/tcpstream.h>
 
29
#include <tnt/httprequest.h>
 
30
#include <tnt/httpparser.h>
 
31
#include <tnt/pointer.h>
 
32
#include <time.h>
 
33
#include "tnt/ssl.h"
 
34
 
 
35
/**
 
36
// in tntnet (mainthread):
 
37
Jobqueue queue;
 
38
void mainloop()
 
39
{
 
40
  while (1)
 
41
  {
 
42
    Jobqueue::JobPtr j = new Tcpjob();
 
43
    j->accept(poller.get());
 
44
    queue.put(j);
 
45
  }
 
46
}
 
47
 
 
48
// in server (workerthread):
 
49
void Server::run()
 
50
{
 
51
  while (1)
 
52
  {
 
53
    Jobqueue::JobPtr j = queue.get();
 
54
    std::iostream& socket = j->getStream();
 
55
    processRequest(socket);
 
56
  }
 
57
}
 
58
*/
 
59
 
 
60
namespace tnt
 
61
{
 
62
  /** Job - one per request */
 
63
  class Job
 
64
  {
 
65
      unsigned keepAliveCounter;
 
66
 
 
67
      HttpRequest request;
 
68
      HttpMessage::Parser parser;
 
69
      time_t lastAccessTime;
 
70
 
 
71
      unsigned refs;
 
72
 
 
73
      static unsigned socket_read_timeout;
 
74
      static unsigned socket_write_timeout;
 
75
      static unsigned keepalive_max;
 
76
      static unsigned socket_buffer_size;
 
77
 
 
78
    public:
 
79
      Job()
 
80
        : keepAliveCounter(keepalive_max),
 
81
          parser(request),
 
82
          lastAccessTime(0),
 
83
          refs(0)
 
84
        { }
 
85
 
 
86
    protected:
 
87
      virtual ~Job();
 
88
 
 
89
    public:
 
90
      unsigned addRef()   { return ++refs; }
 
91
      unsigned release()
 
92
      {
 
93
        if (--refs == 0)
 
94
        {
 
95
          delete this;
 
96
          return 0;
 
97
        }
 
98
        else
 
99
          return refs;
 
100
      }
 
101
 
 
102
      virtual std::iostream& getStream() = 0;
 
103
      virtual int getFd() const = 0;
 
104
      virtual void setRead() = 0;
 
105
      virtual void setWrite() = 0;
 
106
 
 
107
      HttpRequest& getRequest()         { return request; }
 
108
      HttpMessage::Parser& getParser()  { return parser; }
 
109
 
 
110
      unsigned decrementKeepAliveCounter()
 
111
        { return keepAliveCounter > 0 ? --keepAliveCounter : 0; }
 
112
      void clear();
 
113
      void touch()     { time(&lastAccessTime); }
 
114
      int msecToTimeout(time_t currentTime) const;
 
115
 
 
116
      static void setSocketReadTimeout(unsigned ms)     { socket_read_timeout = ms; }
 
117
      static void setSocketWriteTimeout(unsigned ms)    { socket_write_timeout = ms; }
 
118
      static void setKeepAliveMax(unsigned n)       { keepalive_max = n; }
 
119
      static void setSocketBufferSize(unsigned b)   { socket_buffer_size = b; }
 
120
 
 
121
      static unsigned getSocketReadTimeout()        { return socket_read_timeout; }
 
122
      static unsigned getSocketWriteTimeout()       { return socket_write_timeout; }
 
123
      static unsigned getKeepAliveTimeout();
 
124
      static unsigned getKeepAliveMax()       { return keepalive_max; }
 
125
      static unsigned getSocketBufferSize()   { return socket_buffer_size; }
 
126
  };
 
127
 
 
128
  class Tcpjob : public Job
 
129
  {
 
130
      cxxtools::net::iostream socket;
 
131
 
 
132
    public:
 
133
      Tcpjob()
 
134
        : socket(getSocketBufferSize(), getSocketReadTimeout())
 
135
        { }
 
136
 
 
137
      void accept(const cxxtools::net::Server& listener);
 
138
 
 
139
      std::iostream& getStream();
 
140
      int getFd() const;
 
141
      void setRead();
 
142
      void setWrite();
 
143
  };
 
144
 
 
145
#ifdef USE_SSL
 
146
  class SslTcpjob : public Job
 
147
  {
 
148
      ssl_iostream socket;
 
149
 
 
150
    public:
 
151
      SslTcpjob()
 
152
        : socket(getSocketBufferSize(), getSocketReadTimeout())
 
153
        { }
 
154
 
 
155
      void accept(const SslServer& listener);
 
156
 
 
157
      std::iostream& getStream();
 
158
      int getFd() const;
 
159
      void setRead();
 
160
      void setWrite();
 
161
  };
 
162
#endif // USE_SSL
 
163
 
 
164
  /** Jobqueue - one per process */
 
165
  class Jobqueue
 
166
  {
 
167
    public:
 
168
      typedef Pointer<Job> JobPtr;
 
169
 
 
170
      cxxtools::Condition noWaitThreads;
 
171
 
 
172
    private:
 
173
      std::deque<JobPtr> jobs;
 
174
      cxxtools::Mutex mutex;
 
175
      cxxtools::Condition notEmpty;
 
176
      cxxtools::Condition notFull;
 
177
      unsigned waitThreads;
 
178
      unsigned capacity;
 
179
 
 
180
    public:
 
181
      explicit Jobqueue(unsigned capacity_)
 
182
        : waitThreads(0),
 
183
          capacity(capacity_)
 
184
        { }
 
185
 
 
186
      void put(JobPtr j);
 
187
      JobPtr get();
 
188
 
 
189
      void setCapacity(unsigned c)
 
190
        { capacity = c; }
 
191
      unsigned getCapacity() const
 
192
        { return capacity; }
 
193
      unsigned getWaitThreadCount() const
 
194
        { return waitThreads; }
 
195
      bool empty() const
 
196
        { return jobs.empty(); }
 
197
  };
 
198
}
 
199
 
 
200
#endif // TNT_JOB_H
 
201