From 7cfba39a23628cea4ffeb451f7bfed74ad5330ad Mon Sep 17 00:00:00 2001
From: Hannes Matuschek <hmatuschek@gmail.com>
Date: Sun, 14 Jun 2015 16:57:11 +0200
Subject: [PATCH] Fixed FM demod and http server.

---
 cmd/CMakeLists.txt     |   2 +-
 cmd/aprsapplication.cc |  87 ++++++++++++++++
 cmd/aprsapplication.hh |  29 ++++++
 cmd/main.cc            |  23 +----
 cmd/shared/index.html  |  67 ++++++------
 src/aprs.cc            |   8 ++
 src/demod.hh           |  26 ++---
 src/http.cc            | 226 +++++++++++++++++++++++++----------------
 src/http.hh            | 150 +++++++++++++++++----------
 9 files changed, 404 insertions(+), 214 deletions(-)
 create mode 100644 cmd/aprsapplication.cc
 create mode 100644 cmd/aprsapplication.hh

diff --git a/cmd/CMakeLists.txt b/cmd/CMakeLists.txt
index bb11808..55ca216 100644
--- a/cmd/CMakeLists.txt
+++ b/cmd/CMakeLists.txt
@@ -3,6 +3,6 @@
 INCLUDE_DIRECTORIES(${CMAKE_CURRENT_BINARY_DIR})
 link_resources(sdr_cmd_resources shared/index.html)
 
-add_executable(sdr_cmd main.cc)
+add_executable(sdr_cmd main.cc aprsapplication.cc)
 add_dependencies(sdr_cmd sdr_cmd_resources)
 target_link_libraries(sdr_cmd ${LIBS} libsdr )
diff --git a/cmd/aprsapplication.cc b/cmd/aprsapplication.cc
new file mode 100644
index 0000000..5b6f363
--- /dev/null
+++ b/cmd/aprsapplication.cc
@@ -0,0 +1,87 @@
+#include "aprsapplication.hh"
+
+// This file will be generated at build time
+// and contains the static content served.
+#include "sdr_cmd_resources.hh"
+
+
+using namespace sdr;
+
+APRSApplication::APRSApplication(http::Server &server)
+  : APRS(), _server(server), _messages()
+{
+  // Register callbacks
+  server.addStatic("/", std::string(index_html, index_html_size), "text/html");
+  server.addJSON("/spots", this, &APRSApplication::spots);
+  server.addHandler("/update", this, &APRSApplication::update);
+}
+
+APRSApplication::~APRSApplication() {
+  // pass...
+}
+
+
+bool
+APRSApplication::spots(const http::JSON &request, http::JSON &response) {
+  std::list<http::JSON> msg_list;
+  for (std::list<Message>::iterator msg = _messages.begin(); msg != _messages.end(); msg++) {
+    std::map<std::string, http::JSON> message;
+    message["call"] = http::JSON(msg->from().call());
+    if (msg->hasLocation()) {
+      message["lat"] = http::JSON(msg->latitude());
+      message["lon"] = http::JSON(msg->longitude());
+    }
+    time_t time = msg->time();
+    message["time"] = http::JSON(ctime(&time));
+    msg_list.push_back(message);
+  }
+  response = http::JSON(msg_list);
+  return true;
+}
+
+void
+APRSApplication::handleAPRSMessage(const Message &message) {
+  _messages.push_back(message);
+  // Serialize JSON message
+  std::string json_text;
+  if (_clients.size()) {
+    std::map<std::string, http::JSON> msg;
+    msg["call"] = http::JSON(message.from().call());
+    if (message.hasLocation()) {
+      msg["lat"] = http::JSON(message.latitude());
+      msg["lon"] = http::JSON(message.longitude());
+    }
+    time_t time = message.time();
+    msg["time"] = http::JSON(ctime(&time));
+    http::JSON(msg).serialize(json_text);
+  }
+  // a list collecting the closed
+  // signal all clients connected
+  std::list<http::Connection>::iterator client = _clients.begin();
+  while (client != _clients.end()) {
+    if (client->isClosed()) {
+      // remove client from list
+      client = _clients.erase(client);
+    } else {
+      // send event
+      client->send("data: ");
+      client->send(json_text);
+      client->send("\n\n");
+    }
+  }
+}
+
+void
+APRSApplication::update(const http::Request &request, http::Response &response) {
+  // This call back implements a server side event stream, means the response will
+  // be blocked until the connection is closed
+  response.setHeader("Content-Type", "text/event-stream");
+  response.setHeader("Cache-Control", "no-cache");
+  response.setStatus(http::Response::STATUS_OK);
+  response.sendHeaders();
+  // Signal connection thread to exit without closing the connection
+  response.connection().setProtocolUpgrade();
+  // Store connection
+  _clients.push_back(response.connection());
+}
+
diff --git a/cmd/aprsapplication.hh b/cmd/aprsapplication.hh
new file mode 100644
index 0000000..f0c2c2b
--- /dev/null
+++ b/cmd/aprsapplication.hh
@@ -0,0 +1,29 @@
+#ifndef __SDR_APRS_APRSAPPLICATION_HH__
+#define __SDR_APRS_APRSAPPLICATION_HH__
+
+#include "http.hh"
+#include "aprs.hh"
+
+
+namespace sdr {
+
+class APRSApplication: public APRS
+{
+public:
+  APRSApplication(http::Server &server);
+  ~APRSApplication();
+
+  bool spots(const http::JSON &request, http::JSON &response);
+  void update(const http::Request &request, http::Response &response);
+
+  void handleAPRSMessage(const Message &message);
+
+protected:
+  http::Server      &_server;
+  std::list<Message> _messages;
+  std::list<http::Connection> _clients;
+};
+
+}
+
+#endif // APRSAPPLICATION_HH
diff --git a/cmd/main.cc b/cmd/main.cc
index 1b2ce87..e46c7d2 100644
--- a/cmd/main.cc
+++ b/cmd/main.cc
@@ -1,36 +1,26 @@
 #include <stdio.h>
 
 #include "queue.hh"
-#include "http.hh"
 #include "logger.hh"
 
 #include <iostream>
 #include <csignal>
-#include "sdr_cmd_resources.hh"
+
+#include "aprsapplication.hh"
 
 using namespace sdr;
 
-class Application
-{
-public:
-  Application() {}
-
-  bool echo(const http::JSON &request, http::JSON &response) {
-    response = request;
-    return true;
-  }
-};
-
 
 static http::Server *server = 0;
 
 static void __sigint_handler(int signo) {
-  server->stop(true);
+  if (server) { server->stop(true); }
 }
 
+
 int main(int argc, char *argv[]) {
-  Application app;
   server = new http::Server(8080);
+  APRSApplication app(*server);
 
   // Install log handler
   sdr::Logger::get().addHandler(
@@ -39,9 +29,6 @@ int main(int argc, char *argv[]) {
   // Register signal handler:
   signal(SIGINT, __sigint_handler);
 
-  // Register callbacks
-  server->addStatic("/", std::string(index_html, index_html_size), "text/html");
-  server->addJSON("/echo", &app, &Application::echo);
   // start server
   server->start(true);
 
diff --git a/cmd/shared/index.html b/cmd/shared/index.html
index f94428c..c9e3e0b 100644
--- a/cmd/shared/index.html
+++ b/cmd/shared/index.html
@@ -13,43 +13,27 @@
   <script src="https://maps.googleapis.com/maps/api/js?v=3.exp"></script>
 
   <script>
-var map;
-var connections = {};
-var qth = null;
+var map = null;
+var spots = {};
+var source = null;
 
-function onSetQTH (lon, lat) {
-  // Center map at QTH
-  map.setCenter(new google.maps.LatLng(lat, lon));
-  // Add or update the marker at QTH
-  if (null == qth) {
-    qth = new google.maps.Marker({
-      position: new google.maps.LatLng(lat, lon),
-      title: "QTH", map: map});
+function onAddSpot (callsign, lon, lat) {
+  // Check if connection exists
+  if (callsign in spots) {
+    // If connection exists, update SNR
+    connections[spots].marker.setPosition(new google.maps.LatLng(lat, lon));
   } else {
-    qth.setPosition(new google.maps.LatLng(lat, lon));
+    // otherwise, add marker at location with label callsign
+    var marker = new google.maps.Marker({
+       position: new google.maps.LatLng(lat, lon), title: callsign, map: map});
+    spots[callsign] = { marker: marker };
   }
 }
 
-function onAddConnection (callsign, lon, lat, snr) {
-  // Check if connection exists
-  if (callsign in connections) {
-    // If connection exists, update SNR
-    connections[callsign].line.setOptions(options={
-      clickable: false, draggable: false, geodesic: true,
-      path: [qth.getPosition(), new google.maps.LatLng(lat, lon)],
-      strokeColor: '#0000FF', strokeOpacity: (1.0 + Math.min(0, snr/40)), strokeWeight: 2 } );
-    connections[callsign].marker.setPosition(new google.maps.LatLng(lat, lon));
-  } else {
-    // otherwise, add marker at location with label callsign and line from QTH to location
-    var marker = new google.maps.Marker({
-       position: new google.maps.LatLng(lat, lon), title: callsign, map: map});
-    var line = new google.maps.Polyline({
-      clickable: false, draggable: false, geodesic: true,
-      path: [qth.getPosition(), new google.maps.LatLng(lat, lon)],
-      strokeColor: '#0000FF', strokeOpacity: (1.0 + Math.min(0, snr/40)), strokeWeight: 2 });
-    line.setMap(map);
-    connections[callsign] = { marker: marker, line: line };
-  }
+function updateHandler(e) {
+  var msg = JSON.parse(e.data);
+  console.log("RX: " + msg)
+  onAddSpot(msg.call, msg.lon, msg.lat);
 }
 
 function initialize() {
@@ -57,6 +41,25 @@ function initialize() {
   map = new google.maps.Map(
     document.getElementById('map-canvas'),
     {zoom:2, center: new google.maps.LatLng(0,0), streetViewControl:false});
+
+  // Request all spots
+  var xmlhttp = new XMLHttpRequest();
+  var url = "spots";
+  xmlhttp.onreadystatechange = function() {
+    if (xmlhttp.readyState != 4 && xmlhttp.status != 200) { return; }
+    var spots = JSON.parse(xmlhttp.responseText);
+    for (spot in spots) { onAddSpot(spot); }
+  }
+  xmlhttp.open("GET", url, true);
+  xmlhttp.send();
+
+  // Now, connect to event source for updates
+  if (!!window.EventSource) {
+    source = new window.EventSource("update");
+    source.addEventListener('message', updateHandler);
+  } else {
+    alert("Your browser does not support the EventSource -> no live update.")
+  }
 }
 
 google.maps.event.addDomListener(window, 'load', initialize);
diff --git a/src/aprs.cc b/src/aprs.cc
index 242255d..6f0376f 100644
--- a/src/aprs.cc
+++ b/src/aprs.cc
@@ -149,6 +149,14 @@ APRS::Message::Message(const AX25::Message &msg)
   }
 }
 
+APRS::Message::Message(const Message &msg)
+  : AX25::Message(msg), _hasLocation(msg._hasLocation), _latitude(msg._latitude),
+    _longitude(msg._longitude), _symbol(msg._symbol), _hasTime(msg._hasTime),
+    _time(msg._hasTime)
+{
+  // pass...
+}
+
 
 bool
 APRS::Message::_readLocation(size_t &offset) {
diff --git a/src/demod.hh b/src/demod.hh
index 16f59f0..3b1bc59 100644
--- a/src/demod.hh
+++ b/src/demod.hh
@@ -241,26 +241,12 @@ protected:
   /** The actual demodulation. */
   void _process(const Buffer< std::complex<iScalar> > &in, const Buffer<oScalar> &out)
   {
-    // calc first value
-    SScalar a = (SScalar(in[0].real())*SScalar(_last_value.real()))/2
-        + (SScalar(in[0].imag())*SScalar(_last_value.imag()))/2;
-    SScalar b = (SScalar(in[0].imag())*SScalar(_last_value.real()))/2
-        - (SScalar(in[0].real())*SScalar(_last_value.imag()))/2;
-    a >>= Traits<iScalar>::shift; b >>= Traits<iScalar>::shift;
-    // update last value
-    _last_value = in[0];
-    // calc output (prob. overwriting the last value)
-    out[0] = fast_atan2<iScalar, oScalar>(a, b);
-
     // Calc remaining values
     for (size_t i=1; i<in.size(); i++) {
-      a = (SScalar(in[i].real())*SScalar(_last_value.real()))/2
-          + (SScalar(in[i].imag())*SScalar(_last_value.imag()))/2;
-      b = (SScalar(in[i].imag())*SScalar(_last_value.real()))/2
-          - (SScalar(in[i].real())*SScalar(_last_value.imag()))/2;
-      a >>= Traits<iScalar>::shift; b >>= Traits<iScalar>::shift;
-      _last_value = in[i];
-      out[i] = fast_atan2<iScalar,oScalar>(a, b);
+      oScalar phi = fast_atan2<iScalar, oScalar>(in[i].real(), in[i].imag())/2;
+      // dphi
+      out[i] = (_last_value - phi);
+      _last_value = phi;
     }
 
     // propergate result
@@ -271,8 +257,8 @@ protected:
 protected:
   /** Output rescaling. */
   int _shift;
-  /** The last input value. */
-  std::complex<iScalar> _last_value;
+  /** The last angle. */
+  oScalar _last_value;
   /** If true, in-place demodulation is poissible. */
   bool _can_overwrite;
   /** The output buffer, unused if demodulation is performed in-place. */
diff --git a/src/http.cc b/src/http.cc
index d463e8a..9144d3a 100644
--- a/src/http.cc
+++ b/src/http.cc
@@ -98,20 +98,13 @@ inline http::Version to_version(const std::string &version) {
  * Implementation of Server
  * ********************************************************************************************* */
 Server::Server(uint port)
-  : _port(port), _socket(-1)
+  : _port(port), _socket(-1), _queue()
 {
-  // pass...
+  pthread_mutex_init(&_queue_lock, 0);
 }
 
 Server::~Server() {
-  _is_running = false;
-  // Close all connections
-  std::set<Connection *>::iterator con = _connections.begin();
-  for (; con != _connections.end(); con++) {
-    // Wait for the connection to close
-    (*con)->close(true);
-    delete *con;
-  }
+  _is_running = false;  
   // Free all handler
   std::list<Handler *>::iterator item = _handler.begin();
   for (; item != _handler.end(); item++) { delete *item; }
@@ -157,49 +150,70 @@ Server::start(bool wait) {
 void
 Server::stop(bool wait) {
   _is_running = false;
-  // Close all open connections
-  std::set<Connection *>::iterator con = _connections.begin();
-  for (; con != _connections.end(); con++) {
-    (*con)->close(wait);
-  }
   // Close the socket we listen on
   ::close(_socket); _socket = -1;
+  // wait for server to join
+  if (wait) { this->wait(); }
 }
 
 void
 Server::wait() {
   void *ret=0;
   pthread_join(_thread, &ret);
+  // wait for all handlers to join
+  while (_threads.size()) {
+    void *ret = 0;
+    pthread_join(*_threads.begin(), &ret);
+  }
 }
 
 void *
-Server::_listen_main(void *ctx) {
+Server::_listen_main(void *ctx)
+{
+  // Get server instance
   Server *self = (Server *)ctx;
-  while (self->_is_running) {
+  // Whil server is running
+  while (self->_is_running)
+  {
+    // Wait for incomming connections
     listen(self->_socket, 5);
+
+    // Create socket to client
     struct sockaddr_in cli_addr;
     socklen_t clilen = sizeof(cli_addr);
     int socket = accept(self->_socket, (struct sockaddr *) &cli_addr, &clilen);
     if (socket < 0) { continue; }
-    try { self->_connections.insert(new Connection(self, socket)); }
-    catch (...) { }
-    // Free closed connections
-    std::list<Connection *> closed_connections;
-    std::set<Connection *>::iterator item = self->_connections.begin();
-    for (; item != self->_connections.end(); item++) {
-      if ((*item)->isClosed()) { closed_connections.push_back(*item); }
-    }
-    std::list<Connection *>::iterator citem = closed_connections.begin();
-    for (; citem != closed_connections.end(); citem++) {
-      self->_connections.erase(*citem);
-      delete *citem;
+
+    // Construct connection object from socket
+    try {
+      pthread_mutex_lock(&self->_queue_lock);
+      self->_queue.push_back(Connection(self, socket));
+      pthread_t thread;
+      pthread_create(&thread, 0, &Server::_connection_main, self);
+      self->_threads.insert(thread);
+      pthread_mutex_unlock(&self->_queue_lock);
+    } catch (...) {
+      pthread_mutex_unlock(&self->_queue_lock);
     }
   }
   return 0;
 }
 
+void *
+Server::_connection_main(void *ctx) {
+  Server *self = (Server *)ctx;
+  pthread_mutex_lock(&self->_queue_lock);
+  Connection con = self->_queue.front(); self->_queue.pop_front();
+  pthread_mutex_unlock(&self->_queue_lock);
+  con.main();
+  // Remove thread from list of running threads
+  self->_threads.erase(pthread_self());
+  return 0;
+}
+
 void
-Server::dispatch(const Request &request, Response &response) {
+Server::dispatch(const Request &request, Response &response)
+{
   LogMessage msg(LOG_DEBUG);
   msg << "httpd: ";
   switch (request.method()) {
@@ -230,70 +244,120 @@ Server::addHandler(Handler *handler) {
 
 
 /* ********************************************************************************************* *
- * Implementation of HTTPD::Connection
+ * Implementation of http::Connection & http::ConnectionObj
  * ********************************************************************************************* */
-Connection::Connection(Server *server, int socket)
-  : _server(server), _socket(socket)
+ConnectionObj::ConnectionObj(Server *server, int cli_socket)
+  : server(server), socket(cli_socket), refcount(1)
 {
-  // Start new thread to parse requests
-  if (pthread_create(&_thread, 0, &Connection::_main, (void *)this)) {
-    ::close(_socket); _socket = -1;
-    RuntimeError err;
-    err << "httpd: Can not create thread for connection.";
-    throw err;
+  // pass...
+}
+
+ConnectionObj::~ConnectionObj() {
+  ::close(socket); socket=-1;
+}
+
+ConnectionObj *
+ConnectionObj::ref() {
+  refcount++; return this;
+}
+
+void
+ConnectionObj::unref() {
+  refcount--;
+  if (0 == refcount) {
+    delete this;
   }
 }
 
+Connection::Connection()
+  : _object(0)
+{
+  // pass...
+}
+
+Connection::Connection(Server *server, int socket)
+  : _object(new ConnectionObj(server, socket))
+{
+  // pass...
+}
+
+Connection::Connection(const Connection &other)
+  : _object(other._object)
+{
+  if (_object) { _object->ref(); }
+}
+
 Connection::~Connection() {
   // Close the socket
-  this->close(false);
+  if (_object) { _object->unref(); }
 }
 
+Connection &
+Connection::operator =(const Connection &other) {
+  if (_object) { _object->unref(); }
+  _object = other._object;
+  if (_object) { _object->ref(); }
+  return *this;
+}
+
+
 void
 Connection::close(bool wait) {
-  if (-1 != _socket) {
-    int socket = _socket; _socket = -1;
+  if (0 == _object) { return; }
+  if (-1 != _object->socket) {
+    int socket = _object->socket; _object->socket = -1;
     LogMessage msg(LOG_DEBUG);
     msg << "httpd: Close connection " << socket << ".";
     Logger::get().log(msg);
     ::close(socket);
   }
-  if (wait) {
-    // Wait for the thread to exit.
-    void *ret = 0;
-    pthread_join(_thread, &ret);
-  }
 }
 
 bool
 Connection::isClosed() const {
-  return (-1 == _socket);
+  if (0 == _object) { return true; }
+  return (-1 == _object->socket);
 }
 
-void *
-Connection::_main(void *ctx)
+bool
+Connection::send(const std::string &data) const {
+  const char *ptr = data.c_str();
+  size_t count = data.size();
+  while (count) {
+    int c = this->write(ptr, count);
+    if (c < 0) { return false; }
+    count -= c; ptr += c;
+  }
+  return true;
+}
+
+
+void
+Connection::main()
 {
-  Connection *self = (Connection *)ctx;
+  int error = 0; socklen_t errorlen = sizeof(error);
   // While socket is open
-  int error = 0; socklen_t errorlen = sizeof (error);
-  while (0 == getsockopt(self->_socket, SOL_SOCKET, SO_ERROR, &error, &errorlen)) {
-    Request request(self->_socket);
-    Response response(self->_socket);
-    // Parse request
-    if (!request.parse()) {
-      // On parser error or no keep-alive -> exit
-      self->close(); return 0;
+  while (0 == getsockopt(_object->socket, SOL_SOCKET, SO_ERROR, &error, &errorlen)) {
+    // Contstruct request & reponse instances
+    Request request(*this);
+    Response response(*this);
+    // try to parse request
+    if (! request.parse()) {
+      // On parser error close connection -> exit
+      this->close(false); return;
     }
     // on success -> dispatch request by server
-    self->_server->dispatch(request, response);
-    // If the connection is kept alive -> continue
-    if ((! request.isKeepAlive()) || response.closeConnection()) {
-      // Signal server to close connection
-      self->close(self); return 0;
+    _object->server->dispatch(request, response);
+    // on protocol update
+    if (this->protocolUpgrade()) { return; }
+    // If the connection is keep-alive -> continue
+    if (! request.isKeepAlive()) {
+      // close connection -> exit
+      this->close(false); return;
     }
   }
-  // Signal server to close connection
-  self->close(self); return 0;
+  // Close connection
+  this->close(false);
 }
 
 
@@ -449,8 +513,8 @@ typedef enum {
 } HttpRequestParserState;
 
 
-Request::Request(int socket)
-  : _socket(socket), _method(HTTP_UNKNOWN)
+Request::Request(const Connection &connection)
+  : _connection(connection), _method(HTTP_UNKNOWN)
 {
   // pass...
 }
@@ -463,7 +527,7 @@ Request::parse() {
   HttpRequestParserState state = READ_METHOD;
 
   // while getting a char from stream
-  while (::read(_socket, &c, 1)) {
+  while (_connection.read(&c, 1)) {
     switch (state) {
     case READ_METHOD:
       if (is_space(c)) {
@@ -626,7 +690,7 @@ Request::readBody(std::string &body) const {
   size_t N = contentLength(); body.reserve(N);
   char buffer[65536];
   while (N>0) {
-    int res = ::read(_socket, buffer, std::min(N, size_t(65536)));
+    int res = _connection.read(buffer, std::min(N, size_t(65536)));
     if (res>=0) { body.append(buffer, size_t(res)); N -= res; }
     else { return false; }
   }
@@ -637,8 +701,8 @@ Request::readBody(std::string &body) const {
 /* ********************************************************************************************* *
  * Implementation of HTTPD::Response
  * ********************************************************************************************* */
-Response::Response(int socket)
-  : _socket(socket), _status(STATUS_SERVER_ERROR), _close_connection(false)
+Response::Response(const Connection &connection)
+  : _connection(connection), _status(STATUS_SERVER_ERROR), _close_connection(false)
 {
   // pass...
 }
@@ -670,18 +734,6 @@ Response::setContentLength(size_t length) {
   setHeader("Content-Length", buffer.str());
 }
 
-bool
-Response::send(const std::string &data) const {
-  const char *ptr = data.c_str();
-  size_t count = data.size();
-  while (count) {
-    int c = ::write(_socket, ptr, count);
-    if (c < 0) { return false; }
-    count -= c; ptr += c;
-  }
-  return true;
-}
-
 bool
 Response::sendHeaders() const {
   std::stringstream buffer;
@@ -699,7 +751,7 @@ Response::sendHeaders() const {
     buffer << item->first << ": " << item->second << "\r\n";
   }
   buffer << "\r\n";
-  return send(buffer.str());
+  return _connection.send(buffer.str());
 }
 
 
@@ -742,7 +794,7 @@ StaticHandler::handle(const Request &request, Response &response) {
   }
   response.setContentLength(_text.size());
   response.sendHeaders();
-  response.send(_text);
+  response.connection().send(_text);
 }
 
 
@@ -785,7 +837,7 @@ JSONHandler::handle(const http::Request &request, http::Response &response) {
     result.serialize(result_string);
     response.setContentLength(result_string.size());
     response.sendHeaders();
-    response.send(result_string);
+    response.connection().send(result_string);
   } else {
     response.setStatus(http::Response::STATUS_BAD_REQUEST);
     response.setContentLength(0);
diff --git a/src/http.hh b/src/http.hh
index 2b6c51a..1dbe17a 100644
--- a/src/http.hh
+++ b/src/http.hh
@@ -15,7 +15,7 @@
  *
  * using namespace sdr;
  *
- * // Implements an application, a collection of methods being called from the http::Server.
+ * // Implements an application, a collection of methods being called by the http::Server.
  * class Application {
  * public:
  *   // contstructor.
@@ -251,17 +251,90 @@ protected:
 };
 
 
+struct ConnectionObj
+{
+public:
+  ConnectionObj(Server *server, int cli_socket);
+  ~ConnectionObj();
+
+  ConnectionObj *ref();
+  void unref();
+
+public:
+  /** A weak reference to the server instance. */
+  Server *server;
+  /** The connection socket. */
+  int socket;
+  /** If @c true (i.e. set by a handler), the http parser thread will exit without closing
+   * the connection. This allows to "take-over" the tcp connection to the client by the request
+   * handler. */
+  bool protocol_upgrade;
+  /** Reference counter. */
+  size_t refcount;
+};
+
+
+/** Implements a HTTP connection to a client.
+ * @c ingroup http */
+class Connection
+{
+public:
+  /** Empty constructor. */
+  Connection();
+  /** Constructor. */
+  Connection(Server *server, int socket);
+  /** Copy constructor. Implements reference counting. */
+  Connection(const Connection &other);
+  /** Destructor. */
+  ~Connection();
+
+  Connection &operator=(const Connection &other);
+
+  /** Closes the connection.
+   * If @c wait is @c true, the method will wait until the thread listening for incomming
+   * request joined. */
+  void close(bool wait=false);
+  /** Returns @c true if the connection is closed. */
+  bool isClosed() const;
+
+  /** Sets the protocol-update flag. */
+  inline void setProtocolUpgrade() const { _object->protocol_upgrade = true; }
+  inline bool protocolUpgrade() const { return _object->protocol_upgrade; }
+
+  inline ssize_t write(const void *data, size_t n) const {
+    if (0 == _object) { return -1; }
+    return ::write(_object->socket, data, n);
+  }
+
+  inline ssize_t read(void *data, size_t n) const {
+    if (0 == _object) { return -1; }
+    return ::read(_object->socket, data, n);
+  }
+
+  bool send(const std::string &data) const;
+
+  /** Main loop for incomming requests. */
+  void main();
+
+protected:
+  ConnectionObj *_object;
+};
+
+
 /** Represents a HTTP request.
  * @ingroup http */
 class Request
 {
 public:
   /** Constructor. */
-  Request(int socket);
+  Request(const Connection &connection);
 
   /** Parses the HTTP request header, returns @c true on success. */
   bool parse();
 
+  /** Return the connection to the client. */
+  inline const Connection &connection() const { return _connection; }
+
   /** Returns @c true if the connection to the client is kept alive after the response
      * has been send. */
   bool isKeepAlive() const;
@@ -280,17 +353,13 @@ public:
   inline Method method() const { return _method; }
   /** Returns the request URL. */
   inline const URL &url() const { return _url; }
-  /** Allows to read data from the connection (i.e. the request body). */
-  inline int read(const char *data, size_t size) {
-    return ::read(_socket, (void *)data, size);
-  }
   /** Reads the complete body (if Content-Length header is present).
-     * Retruns @c true on success.*/
+   * Retruns @c true on success.*/
   bool readBody(std::string &body) const;
 
 protected:
   /** The connection socket. */
-  int _socket;
+  Connection _connection;
   /** The request method. */
   Method _method;
   /** The HTTP version. */
@@ -315,9 +384,11 @@ public:
   } Status;
 
 public:
-  /** Constructor.
-   * @param socket Specifies the socket over which the response will be send.*/
-  Response(int socket);
+  /** Constructor. */
+  Response(const Connection &connnection);
+
+  /** Return the connection to the client. */
+  inline const Connection &connection() const { return _connection; }
 
   /** Specifies the response code. */
   void setStatus(Status status);
@@ -333,19 +404,9 @@ public:
   /** Sends the response code and all defined headers. */
   bool sendHeaders() const;
 
-  /** Sends some data through the socket. To send the response body, call
-   * @c sendHeaders() first. */
-  bool send(const std::string &data) const;
-
-  /** Returns @c true if the connection will be closed after the response has been send.
-   * @c I.e. if there was an error. */
-  inline bool closeConnection() const { return _close_connection; }
-  /** Marks that the connection will be closed after the response has been send. */
-  inline void setCloseConnection() { _close_connection = true; }
-
 protected:
   /** The socket over which the response will be send. */
-  int _socket;
+  Connection _connection;
   /** The response code. */
   Status _status;
   /** The response headers. */
@@ -355,37 +416,6 @@ protected:
 };
 
 
-/** Implements a HTTP connection to a client.
- * @c ingroup http */
-class Connection
-{
-public:
-  /** Constructor. */
-  Connection(Server *server, int socket);
-  /** Destructor. */
-  ~Connection();
-
-  /** Closes the connection.
-   * If @c wait is @c true, the method will wait until the thread listening for incomming
-   * request joined. */
-  void close(bool wait=false);
-  /** Returns @c true if the connection is closed. */
-  bool isClosed() const;
-
-protected:
-  /** Main loop for incomming requests. */
-  static void *_main(void *ctx);
-
-protected:
-  /** A weak reference to the server instance. */
-  Server *_server;
-  /** The connection socket. */
-  int _socket;
-  /** The thread processing requests for this connection. */
-  pthread_t _thread;
-};
-
-
 /** Base class of all HTTP request handlers.
  * @ingroup http */
 class Handler
@@ -434,7 +464,7 @@ protected:
 };
 
 
-/** Utility function to provide a handler as a delegate.
+/** Utility class to provide a handler as a delegate.
  * @ingroup http */
 template <class T>
 class DelegateHandler: public Handler
@@ -530,9 +560,11 @@ public:
   /** Starts the server.
    * If @c wait is @c true, the call to this method will bock until the server thread stops. */
   void start(bool wait=false);
+
   /** Stops the server.
    * If @c wait is @c true, the call will block until the server thread stopped. */
   void stop(bool wait=false);
+
   /** Wait for the server thread to join. */
   void wait();
 
@@ -564,6 +596,8 @@ protected:
   void dispatch(const Request &request, Response &response);
   /** The thread waiting for incomming connections. */
   static void *_listen_main(void *ctx);
+  /** The thread handling connections. */
+  static void *_connection_main(void *ctx);
 
 protected:
   /** Port to bind to. */
@@ -576,8 +610,12 @@ protected:
   pthread_t _thread;
   /** All registered handler. */
   std::list<Handler *> _handler;
-  /** All open connections. */
-  std::set<Connection *> _connections;
+  /** The connection queue. */
+  std::list<Connection> _queue;
+  /** The queue lock. */
+  pthread_mutex_t _queue_lock;
+  /** The set of handler threads. */
+  std::set<pthread_t> _threads;
 
   /* Allow Connection to access dispatch(). */
   friend class Connection;