root/trunk/whisperer/stream_poster.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 <netdb.h>

#include <whisperlib/common/base/common.h>
#include "stream_poster.h"

#include "stream_manager.h"

// our namespace
using namespace media;

DEFINE_int32(publishing_retry_timeout, 1000,
    "The default stream publishing request timeout");

StreamPoster::StreamPoster(StreamManager *manager,
    const http::ClientParams *http_client_params) :
  manager_(manager),
  http_client_params_(http_client_params),
  done_callback_(NULL),
  request_(NULL),
  started_(false),
  stopped_(false) {
  CHECK_NOT_NULL(manager_);
  CHECK_NOT_NULL(http_client_params_);
}
StreamPoster::~StreamPoster() {
  if (done_callback_)
    done_callback_->Run();
}

bool StreamPoster::Start(const URL& url,
                         int retry_timeout,
                         const string& user,
                         const string& pass,
                         Closure* done_callback) {
  url_ = url;
  retry_timeout_ = retry_timeout;
  user_ = user;
  pass_ = pass;

  if (!StreamManager::ParseQueryString(
      url_,
      relay_,
      wrapped_,
      encoder_type_,
      audio_params_,
      video_params_,
      audio_port_,
      video_port_)) {
    LOG_ERROR << "Couldn't parse the parameters from [" << url_ << "].";
    return false;
  }

  // run the request
  LOG_DEBUG << "Scheduling a publishing POST to [" << url_ << "].";
  manager_->selector()->RunInSelectLoop(
      NewCallback(this, &StreamPoster::Run));
  started_ = true;

  done_callback_ = done_callback;
  return true;
}
void StreamPoster::Stop() {
  if (started_) {
    stopped_ = true;
    if (request_) {
      request_->Stop();
    }
  } else {
    delete this;
  }
}

void StreamPoster::Run() {
  net::HostPort server(url_.host().c_str(),
                       url_.IntPort() == -1 ? 80 : url_.IntPort());

  CHECK_NULL(request_);
  request_ = new PosterRequest(url_, server,
                               user_, pass_,
                               http_client_params_,
                               NewCallback(this, &StreamPoster::Done));

  LOG_INFO << "Running a publishing POST to [" << url_ << "].";
  manager_->RunRequest(
      request_,
      url_.ExtractFileName().c_str(),
      relay_, wrapped_, encoder_type_, &audio_params_, &video_params_,
      audio_port_, video_port_);
}
void StreamPoster::Done() {
  CHECK_NOT_NULL(request_);
  request_ = NULL;

  LOG_INFO << "The publishing POST to [" << url_ << "] has finished.";
  if (stopped_) {
    delete this;
    return;
  }

  int retry_timeout = (retry_timeout_ == INT_MIN) ?
  FLAGS_publishing_retry_timeout : retry_timeout_;

  if (retry_timeout > 0) {
    LOG_INFO <<
    "Restarting the publishing POST to [" <<
    url_ <<
    "] in " << retry_timeout << " miliseconds.";

    manager_->selector()->RegisterAlarm(
        NewCallback(this, &StreamPoster::Run), retry_timeout);
  } else {
    LOG_INFO <<
    "Restarting the publishing POST to [" << url_ << "].";
    manager_->selector()->RunInSelectLoop(
        NewCallback(this, &StreamPoster::Run));
  }
}

// Implementation of StreamPoster::ServerRequest

void StreamPoster::PosterRequest::Start() {
  CHECK_NULL(request_);
  request_ = new http::ClientRequest(http::METHOD_POST, &url_);

  CHECK_NULL(protocol_);
  protocol_ = new http::ClientStreamingProtocol(http_client_params_,
    new http::SimpleClientConnection(selector_, net_factory_, net_protocol_),
    server_);

  http::Header* const hs = request_->request()->client_header();
  hs->SetChunkedTransfer(true);
  if ( !user_.empty() || !pass_.empty() ) {
    hs->SetAuthorizationField(user_, pass_);
  }

  if (wrapped_) {
    hs->AddField(http::kHeaderContentType, streaming::kInternalMimeType, true);
  } else {
    hs->AddField(http::kHeaderContentType, mime_type_.c_str(), true);
  }

  protocol_->BeginStreaming(request_,
      NewPermanentCallback(this, &StreamPoster::PosterRequest::RequestProcess));
}
void StreamPoster::PosterRequest::Fail(http::HttpReturnCode result) {
  selector_->DeleteInSelectLoop(this);
}

void StreamPoster::PosterRequest::ReaderProcess(io::MemoryStream* inbuf) {
  protocol_->UnpauseWriting();
}
void StreamPoster::PosterRequest::ReaderClose() {
  reader_ = NULL;
}

bool StreamPoster::PosterRequest::RequestProcess(int32 available_out) {
  DLOG_DEBUG << "RequestProcess() '" << url_ << "'...";
  if (available_out < 0) {  // terminated
    if (reader_ != NULL) {
      reader_->Close();
    }
    selector_->DeleteInSelectLoop(this);
    return false;
  }

  if (reader_ == NULL) {
    selector_->DeleteInSelectLoop(this);
    return false;
  }

  CHECK_NOT_NULL(processor_);

  while (available_out > 0) {
    // TODO(mihai): allow specification of chunk size
    int32 to_send = min(8192, available_out);

    bool send_frame;
    if ( !processor_->Process(request_->request()->client_data(),
                              reader_->inbuf(), to_send, send_frame) ) {
      LOG_ERROR << "The processor returned failure for the publishing "
                << " POST to [" << url_.spec() << "], terminating the request.";
      reader_->Close();
      return false;
    }

    available_out -= to_send;
  }
  return true;
}