root/trunk/whispercast/stream_server.cc

// Copyright (c) 2009, Whispersoft s.r.l.
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
// * Neither the name of Whispersoft s.r.l. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// Author: Catalin Popescu
//
// This is the main file of whispercast server..
//

#include <whisperlib/common/app/app.h>

#include <whisperlib/common/base/types.h>
#include <whisperlib/common/base/log.h>
#include <whisperlib/common/base/system.h>
#include <whisperlib/common/base/gflags.h>
#include <whisperlib/common/base/errno.h>
#include <whisperlib/common/base/strutil.h>
#include <whisperlib/common/sync/thread.h>
#include <whisperlib/common/base/callback.h>

#include <whisperlib/common/io/file/file_input_stream.h>
#include <whisperlib/net/base/selector.h>

#include <whisperlib/net/rpc/lib/server/rpc_http_server.h>

#include <whisperstreamlib/stats2/stats_collector.h>
#include <whisperstreamlib/stats2/log_stats_saver.h>
#include <whisperstreamlib/base/element.h>
#include <whisperlib/net/util/ipclassifier.h>

#include <whisperlib/net/base/user_authenticator.h>
#include <whisperlib/net/http/http_server_protocol.h>

#include <whisperstreamlib/rtmp/rtmp_connection.h>
#include <whisperstreamlib/elements/factory.h>

#include <whisperlib/common/io/ioutil.h>
#include "media_mapper.h"
#include "stream_request.h"

//////////////////////////////////////////////////////////////////////

DEFINE_int32(http_port,
             8080,
             "The port on which we accept HTTP connections");
DEFINE_string(http_local_address,
              "0.0.0.0",
              "The local address to bind to when accepting HTTP connections");
DEFINE_int32(https_port,
             8443,
             "The port on which we accept HTTPS connections");
DEFINE_string(https_local_address,
              "0.0.0.0",
              "The local address to bind to when accepting HTTPS connections");
DEFINE_int32(http_connection_max_media_outbuf_size,
             65536 + 65536 + 16384,
             "Sending buffer per HTTP connection for all media data");
DEFINE_int32(http_connection_read_timeout,
             20000,
             "Timeout for reading the HTTP client request");
DEFINE_int32(http_connection_write_timeout,
             20000,
             "Timeout for a HTTP client connection before disconnecting");
DEFINE_int64(http_connection_write_ahead_ms,
             4000,
             "How much HTTP data to write ahead of the element time, "
             "in miliseconds");
DEFINE_bool(http_connection_dlog_level,
            false,
            "Turn on deep debugging messages on HTTP connections");
DEFINE_int32(http_max_num_connections,
             1000,
             "Accept at most these many concurrent HTTP connections");

DEFINE_string(http_default_content_type,
              "text/html",
              "We send this content type when we don't know what "
              "to send");
DEFINE_string(http_extensions_to_content_types_file,
             "",
             "If set we read mappings from extensions to "
             "content types from here");
DEFINE_string(http_ssl_certificate,
              "",
              "Path to an SSL certificate file. We need both certificate & key"
              " to use HTTP over SSL; else we use regular HTTP over TCP");
DEFINE_string(http_ssl_key,
              "",
              "Path to an SSL key file. We need both certificate & key"
              " to use HTTP over SSL; else we use regular HTTP over TCP");

//////////////////////////////////////////////////////////////////////

DEFINE_int32(rtmp_port,
             1935,
             "The port on which we accept RTMP connections");
DEFINE_string(rtmp_local_address,
              "0.0.0.0",
              "The local address to bind to when accepting RTMP connections");
DEFINE_string(rtmp_publishing_application,
              "live",
              "If this is set we allow client publishing under "
              "this rtmp application");

//////////////////////////////////////////////////////////////////////

DEFINE_string(ip_classifiers,
              "",
              "Comma sepparated list of classifiers - first class 0, etc");

//////////////////////////////////////////////////////////////////////


DEFINE_string(log_stats_dir,
              "",
              "The directory where the stats log files will be written.");
DEFINE_string(log_stats_file_base,
              "",
              "The base name for stats log files.");

//////////////////////////////////////////////////////////////////////

DEFINE_string(server_id,
              "",
              "The ID of the server, as it will be logged");

//////////////////////////////////////////////////////////////////////

DEFINE_string(media_config_dir,
              "",
              "Where to load / save the configs");
DEFINE_string(base_media_dir,
              "",
              "Media is under this directory");
DEFINE_string(media_path,
              "/media",
              "We export our media elements under this path for http");
DEFINE_string(media_state_dir,
              "",
              "Where we may save our state");
DEFINE_string(media_state_name,
              "whispercast",
              "We save the state under this name");
DEFINE_string(local_media_state_name,
              "local_whispercast",
              "We save non config state under this name");

//////////////////////////////////////////////////////////////////////

DEFINE_string(admin_passwd,
              "",
              "An admin password for server administration - put this in a flag file");
DEFINE_string(authorization_realm,
              "",
              "If you set an admin password, you also need this");
DEFINE_string(rpc_config_allow_classifier,
              "",
              "If not empty we allow config rpc-s only from guys who "
              "qualify in the classifier created by this IpClassifier (e.g. "
              "'OR(IPFILTER(201.1.2.240/26), "
              "IPFILTERFILE(/etc/whispercast/good_rpc_ips))' )");
DEFINE_string(rpc_stats_allow_classifier,
              "",
              "If not empty we allow stat rpc-s only from guys who qualify "
              "in the classifier created by this IpClassifier (e.g. "
              "'OR(IPFILTER(201.1.2.240/26), "
              "IPFILTERFILE(/etc/whispercast/good_rpc_ips))' )");

DECLARE_int32(rtmp_max_num_connections);

//////////////////////////////////////////////////////////////////////

// We don't want to get crawled - disable all robots
void RobotsProcessor(http::ServerRequest* req) {
  req->request()->server_header()->AddField(
    http::kHeaderContentType, "text/plain", true);
  req->request()->server_data()->Write(
    "User-agent: *\n"
    "Disallow: /\n");
  req->Reply();
}

void QuickStats(
  streaming::StatsCollector* collector,
  http::ServerRequest* req) {
  req->request()->server_header()->AddField(
    http::kHeaderContentType, "text/plain", true);
  // TODO [cpopescu]:
  req->Reply();
}

string DecodeAdminPasswd(const string& s) {
  const string tmp = URL::UrlUnescape(s);
  string ret;
  for ( int i = 0; i < tmp.size(); ++i ) {
    ret += int8(tmp[i]) ^ 0xb6;
  }
  DLOG_INFO << " Decoded passwd: [" << ret << "]";
  return ret;
}

//////////////////////////////////////////////////////////////////////

class Whispercast : public app::App {
public:
  Whispercast(int &argc, char **&argv);
  virtual ~Whispercast();

protected:
  int Initialize();

  void Run() {
    selector_->Loop();
  }
  void Stop() {
    selector_->RunInSelectLoop(NewCallback(selector_,
                                           &net::Selector::MakeLoopExit));
  }
  void Shutdown();

  void ProcessMediaRequest(http::ServerRequest* req);
  void ProcessAuthorizedMediaRequest(
      http::ServerRequest* req,
      streaming::Request* mreq,
      streaming::Authorizer* auth);

 private:
  net::Selector* selector_;
  MediaMapper* media_mapper_;

  SSL_CTX* http_ssl_ctx_;
  net::NetFactory* http_net_factory_;
  http::Server* http_server_;
  http::ServerParams* http_server_params_;

  rtmp::StandardStreamManager* rtmp_stream_manager_;
  rtmp::ServerAcceptor* rtmp_server_;
  rtmp::ProtocolFlags rtmp_flags_;

  net::SimpleUserAuthenticator* admin_authenticator_;
  rpc::HttpServer* rpc_stat_processor_;
  rpc::HttpServer* rpc_processor_;

  //////////////////////////////////////////////////////////////////////

  streaming::StatsCollector* stats_collector_;

  map<string, string> ext2ct_;   // maps extensions to content type

  vector<const net::IpClassifier*>* classifiers_;
  int64 connection_id_;
};

//////////////////////////////////////////////////////////////////////

Whispercast::Whispercast(int &argc, char **&argv)
  : app::App(argc, argv),
    selector_(NULL),
    media_mapper_(NULL),
    http_ssl_ctx_(NULL),
    http_net_factory_(NULL),
    http_server_(NULL),
    http_server_params_(NULL),
    rtmp_stream_manager_(NULL),
    rtmp_server_(NULL),
    admin_authenticator_(NULL),
    rpc_stat_processor_(NULL),
    rpc_processor_(NULL),
    stats_collector_(NULL),
    classifiers_(NULL),
    connection_id_(0) {
  google::SetUsageMessage("Whispercast - stream broadcasting server.");
}

Whispercast::~Whispercast() {
  CHECK_NULL(selector_);
  CHECK_NULL(media_mapper_);
  CHECK_NULL(http_ssl_ctx_);
  CHECK_NULL(http_net_factory_);
  CHECK_NULL(http_server_);
  CHECK_NULL(http_server_params_);
  CHECK_NULL(rtmp_stream_manager_);
  CHECK_NULL(rtmp_server_);
  CHECK_NULL(rpc_stat_processor_);
  CHECK_NULL(rpc_processor_);
  CHECK_NULL(stats_collector_);
  CHECK_NULL(classifiers_);
}


int Whispercast::Initialize() {
  common::Init(argc_, argv_);

  //////////////////////////////////////////////////////////////////////

  selector_ = new net::Selector();

  //////////////////////////////////////////////////////////////////////

  // Create the classifiers
  if ( !FLAGS_ip_classifiers.empty() ) {
      classifiers_ = new vector<const net::IpClassifier*>();
      vector<string> components;
      CHECK(strutil::SplitBracketedString(FLAGS_ip_classifiers.c_str(),
                                          ',', '(', ')',  &components))
                                            << " Invalid classifier spec: ["
                                            << FLAGS_ip_classifiers << "]";
      for ( int i = 0; i < components.size(); ++i ) {
        classifiers_->push_back(
          net::IpClassifier::CreateClassifier(components[i]));
      }
    }

  //////////////////////////////////////////////////////////////////////

  // Initialize the stats collection
  vector<streaming::StatsSaver*> stats_savers;
  if(FLAGS_log_stats_dir != "") {
    stats_savers.push_back(new streaming::LogStatsSaver(
                                FLAGS_log_stats_dir.c_str(),
                                FLAGS_log_stats_file_base.c_str()));
  }
  stats_collector_ = new streaming::StatsCollector(
      selector_, FLAGS_server_id, timer::Date::Now(), &stats_savers);
  CHECK(stats_collector_->Start());

  //////////////////////////////////////////////////////////////////////

  // Create the HTTP server

  // TODO : add http_connection_write_ahead
  http_net_factory_ = new net::NetFactory(selector_);
  if ( FLAGS_http_ssl_certificate != "" &&
       FLAGS_http_ssl_key != "" ) {
    http_ssl_ctx_ = net::SslConnection::SslCreateContext(
        FLAGS_http_ssl_certificate, FLAGS_http_ssl_key);
    if ( http_ssl_ctx_ != NULL ) {
      net::SslConnectionParams ssl_connection_params;
      ssl_connection_params.ssl_context_ = http_ssl_ctx_;
      net::SslAcceptorParams ssl_acceptor_params(ssl_connection_params);
      http_net_factory_->SetSslParams(ssl_acceptor_params,
                                      ssl_connection_params);
    }
  }

  http_server_params_ = new http::ServerParams();
  http::ServerParams* const p = http_server_params_;  // shortcut
  p->max_reply_buffer_size_ = FLAGS_http_connection_max_media_outbuf_size;
  p->reply_write_timeout_ms_ = FLAGS_http_connection_write_timeout;
  p->request_read_timeout_ms_ = FLAGS_http_connection_read_timeout;
  p->dlog_level_ = FLAGS_http_connection_dlog_level;
  p->max_concurrent_connections_ = FLAGS_http_max_num_connections;
  // Is essential to have the keep-alive active as our rpc-s should keep-alive
  // The loss for streaming is not big deal..
  //
  // p->keep_alive_timeout_sec_ = 0;  // no keep-alive for us

  http_server_ = new http::Server(
    "Whispercast 0.03",
    selector_,
    http_net_factory_,
    http_server_params_,
    NewPermanentCallback(&http::SimpleServerConnectionFactory),
    0);  // no thread pool -- all processing in thread :)

  http_server_->RegisterProcessor(FLAGS_media_path,
                                  NewPermanentCallback(
                                      this, &Whispercast::ProcessMediaRequest),
                                  true);

  //////////////////////////////////////////////////////////////////////

  // Add RPC statistics mapping to HTTP server

  if ( !FLAGS_authorization_realm.empty() ) {
    admin_authenticator_ =
        new net::SimpleUserAuthenticator(FLAGS_authorization_realm);
    admin_authenticator_->set_user_password(string("admin"), FLAGS_admin_passwd);
  } else {
    admin_authenticator_ = NULL;
  }
  rpc_stat_processor_ = new rpc::HttpServer(selector_,
                                            http_server_,
                                            NULL,  // authenticator_,
                                            "/rpc-stats", true,
                                            true, 10,
                                            FLAGS_rpc_stats_allow_classifier);
  rpc_processor_ = new rpc::HttpServer(selector_,
                                       http_server_,
                                       admin_authenticator_,
                                       "/rpc-config", true,
                                       true, 10,
                                       FLAGS_rpc_config_allow_classifier);

  rtmp_stream_manager_ = new rtmp::StandardStreamManager(
      selector_,
      NULL,  // will set later..
      stats_collector_,
      FLAGS_rtmp_publishing_application);

  //////////////////////////////////////////////////////////////////////

  CHECK(!FLAGS_media_config_dir.empty() &&
        io::IsDir(FLAGS_media_config_dir.c_str()))
    << " [" << FLAGS_media_config_dir << "]";
  CHECK(FLAGS_base_media_dir.empty() ||
        io::IsDir(FLAGS_base_media_dir.c_str()))
    << " [" << FLAGS_base_media_dir << "]";
  media_mapper_ = new MediaMapper(selector_,
                                  http_server_,
                                  rpc_processor_,
                                  rtmp_stream_manager_,
                                  admin_authenticator_,
                                  FLAGS_media_config_dir,
                                  FLAGS_base_media_dir,
                                  FLAGS_media_state_dir,
                                  FLAGS_media_state_name,
                                  FLAGS_local_media_state_name);
  const MediaMapper::LoadError load_err = media_mapper_->Load();
  if ( load_err != MediaMapper::LOAD_OK ) {
    // TODO email
    LOG_ERROR << "\n=========================== BAD ERROR: ==================="
              << "\nSome error loading config file - this may render some "
              << "elements unavailable!"
              << "\n=========================================================";
    CHECK_EQ(load_err, MediaMapper::LOAD_FILE_ERROR)
      << " We encountered some data consistency error loading configuration "
      << " - failing !!";
  }
  rtmp_stream_manager_->set_element_mapper(media_mapper_->mapper());

  //////////////////////////////////////////////////////////////////////

  // TODO: make private
  CHECK(rpc_processor_->RegisterService(media_mapper_));
  CHECK(rpc_stat_processor_->RegisterService(stats_collector_));

  //////////////////////////////////////////////////////////////////////

  // TODO: Add Stream elements
  //       Add exported streams

  //////////////////////////////////////////////////////////////////////

  // Create the RTMP server
  rtmp::GetProtocolFlags(&rtmp_flags_);
  rtmp_server_ = new rtmp::ServerAcceptor(selector_,
                                          rtmp_stream_manager_,
                                          "rtmp",
                                          stats_collector_,
                                          classifiers_,
                                          &rtmp_flags_);

  //////////////////////////////////////////////////////////////////////

  // Register various path mappings

  // Register robots porcessor
  http_server_->RegisterProcessor(
    "/robots.txt", NewPermanentCallback(&RobotsProcessor), true);
  // Register stats processor
  http_server_->RegisterProcessor(
    "/__stats__", NewPermanentCallback(QuickStats,  stats_collector_),
    true);  // public -> todo make private ..

  //////////////////////////////////////////////////////////////////////

  if ( !FLAGS_http_extensions_to_content_types_file.empty() ) {
    const string s = io::FileInputStream::ReadFileOrDie(
        FLAGS_http_extensions_to_content_types_file.c_str());
    vector< pair<string, string> > exts;
    strutil::SplitPairs(s, "\n",  ":", &exts);
    for ( int i = 0; i < exts.size(); ++i ) {
      ext2ct_[strutil::StrTrim(exts[i].first)] =
          strutil::StrTrim(exts[i].second);
    }
  }

  // Start http serving
  if ( FLAGS_http_port != 0 ) {
    http_server_->AddAcceptor(net::PROTOCOL_TCP,
                              net::HostPort(FLAGS_http_local_address.c_str(),
                                            FLAGS_http_port));
  }
  if ( FLAGS_https_port != 0 && http_ssl_ctx_ != NULL ) {
    http_server_->AddAcceptor(net::PROTOCOL_SSL,
                              net::HostPort(FLAGS_https_local_address.c_str(),
                                            FLAGS_https_port));
  }
  selector_->RunInSelectLoop(NewCallback(http_server_,
                                         &http::Server::StartServing));
  // Start rtmp serving
  selector_->RunInSelectLoop(NewCallback(rtmp_server_,
                                         &rtmp::ServerAcceptor::StartServing,
                                         FLAGS_rtmp_port,
                                         FLAGS_rtmp_local_address.c_str()));

  return 0;
}

void Whispercast::Shutdown() {
  //////////////////////////////////////////////////////////////////////

  delete media_mapper_;
  media_mapper_ = NULL;

  LOG_INFO << " Media mapper deleted !";

  //////////////////////////////////////////////////////////////////////


  stats_collector_->Stop();
  delete stats_collector_;
  stats_collector_ = NULL;

  //////////////////////////////////////////////////////////////////////

  // Delete the classifiers
  if ( classifiers_ != NULL ) {
    for ( int i = 0; i < classifiers_->size(); ++i ) {
      delete (*classifiers_)[i];
    }
    delete classifiers_;
    classifiers_ = NULL;
  }

  // Delete the HTTP server
  delete http_server_;
  http_server_ = NULL;
  delete http_server_params_;
  http_server_params_ = NULL;
  delete http_net_factory_;
  http_net_factory_ = NULL;
  net::SslConnection::SslDeleteContext(http_ssl_ctx_);
  http_ssl_ctx_ = NULL;

  //////////////////////////////////////////////////////////////////////

  // Close the RTMP server

  delete rtmp_stream_manager_;
  rtmp_stream_manager_ = NULL;
  rtmp_server_ = NULL;  // auto-deleted...

  //////////////////////////////////////////////////////////////////////

  // Stop RPC
  delete rpc_processor_;
  rpc_processor_ = NULL;
  delete rpc_stat_processor_;
  rpc_stat_processor_ = NULL;
  delete admin_authenticator_;
  admin_authenticator_ = NULL;

  //////////////////////////////////////////////////////////////////////

  delete selector_;
  selector_ = NULL;
}

void Whispercast::ProcessMediaRequest(http::ServerRequest* req) {
  URL* const url = req->request()->url();
  CHECK(url != NULL);
  const string url_path(url->UrlUnescape(url->path().c_str(),
                                           url->path().size()));
  const string media_path(url_path.substr(FLAGS_media_path.size()));
  int ip_class = -1;
  if ( classifiers_ != NULL ) {
    for ( int i = 0; i < classifiers_->size(); ++i ) {
      if ( (*classifiers_)[i]->IsInClass(
               req->remote_address().ip_object()) ) {
        ip_class = i;
        break;
      }
    }
  }
  streaming::Request* mreq = new streaming::Request(*url);

  mreq->mutable_info()->remote_address_ =
      req->remote_address();
  mreq->mutable_info()->local_address_ =
      net::HostPort("127.0.0.1", FLAGS_http_port);
  mreq->mutable_info()->ip_class_ = ip_class;

  mreq->mutable_info()->auth_req_.net_address_ =
      req->remote_address().ToString();
  mreq->mutable_info()->auth_req_.resource_ = url->spec();
  mreq->mutable_info()->auth_req_.action_ = streaming::kActionView;

  if ( mreq->mutable_info()->auth_req_.token_.empty() ) {
    mreq->mutable_info()->auth_req_.token_ =
        req->request()->client_header()->FindField(http::kHeaderCookie);
  }
  if ( mreq->mutable_info()->auth_req_.user_.empty() ) {
    req->request()->client_header()->GetAuthorizationField(
        &mreq->mutable_info()->auth_req_.user_,
        &mreq->mutable_info()->auth_req_.passwd_);
  }

  if ( !media_mapper_->mapper()->GetMediaDetails("http",
                                                 media_path,
                                                 mreq) ) {
    delete mreq;
    req->ReplyWithStatus(http::NOT_FOUND);
    return;
  }
  if ( mreq->serving_info().flavour_mask_is_set_ ) {
    mreq->mutable_caps()->flavour_mask_ = mreq->serving_info().flavour_mask_;
  }
  if ( mreq->serving_info().authorizer_name_.empty() ) {
    mreq->mutable_auth_reply()->allowed_ = true;
    ProcessAuthorizedMediaRequest(req, mreq, NULL);
  } else {
    streaming::Authorizer* const auth = media_mapper_->mapper()->GetAuthorizer(
        mreq->serving_info().authorizer_name_);
    if ( auth == NULL ) {
      mreq->mutable_auth_reply()->allowed_ = true;
      ProcessAuthorizedMediaRequest(req, mreq, NULL);
    } else {
      auth->IncRef();
      auth->Authorize(mreq->info().auth_req_,
                      mreq->mutable_auth_reply(),
                      NewCallback(this,
                                  &Whispercast::ProcessAuthorizedMediaRequest,
                                  req, mreq, auth));
    }
  }
}

void Whispercast::ProcessAuthorizedMediaRequest(
    http::ServerRequest* req,
    streaming::Request* mreq,
    streaming::Authorizer* auth) {
  if ( !mreq->auth_reply().allowed_ ) {
    LOG_INFO << "Unauthorized request: " << req->ToString();
    req->request()->server_data()->Write("Unauthorized");
    req->ReplyWithStatus(http::UNAUTHORIZED);
    auth->DecRef();
    delete mreq;
    return;
  }
  streaming::AuthorizeHelper* auth_helper = NULL;
  if ( auth != NULL ) {
    if ( mreq->auth_reply().reauthorize_interval_ms_ > 0 ) {
      auth_helper = new streaming::AuthorizeHelper(auth);
      *(auth_helper->mutable_req()) = mreq->info().auth_req_;
    }
    auth->DecRef();
  }

  URL* const url = req->request()->url();
  CHECK(url != NULL);
  const string url_path(url->UrlUnescape(url->path().c_str(),
                                         url->path().size()));
  // send the custom headers, if any
  const vector< pair<string, string> >&
      extra = mreq->serving_info().extra_headers_;
  if ( !extra.empty() ) {
    for ( int i = 0; i < extra.size(); ++i ) {
      req->request()->server_header()->AddField(extra[i].first,
                                                extra[i].second,
                                                true);
    }
  }

  // content type setting
  if ( mreq->serving_info().content_type_.empty() ) {
    const size_t dot_pos = url_path.rfind('.');
    if ( dot_pos != string::npos ) {
      const string ext(url_path.substr(dot_pos));
      map<string, string>::const_iterator it_ext = ext2ct_.find(ext);
      if ( it_ext != ext2ct_.end() ) {
        req->request()->server_header()->AddField(
            http::kHeaderContentType, it_ext->second, true);
      } else {
        req->request()->server_header()->AddField(
            http::kHeaderContentType, FLAGS_http_default_content_type, true);
      }
    } else {
      req->request()->server_header()->AddField(
          http::kHeaderContentType, FLAGS_http_default_content_type, true);
    }
  } else {
    req->request()->server_header()->AddField(
        http::kHeaderContentType, mreq->serving_info().content_type_, true);
  }

  // Will autodelete if StartRequest succeeds
  StreamRequest* const sreq = new StreamRequest(selector_,
                                                connection_id_++,
                                                stats_collector_,
                                                media_mapper_->mapper(),
                                                mreq, req, auth_helper);
  if ( !sreq->StartRequest() ) {
    delete sreq;
    req->ReplyWithStatus(http::NOT_FOUND);
    return;
  }
  // else, will auto-delete
}

int main(int argc, char* argv[]) {
  Whispercast app(argc, argv);
  return app.Main();
}