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

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

#include "stream_manager.h"

// our namespace
using namespace media;

StreamServer::StreamServer(StreamManager* manager,
                           http::Server* http_server,
                           const char* streaming_root) :
  manager_(manager),
  http_server_(http_server),
  root_((streaming_root == NULL) ? "" : streaming_root) {
  CHECK_NOT_NULL(manager_);
  CHECK_NOT_NULL(http_server_);

  http_server_->RegisterProcessor(
    root_, NewPermanentCallback(this, &StreamServer::CreateCallback), true);
}

StreamServer::~StreamServer() {
  http_server_->UnregisterProcessor(root_);
}

void StreamServer::CreateCallback(http::ServerRequest* request) {
  URL* const url = request->request()->url();
  CHECK_NOT_NULL(url);
  DLOG_DEBUG << "Creating encoder for '" << url->spec() << "'.";

  const string url_path(
    url->UrlUnescape(url->path().c_str(), url->path().size()));

  // no caching, ever
  request->request()->server_header()->AddField(
      "Cache-Control", "no-cache, must-revalidate, max-age=0", true);

  // check the method, we only accept GET
  if (request->request()->client_header()->method() != http::METHOD_GET) {
    request->ReplyWithStatus(http::METHOD_NOT_ALLOWED);
    return;
  }

  static const string root = root_ + "/";
  if (!strutil::StrPrefix(url_path.c_str(), root.c_str())) {
    request->ReplyWithStatus(http::NOT_FOUND);
    return;
  }

  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;
  if (!StreamManager::ParseQueryString(
        *url, relay, wrapped, encoder_type, audio_params, video_params,
        audio_port, video_port)) {
    request->ReplyWithStatus(http::BAD_REQUEST);
    return;
  }

  // run the request
  manager_->RunRequest(new ServerRequest(request),
                       url_path.substr(root.length()).c_str(),
                       relay,
                       wrapped,
                       encoder_type,
                       &audio_params, &video_params,
                       audio_port, video_port);
}

// Implementation of StreamServer::ServerRequest

void StreamServer::ServerRequest::Start() {
  http::Header* const hs = request_->request()->server_header();
  if (wrapped_) {
    hs->AddField(http::kHeaderContentType, streaming::kInternalMimeType, true);
  } else {
    hs->AddField(http::kHeaderContentType, mime_type_.c_str(), true);
  }

  request_->BeginStreamingData(http::OK,
      NewCallback(this, &StreamServer::ServerRequest::RequestProcess),
      NewCallback(this, &StreamServer::ServerRequest::RequestClose));
}
void StreamServer::ServerRequest::Fail(http::HttpReturnCode result) {
  request_->ReplyWithStatus(result);
  selector_->DeleteInSelectLoop(this);
}

void StreamServer::ServerRequest::ReaderProcess(io::MemoryStream* inbuf) {
  RequestProcess();
}
void StreamServer::ServerRequest::ReaderClose() {
  DLOG_DEBUG << "Filereader closed for '"
             << request_->request()->url()->spec() << "'";
  selector_->DeleteInSelectLoop(reader_);
  reader_ = NULL;

  if (request_valid_) {
    request_valid_ = false;
    request_->EndStreamingData();
    return;
  }
  selector_->DeleteInSelectLoop(this);
}

void StreamServer::ServerRequest::RequestProcess() {
  CHECK_NOT_NULL(request_);
  if (request_->is_orphaned()) {
    if (reader_ != NULL) {
      reader_->Close();
    }
    return;
  }

  CHECK_NOT_NULL(reader_);

  int32 available_out;
  available_out = request_->free_output_bytes();

  CHECK_NOT_NULL(processor_);

  bool send_frame;
  if (!processor_->Process(request_->request()->server_data(),
                           reader_->inbuf(), available_out, send_frame)) {
    LOG_ERROR << "The processor returned failure for "
              << request_->request()->url()->spec()
              << "', terminating the request.";
    reader_->Close();
    return;
  }

  if (send_frame) {
    request_->ContinueStreamingData(
      NewCallback(this, &StreamServer::ServerRequest::RequestProcess));
  }
}
void StreamServer::ServerRequest::RequestClose() {
  DLOG_DEBUG << "Request closed for '"
             << request_->request()->url()->spec() << "'.";
  if (request_valid_) {
    request_valid_ = false;
    request_->EndStreamingData();
  }

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