root/trunk/whisperer/stream_manager.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/io/file/file.h>
#include <whisperlib/common/io/file/file_output_stream.h>
#include <whisperlib/common/io/file/file_input_stream.h>

#include "stream_manager.h"

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

DEFINE_int32(max_lag,
             1024*1024,
             "The default maximum lag on the providers");

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

// our namespace
using namespace media;

static
http::HttpReturnCode ProviderResult2HTTPCode(
  media::StreamProvider::Result result) {
  switch (result) {
  case media::StreamProvider::NoError:
    return http::OK;
  case media::StreamProvider::PipelineParametersError:
    return http::BAD_REQUEST;
  default:
    return http::INTERNAL_SERVER_ERROR;
  }
}

bool StreamManager::ParseQueryString(
    const URL& url,
    bool& relay,
    bool& wrapped,
    media::Stream::Type& encoder_type,
    media::Encoder::Audio::Params& audio_params,
    media::Encoder::Video::Params& video_params,
    int& audio_port,
    int& video_port) {
  vector< pair<string, string> > query;
  url.GetQueryParameters(&query, true);

  parsable::Boolean relay_;
  if (!media::ParseFromValues(&relay_, "relay", query)) {
    LOG_ERROR << "Couldn't parse the relay flag from '" << url.spec() << "'.";
    return false;
  }
  if (!relay_.IsSet()) {
    relay = false;
  } else {
    relay = relay_;
  }
  parsable::Boolean wrapped_;
  if (!media::ParseFromValues(&wrapped_, "wrapped", query)) {
    LOG_ERROR << "Couldn't parse the wrapped flag from '" << url.spec() << "'.";
    return false;
  }
  if (!wrapped_.IsSet()) {
    wrapped = false;
  } else {
    wrapped = wrapped_;
  }

  encoder_type = media::Stream::Unknown;

  if (!ParseEncoderParameters(query, encoder_type,
                              audio_params, video_params,
                              audio_port, video_port)) {
    LOG_ERROR << "Couldn't parse the encoding parameters from '"
              << url.spec() << "'.";
    return false;
  }
  return true;
}

StreamManager::StreamManager(net::Selector* selector,
                             net::NetFactory* http_net_factory,
                             net::PROTOCOL http_net_protocol,
                             io::StateKeeper* state_keeper,
                             http::Server* http_server,
                             int http_server_port,
                             const http::ClientParams* http_client_params,
                             const char* media_root,
                             rpc::HttpServer* rpc_server,
                             const char* config_file) :
  ServiceInvokerMediaProviderService(
      ServiceInvokerMediaProviderService::GetClassName()),
  selector_(selector),
  http_net_factory_(http_net_factory),
  http_net_protocol_(http_net_protocol),
  state_keeper_(state_keeper),
  stream_server_(NULL),
  http_server_(http_server),
  rpc_server_(rpc_server),
  http_server_port_(http_server_port),
  http_client_params_(http_client_params),
  provider_media_root_((media_root == NULL) ? "" : media_root),
  config_file_((config_file == NULL) ? "" : config_file),
  running_(false) {
  CHECK_NOT_NULL(selector_);
}

StreamManager::~StreamManager() {
  CHECK_NULL(stream_server_);
}

void StreamManager::AddProvider(media::StreamProvider* provider) {
  CHECK(stream_providers_.find(provider->name()) == stream_providers_.end());
  stream_providers_[provider->name()] = provider;

  provider->Initialize(selector_);
  provider->SetMessageQueueCallback(
    NewPermanentCallback(this, &StreamManager::PostMessageCallback, provider));

  if (running_) {
    CHECK(provider->Start());
  }

  CHECK(stream_rpc_.find(provider) == stream_rpc_.end());

  if (rpc_server_ != NULL) {
    CHECK(rpc_server_->RegisterService(
        (std::string(provider->name()) + "/"), provider));
  }

  RPC rpc;
  stream_rpc_[provider] = rpc;
  LOG_INFO << "Provider '" << provider->name() << "' was added.";
}

void StreamManager::RemoveProvider(const char* name) {
  ProviderMap::iterator it = stream_providers_.find(name);
  if (it != stream_providers_.end()) {
    StreamProvider* provider = it->second;
    stream_providers_.erase(it);

    stream_providers_removed_.insert(provider);

    RemoveRPC(provider);
    provider->Stop();
  }
  AuthenticatorMap::iterator it_auth = authenticators_.find(name);
  if ( it_auth != authenticators_.end() ) {
    delete it_auth->second;
    authenticators_.erase(it_auth);
  }
  AuthenticatorUserMap::iterator it_auth_user = authenticators_users_.find(name);
  if ( it_auth_user != authenticators_users_.end() ) {
    delete it_auth_user->second;
    authenticators_users_.erase(it_auth_user);
  }
}

bool StreamManager::Publish(const URL& url,
                            int retry_timeout,
                            const string& user,
                            const string& pass) {
  CHECK(url.is_valid());

  StreamPoster* poster = new StreamPoster(this, http_client_params_);
  if ( poster->Start(
           url,
           retry_timeout,
           user, pass,
           NewCallback(this, &StreamManager::PosterDone, poster)) ) {
    published_[url.spec()] = poster;
    return true;
  }

  poster->Stop(); // will also auto-delete itself
  return false;
}
void StreamManager::Unpublish(const URL& url) {
  CHECK(url.is_valid());

  PosterMap::iterator poster = published_.find(url.spec());
  if (poster != published_.end()) {
    poster->second->Stop();
  }
}

bool StreamManager::RunRequest(
    StreamRequest* request,
    const char* name,
    bool relay,
    bool wrapped,
    media::Stream::Type encoder_type,
    const media::Encoder::Audio::Params* const audio_params,
    const media::Encoder::Video::Params* const video_params,
    int audio_port,
    int video_port) {
  request->selector_ = selector_;
  request->net_factory_ = http_net_factory_;
  request->net_protocol_ = http_net_protocol_;

  AuthenticatorMap::iterator it_auth = authenticators_.find(name);
  if ( it_auth != authenticators_.end() ) {
    // TODO(cpopescu): use the async interface at some point
    if ( !request->Authenticate(it_auth->second) ) {
      return false;
    }
  }

  ProviderMap::iterator provider = stream_providers_.find(name);
  if (provider == stream_providers_.end()) {
    request->Fail(http::BAD_REQUEST);
    return false;
  }

  media::StreamProvider::Result result;
  if ((result = provider->second->CreateSource()) !=
      media::StreamProvider::NoError) {
    request->Fail(ProviderResult2HTTPCode(result));
    return false;
  }

  int fd;
  const char* mime_type;

  if (!relay) {
    media::StreamProvider::Result result;
    if ((result = provider->second->CreateEncoder(encoder_type,
        audio_params, video_params, audio_port, video_port,
        wrapped, fd, mime_type)) != media::StreamProvider::NoError) {
      request->Fail(ProviderResult2HTTPCode(result));
      return false;
    }
  } else {
    // we need to get the source's mime-type, as we cannot determine it
    mime_type = provider->second->source_mime_type().c_str();

    if ((result = provider->second->CreateRelay(wrapped, fd)) !=
        media::StreamProvider::NoError) {
      request->Fail(ProviderResult2HTTPCode(result));
      return false;
    }
  }

  request->mime_type_ = mime_type;

  if (provider->second->max_lag() == INT_MIN)
    request->max_lag_ = FLAGS_max_lag;
  else
    request->max_lag_ = provider->second->max_lag();
  CHECK_GT(request->max_lag_, 0);

  CHECK_NULL(request->reader_);
  request->reader_ = new net::SelectableFilereader(selector_);

  request->wrapped_ = wrapped;
  if (wrapped) {
    request->processor_ = new StreamFrameProcessor();
  } else {
    request->processor_ = new StreamCopyProcessor();
  }

  if (!request->reader_->InitializeFd(
        fd,
        NewPermanentCallback(request, &StreamRequest::ReaderProcess_),
        NewCallback(request, &StreamRequest::ReaderClose_)) ) {
    ::close(fd);

    delete request->reader_;
    request->reader_ = NULL;
    delete request->processor_;
    request->processor_ = NULL;

    request->Fail(http::INTERNAL_SERVER_ERROR);
    return false;
  }

  request->Start();
  return true;
}

void StreamManager::Run() {
  running_ = true;
  LoadConfig();

  stream_server_ = new StreamServer(this, http_server_, provider_media_root_.c_str());

  http_server_->AddAcceptor(http_net_protocol_,
                            net::HostPort(0, http_server_port_));
  selector_->RunInSelectLoop(
    NewCallback(http_server_, &http::Server::StartServing));

  selector_->Loop();
  CHECK(!running_);

  CHECK(stream_rpc_.empty());
  CHECK(stream_providers_.empty());
  CHECK(authenticators_.empty());
  CHECK(authenticators_users_.empty());
}

void StreamManager::Stop() {
  running_ = false;
  if (!CheckTerminate()) {
    LOG_INFO << "Stopping the published streams...";
    for ( PosterMap::iterator poster = published_.begin();
         poster != published_.end(); ++poster ) {
      poster->second->Stop();
    }
    LOG_INFO << "Stopping the providers...";
    for ( ProviderMap::iterator provider = stream_providers_.begin();
          provider != stream_providers_.end(); ++provider ) {

      RemoveRPC(provider->second);
      provider->second->Stop();
    }
  }

  http_server_->StopServing();

  delete stream_server_;
  stream_server_ = NULL;

  // The manager will asynchronously stop and shutdown when
  // all the providers/posts are terminated...
}

bool StreamManager::CheckTerminate() {
  if (!running_) {
    if ( stream_providers_.empty() &&
         stream_providers_removed_.empty() &&
         published_.empty() ) {
      LOG_INFO << "No providers or publishing requests are running, "
               << "stopping selector";
      selector_->RunInSelectLoop(
        NewCallback(selector_, &net::Selector::MakeLoopExit));
      return true;
    }
  }
  return false;
}

void StreamManager::ProviderDone(StreamProvider* provider) {
  // make sure we're removed...
  RemoveRPC(provider);

  AuthenticatorMap::iterator it_auth =
      authenticators_.find(provider->name());
  if ( it_auth != authenticators_.end() ) {
    delete it_auth->second;
    authenticators_.erase(it_auth);
  }
  AuthenticatorUserMap::iterator it_auth_user =
      authenticators_users_.find(provider->name());
  if ( it_auth_user != authenticators_users_.end() ) {
    delete it_auth_user->second;
    authenticators_users_.erase(it_auth_user);
  }

  stream_providers_.erase(provider->name());
  const ProviderSet::iterator it = stream_providers_removed_.find(provider);
  if ( it != stream_providers_removed_.end() ) {
    provider->DeleteState();
    stream_providers_removed_.erase(it);
  }

  LOG_INFO << "Provider '" << provider->name() << "' is done.";
  delete provider;
  CheckTerminate();
}

void StreamManager::PosterDone(StreamPoster* poster) {
  published_.erase(poster->url().spec());
  LOG_INFO << "Stopped publishing to [" << poster->url() << "].";

  CheckTerminate();
}

void StreamManager::RemoveRPC(StreamProvider* provider) {
  // remove RPC
  RpcMap::iterator rpc = stream_rpc_.find(provider);
  if (rpc != stream_rpc_.end()) {
    CHECK_NOT_NULL(rpc_server_);
    rpc_server_->UnregisterService(
      (std::string(provider->name()) + "/"), provider);
    stream_rpc_.erase(rpc);
  }
}

bool StreamManager::LoadConfig() {
  if (config_file_.empty()) {
    LOG_ERROR << "Config file was not set, state loading failed...";
    return false;
  }
  string config_str;
  if ( !io::FileInputStream::TryReadFile(config_file_.c_str(),
                                         &config_str) ) {
    LOG_ERROR << "Cannot open the config file [" << config_file_ << "]";
    return false;
  }
  io::MemoryStream iomis;
  iomis.Write(config_str);
  rpc::JsonDecoder decoder(iomis);

  bool result = true;
  rpc::Array< ProviderSpec > providers;

  if (decoder.Decode(providers) == rpc::DECODE_RESULT_SUCCESS) {
    for (uint32 i = 0; i < providers.Size(); ++i) {
      MediaServerResult ret;
      if (!AddProvider(&providers.Get(i), ret)) {
        LOG_ERROR
          << "Couldn't add a provider while loading the configuration from ["
          << config_file_ << "], error: \""
          << ret.description_.Get().StdStr() << "\"...";
      }
    }
  } else {
    LOG_ERROR << "Couldn't load the providers from ["
              << config_file_ << "]...";
    result = false;
  }

  rpc::Array< PublishSpec > published;
  if (decoder.Decode(published) == rpc::DECODE_RESULT_SUCCESS) {
    for (uint32 i = 0; i < published.Size(); ++i) {
      MediaServerResult ret;
      if (!AddPublished(&published.Get(i), ret)) {
        LOG_ERROR
          << "Couldn't publish a stream while loading the configuration from ["
          << config_file_ << "], error: \""
          << ret.description_.Get().StdStr() << "\"...";
      }
    }
  } else {
    LOG_ERROR << "Couldn't load the published streams from ["
              << config_file_ << "]...";
    result = false;
  }
  return result;
}
// Saves the state into the config file.
bool StreamManager::SaveConfig() {
  if (config_file_.empty()) {
    LOG_ERROR << "Config file was not set, state saving failed...";
    return false;
  }

  const string tmp_file(config_file_ + "__temp__");

  io::File* const f = new io::File();
  if (!f->Open(tmp_file.c_str(),
                io::File::GENERIC_READ_WRITE,
                io::File::CREATE_ALWAYS)) {
    LOG_ERROR << "Cannot open the temporary config file ["
              << tmp_file << "] for writing...";
    delete f;
    return false;
  }

  {
    io::FileOutputStream fos(f);
    {
      rpc::Array< ProviderSpec > providers;
      ListProviders(providers);
      fos.WriteString(rpc::JsonEncoder::EncodeObject(providers));
    }
    {
      rpc::Array< PublishSpec > published;
      ListPublished(published);
      fos.WriteString(rpc::JsonEncoder::EncodeObject(published));
    }
  }

  if (rename(tmp_file.c_str(), config_file_.c_str())) {
    LOG_ERROR << "Cannot move the temporary config file ["
              << tmp_file << "] over the configuration file ["
              << config_file_ << "]...";
    return false;
  }
  return true;
}

void StreamManager::ProcessMessageCallback(
  media::StreamProvider* provider) {
  DLOG_DEBUG << "Iterating message for '" << provider->name() << "'.";
  if (!provider->Iterate()) {
    ProviderDone(provider);
  }
}
void StreamManager::PostMessageCallback(
  media::StreamProvider* provider,
  Message* message) {
  DLOG_DEBUG << "Scheduling message iteration for '"
             << provider->name() << "' - " << message->m_kind << ".";
  selector_->RunInSelectLoop(
    NewCallback(this, &StreamManager::ProcessMessageCallback, provider));
}

bool StreamManager::AddProvider(const ProviderSpec* spec,
                                MediaServerResult& ret) {
  if (!running_) {
    ret.result_.Ref() = -1;
    ret.description_.Ref() = "Cannot add a provider when the server is not running.";
    return false;
  }
  if (spec->name_.Get().StdStr().empty() ||
      spec->description_.Get().StdStr().empty()) {
    ret.result_.Ref() = -4;
    ret.description_.Ref() = "Both the name and the description must be provided.";
    return false;
  }

  if (stream_providers_.find(spec->name_.Get().StdStr()) !=
    stream_providers_.end()) {
    ret.result_.Ref() = -3;
    ret.description_.Ref() = "The provider already exists.";
    return false;
  }

  // convert the controlled element to what we need
  vector<string> controlled_elements;
  if (spec->controlled_elements_.IsSet()) {
    for (uint32 i = 0; i < spec->controlled_elements_.Get().Size(); i++)
    controlled_elements.push_back(
        spec->controlled_elements_.Get().Get(i).StdStr());
  }
  if ( spec->access_data_.IsSet() && spec->access_data_.Get().Size() > 0 ) {
    net::SimpleUserAuthenticator*
        user_authenticator = new net::SimpleUserAuthenticator(
            spec->name_.Get().StdStr());
    map<string, string>* const users = new map<string, string>;
    for ( int i = 0; i < spec->access_data_.Get().Size(); ++i ) {
      user_authenticator->set_user_password(
          spec->access_data_.Get().Get(i).remote_user_.Get().StdStr(),
          spec->access_data_.Get().Get(i).remote_password_.Get().StdStr());
      users->insert(
          make_pair(
              spec->access_data_.Get().Get(i).remote_user_.Get().StdStr(),
              spec->access_data_.Get().Get(i).remote_password_.Get().StdStr()));
    }
    authenticators_.insert(make_pair(spec->name_.Get().StdStr(),
                                     user_authenticator));
    authenticators_users_.insert(make_pair(spec->name_.Get().StdStr(),
                                           users));
  }

  // perform
  AddProvider(
    new StreamProvider(
      // name
      spec->name_.Get().StdStr().c_str(),
      // description
      spec->description_.Get().StdStr().c_str(),
      // audio ports
      spec->audio_ports_.IsSet() ?
      spec->audio_ports_.Get().Get() : 1,
      // video ports
      spec->video_ports_.IsSet() ?
      spec->video_ports_.Get().Get() : 1,
      // mime type
      spec->mime_type_.IsSet() ?
      spec->mime_type_.Get().StdStr().c_str() : "",
      // source fireup timeout
      spec->source_fireup_timeout_.IsSet() ?
      spec->source_fireup_timeout_.Get().Get() : INT_MIN,
      // encoder fireup timeout
      spec->encoder_fireup_timeout_.IsSet() ?
      spec->encoder_fireup_timeout_.Get().Get() : INT_MIN,
      // source idle timeout
      spec->source_idle_timeout_.IsSet() ?
      spec->source_idle_timeout_.Get().Get() : INT_MIN,
      // encoder idle timeout
      spec->encoder_idle_timeout_.IsSet() ?
      spec->encoder_idle_timeout_.Get().Get() : INT_MIN,
      // maximum acceptable lag
      spec->max_lag_.IsSet() ?
      spec->max_lag_.Get().Get() : INT_MIN,
      &controlled_elements,
      state_keeper_));
  ret.result_.Ref() = 0;
  return true;
}

void StreamManager::ListProviders(rpc::Array<ProviderSpec>& providers) {
  for ( ProviderMap::iterator provider = stream_providers_.begin();
       provider != stream_providers_.end(); ++provider ) {
    ProviderSpec spec;

    spec.name_.Ref() = provider->first;
    spec.description_.Ref() = provider->second->source_pipeline_description();

    if (provider->second->source_mime_type() != "")
      spec.mime_type_.Ref() = provider->second->source_mime_type();

    if (provider->second->source_fireup_timeout() != INT_MIN)
      spec.source_fireup_timeout_.Ref() =
        provider->second->source_fireup_timeout();
    if (provider->second->encoder_fireup_timeout() != INT_MIN)
      spec.encoder_fireup_timeout_.Ref() =
        provider->second->encoder_fireup_timeout();

    if (provider->second->source_idle_timeout() != INT_MIN)
      spec.source_idle_timeout_.Ref() =
        provider->second->source_idle_timeout();
    if (provider->second->encoder_idle_timeout() != INT_MIN)
      spec.encoder_idle_timeout_.Ref() =
        provider->second->encoder_idle_timeout();

    if (provider->second->max_lag() != INT_MIN)
      spec.max_lag_.Ref() =
        provider->second->max_lag();

    const media::StreamProvider::ControllerMap&
      controllers = provider->second->controllers();
    if (!controllers.empty()) {
      rpc::Array<rpc::String>& c = spec.controlled_elements_.Ref();
      for ( media::StreamProvider::ControllerMap::const_iterator
              it = controllers.begin(); it != controllers.end(); ++it ) {
        c.Append(it->first);
      }
    }
    const AuthenticatorUserMap::const_iterator
        it_auth = authenticators_users_.find(provider->second->name());
    if ( it_auth != authenticators_users_.end() ) {
      rpc::Array<WhispererSpecAccessStruct>& acc = spec.access_data_.Ref();
      for ( map<string, string>::const_iterator it_user = it_auth->second->begin();
            it_user != it_auth->second->end(); ++it_user ) {
        acc.Append(WhispererSpecAccessStruct(it_user->first, it_user->second));
      }
    }
    providers.PushBack(spec);
  }
}

bool StreamManager::AddPublished(
  const PublishSpec* spec,
  MediaServerResult& ret) {
  if ( !running_ ) {
    ret.result_.Ref() = -1;
    ret.description_.Ref() = "Cannot publish when the server is not running.";
    return false;
  }
  if (spec->url_.Get().StdStr().empty()) {
    ret.result_.Ref() = -4;
    ret.description_.Ref() = "The URL to publish to must be provided.";
    return false;
  }
  URL url(spec->url_.Get().StdStr());
  if (!url.is_valid()) {
    ret.result_.Ref() = -4;
    ret.description_.Ref() = "The URL to publish to is invalid.";
    return false;
  }
  if (published_.find(url.spec()) !=
    published_.end()) {
    ret.result_.Ref() = -3;
    ret.description_.Ref() = "Publishing to the given URL is already running.";
    return false;
  }
  const string user(spec->user_.IsSet()
                    ? spec->user_.Get().remote_user_.Get().CStr() : "" );
  const string pass(spec->user_.IsSet()
                    ? spec->user_.Get().remote_password_.Get().CStr() : "" );
  if ( !Publish(url,  // encoder fireup timeout
                spec->retry_timeout_.IsSet() ?
                spec->retry_timeout_.Get().Get() : INT_MIN,
                user, pass) ) {
    ret.result_.Ref() = -4;
    ret.description_.Ref() = ("Couldn't publish to the given URL "
                              "(probably the URL is invalid).");
    return false;
  }
  ret.result_.Ref() = 0;
  return true;
}

void StreamManager::ListPublished(rpc::Array<PublishSpec>& published) {
  for ( PosterMap::iterator poster = published_.begin();
        poster != published_.end(); ++poster ) {
    PublishSpec spec;
    spec.url_.Ref() = poster->first;
    if (poster->second->retry_timeout() != INT_MIN) {
      spec.retry_timeout_.Ref() =
        poster->second->retry_timeout();
    }
    if ( !poster->second->user().empty() || !poster->second->pass().empty() ) {
      spec.user_.Ref() = WhispererSpecAccessStruct(poster->second->user(),
                                                   poster->second->pass());
    }

    published.PushBack(spec);
  }
}

// Implementation of ServiceInvokerMediaProviderService
void StreamManager::AddProvider(
  rpc::CallContext<MediaServerResult>* call,
  const ProviderSpec* spec) {
  MediaServerResult ret;
  AddProvider(spec, ret);
  call->Complete(ret);
}

void StreamManager::RemoveProvider(
    rpc::CallContext< MediaServerResult >* call,
    const rpc::String* name) {
  MediaServerResult ret;
  if ( !running_ ) {
    ret.result_.Ref() = -1;
    ret.description_.Ref() =
      "Cannot remove a provider while the server is not running.";
    call->Complete(ret);
    return;
  }
  if ( stream_providers_.find(name->StdStr()) == stream_providers_.end() ) {
    ret.result_.Ref() = -2;
    ret.description_.Ref() = "The provider does not exist.";
    call->Complete(ret);
    return;
  }
  RemoveProvider(name->StdStr().c_str());
  ret.result_.Ref() = 0;
  call->Complete(ret);
}

void StreamManager::ListProviders(
  rpc::CallContext< rpc::Array< ProviderSpec > >* call) {
  rpc::Array< ProviderSpec > ret;
  ListProviders(ret);
  call->Complete(ret);
}
void StreamManager::Publish(
  rpc::CallContext<MediaServerResult>* call, const PublishSpec* spec) {
  MediaServerResult ret;
  AddPublished(spec, ret);
  call->Complete(ret);
}
void StreamManager::Unpublish(
  rpc::CallContext<MediaServerResult>* call, const rpc::String* url) {
  MediaServerResult ret;
  if (!running_) {
    ret.result_.Ref() = -1;
    ret.description_.Ref() =
        "Cannot unpublish while the server is not running.";
    call->Complete(ret);
    return;
  }
  URL gurl(url->StdStr());
  if (!gurl.is_valid()) {
    ret.result_.Ref() = -4;
    ret.description_.Ref() = "The URL to stop publishing to is invalid.";
    return;
  }
  if (published_.find(gurl.spec()) == published_.end()) {
    ret.result_.Ref() = -2;
    ret.description_.Ref() =
      "There is no publishing performed to the given URL.";
    call->Complete(ret);
    return;
  }
  Unpublish(gurl);
  ret.result_.Ref() = 0;
  call->Complete(ret);
}

void StreamManager::ListPublished(
    rpc::CallContext< rpc::Array< PublishSpec > >* call) {
  rpc::Array< PublishSpec > ret;
  ListPublished(ret);
  call->Complete(ret);
}

void StreamManager::SaveConfig(rpc::CallContext<rpc::Bool>* call) {
  call->Complete(SaveConfig());
}