root/trunk/whisperer/media_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: Mihai Ianculescu

#include <whisperlib/common/base/common.h>
#include <whisperlib/common/base/errno.h>
#include <whisperlib/common/base/signal_handlers.h>
#include <whisperlib/common/base/system.h>

#include <whisperlib/net/base/user_authenticator.h>
#include <whisperlib/common/app/app.h>
#include <whisperlib/net/rpc/lib/server/rpc_http_server.h>

#include "stream_manager.h"

// our namespace
using namespace media;

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

DEFINE_int32(http_port,
             80,
             "The port on which we accept HTTP connections");
DEFINE_bool(http_connection_dlog_level,
            false,
            "Turn on deep debugging messages on HTTP connections");
DEFINE_string(media_root,
              "/media",
              "The path on the HTTP server that serves as root for "
              "the providers");
DEFINE_string(media_config_dir,
              "",
              "Where to load / save the configs");
DEFINE_string(media_state_dir,
              "",
              "Where we may save our state");
DEFINE_string(media_state_name,
              "whisperer",
              "We save the 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))' )");
//////////////////////////////////////////////////////////////////////

class App : public app::App {
 public:
  App(int argc, char** &argv) :
      app::App(argc, argv),
      selector_(NULL),
      http_net_factory_(NULL),
      http_server_(NULL),
      state_keeper_(NULL),
      stream_manager_(NULL),
      rpc_server_(NULL),
      admin_authenticator_(NULL) {
    google::SetUsageMessage("Whisperer - stream providing server.");
  }
  virtual ~App() {
    CHECK_NULL(selector_);
    CHECK_NULL(http_net_factory_);
    CHECK_NULL(http_server_);
    CHECK_NULL(stream_manager_);
    CHECK_NULL(state_keeper_);
    CHECK_NULL(rpc_server_);
  }

 protected:
  int Initialize() {
    // initialize streaming (this will also call common::Init())
    media::Init(argc_, argv_);

    selector_ = new net::Selector();

    http_server_params_.max_num_chunks_ = -1;
    http_server_params_.dlog_level_ = FLAGS_http_connection_dlog_level;

    http_client_params_.max_chunk_size_ = 1024;
    http_client_params_.dlog_level_ = FLAGS_http_connection_dlog_level;

    http_net_factory_ = new net::NetFactory(selector_);

    http_server_ = new http::Server("Whispercast Media Server"VERSION_TEXT,
                                    selector_,
                                    http_net_factory_,
                                    &http_server_params_,
                                    NewPermanentCallback(
                                      &http::SimpleServerConnectionFactory),
                                    0);
    if (!FLAGS_media_state_dir.empty()) {
      state_keeper_ =
      new io::StateKeeper(FLAGS_media_state_dir.c_str(),
          FLAGS_media_state_name.c_str());
      state_keeper_->Initialize();
    } else {
      state_keeper_ = NULL;
    }

    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_server_ = new rpc::HttpServer(selector_,
                                      http_server_,
                                      admin_authenticator_,
                                      "/rpc-config",
                                      true,
                                      true, 10,
                                      FLAGS_rpc_config_allow_classifier);

    stream_manager_ = new StreamManager(
        selector_,
        http_net_factory_,
        net::PROTOCOL_TCP,
        state_keeper_,
        http_server_,
        FLAGS_http_port,
        &http_client_params_,
        FLAGS_media_root.c_str(),
        rpc_server_,
        (FLAGS_media_config_dir + "/whisperer.config").c_str());

    CHECK(rpc_server_->RegisterService(stream_manager_));
    return 0;
  }

  void Run() {
    stream_manager_->Run();
  }
  void Stop() {
    selector_->RunInSelectLoop(NewCallback(stream_manager_,
                                           &StreamManager::Stop));
  }

  void Shutdown() {
    delete rpc_server_;
    rpc_server_ = NULL;

    delete stream_manager_;
    stream_manager_ = NULL;

    delete state_keeper_;
    state_keeper_ = NULL;

    delete http_server_;
    http_server_ = NULL;

    delete http_net_factory_;
    http_net_factory_ = NULL;

    delete selector_;
    selector_ = NULL;

    delete admin_authenticator_;
    admin_authenticator_ = NULL;

    // uninitialize streaming
    media::Shutdown();
  }

 private:
  net::Selector* selector_;

  http::ServerParams http_server_params_;
  http::ClientParams http_client_params_;

  net::NetFactory* http_net_factory_;
  http::Server* http_server_;

  io::StateKeeper* state_keeper_;
  StreamManager* stream_manager_;

  rpc::HttpServer* rpc_server_;
  net::SimpleUserAuthenticator* admin_authenticator_;
};

int main(int argc, char* argv[]) {
  return common::Exit(App(argc, argv).Main());
}

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