root/trunk/whispercast/stream_request.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

#include <vector>
#include "stream_request.h"
#include <whisperlib/net/util/ipclassifier.h>

#include <whisperstreamlib/internal/internal_frame.h>

#include <whisperstreamlib/aac/aac_tag_splitter.h>
#include <whisperstreamlib/flv/flv_tag_splitter.h>
#include <whisperstreamlib/mp3/mp3_tag_splitter.h>
#include <whisperstreamlib/internal/internal_tag_splitter.h>
#include <whisperstreamlib/raw/raw_tag_splitter.h>

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

DECLARE_int32(http_port);
DECLARE_int64(http_connection_write_ahead_ms);
DECLARE_int32(http_connection_max_media_outbuf_size);

DEFINE_bool(stream_log_detailed_media_tags,
            false,
            "Log each media tag that we send -- really detailed :)");

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

static
const char* ResultName(http::HttpReturnCode result) {
  if ( result >= 200 && result < 300 )
    return "SUCCESS";
  if ( result >= 300 && result < 400 )
    return "REDIRECTED";
  if ( result >= 500 )
    return "SERVER ERROR";

  // We are left with 4xx and strange stuff
  switch ( result ) {
  case http::BAD_REQUEST:
    return "INVALID REQUEST";
  case http::UNAUTHORIZED:
  case http::PAYMENT_REQUIRED:
  case http::FORBIDDEN:
  case http::METHOD_NOT_ALLOWED:
  case http::NOT_ACCEPTABLE:
  case http::PROXY_AUTHENTICATION_REQUIRED :
    return "UNAUTHORIZED";
  case http::NOT_FOUND:
    return "NOT FOUND";
  case http::REQUEST_TIME_OUT:
    return "TIMEOUT";
  default:
    return "INVALID REQUEST";
  }
  return "UNKNOWN";
}

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

StreamRequest::StreamRequest(
    net::Selector* selector,
    int64 connection_id,
    streaming::StatsCollector* collector,
    streaming::ElementMapper* mapper,
    streaming::Request* media_request,
    http::ServerRequest* http_request,
    streaming::AuthorizeHelper* auth_helper)
    : selector_(selector),
      mapper_(mapper),
      stats_collector_(collector),
      media_request_(media_request),
      http_request_(http_request),
      auth_helper_(auth_helper),
      element_seq_id_(0),
      processing_callback_(
          NewPermanentCallback(this, &StreamRequest::ProcessTag)),
      tag_type_(streaming::INVALID_STREAM_TYPE),
      serializer_(NULL),
      unpause_buffer_callback_(NULL),
      is_first_tag_(true),
      is_audio_dropping_(false),
      is_video_dropping_(false),
      first_tag_time_(0),
      last_tag_stream_time_(0),
      media_id_(0),
      start_time_(selector->now()),
      reauthorize_callback_(NULL),
      terminate_callback_(NULL) {
  // Prepare connection stats
  const int64 now = timer::Date::Now();
  connection_begin_stats_.connection_id_ =
      strutil::StringPrintf("%lld", static_cast<long long int>(connection_id));
  connection_begin_stats_.timestamp_utc_ms_ = now;
  connection_begin_stats_.remote_host_ =
      http_request_->remote_address().ip_object().ToString();
  connection_begin_stats_.remote_port_ =
      http_request_->remote_address().port();
  connection_begin_stats_.local_host_ = "127.0.0.1";
  connection_begin_stats_.local_port_ = FLAGS_http_port;
  connection_begin_stats_.protocol_ = "http";

  // Prepare connection end stats for incrementing things..
  connection_end_stats_.connection_id_ = connection_begin_stats_.connection_id_;
  connection_end_stats_.timestamp_utc_ms_ = now;
  connection_end_stats_.bytes_up_ = 0;
  connection_end_stats_.bytes_down_ = 0;
  connection_end_stats_.result_ = ConnectionResult("ACTIVE");

  // Prepare stream stats (one stream per connection .. so far...)
  stream_begin_stats_.stream_id_ = strutil::StringPrintf("%s.%lld",
      connection_begin_stats_.connection_id_.Get().CStr(), 0LL);
  stream_begin_stats_.timestamp_utc_ms_ = now;
  stream_begin_stats_.connection_ = connection_begin_stats_;
  stream_begin_stats_.session_id_ = media_request_->info().session_id_;
  stream_begin_stats_.affiliate_id_ = media_request_->info().affiliate_id_;
  stream_begin_stats_.client_id_ = media_request_->info().client_id_;
  stream_begin_stats_.user_agent_ =
    http_request_->request()->client_header()->FindField(
          http::kHeaderUserAgent);

  // Prepare stream end stats
  stream_end_stats_.stream_id_ = stream_begin_stats_.stream_id_;
  stream_end_stats_.timestamp_utc_ms_ = now;
  stream_end_stats_.result_ = StreamResult("ACTIVE");

  // Mark these as active ..
  stats_collector_->StartStats(&connection_begin_stats_,
                               &connection_end_stats_);
  stats_collector_->StartStats(&stream_begin_stats_,
                               &stream_end_stats_);

  OpenMediaStats(media_request->serving_info().media_name_);
  media_name_ = media_request->serving_info().media_name_;
  if ( auth_helper_ != NULL &&
       media_request_->auth_reply().reauthorize_interval_ms_ > 0 ) {
    reauthorize_callback_ =
        NewPermanentCallback(
            this, &StreamRequest::Reauthorize);
    selector_->RegisterAlarm(
        reauthorize_callback_,
        media_request_->auth_reply().reauthorize_interval_ms_);
  }
  if ( media_request_->auth_reply().time_limit_ms_ > 0 ) {
    terminate_callback_ =
        NewPermanentCallback(this, &StreamRequest::TerminateRequest);
    selector_->RegisterAlarm(
        terminate_callback_,
        media_request_->auth_reply().time_limit_ms_);
  }

}

void StreamRequest::ReauthorizeCallback() {
  CHECK(auth_helper_ != NULL);

  delete unpause_buffer_callback_;
  unpause_buffer_callback_ = NULL;

  *(media_request_->mutable_auth_reply()) = auth_helper_->reply();
  if ( !auth_helper_->reply().allowed_ ) {
    LOG_INFO << " Reauthorization failed for "
             << media_request_->ToString();
    ClearRequest(true);
  } else {
    if ( auth_helper_->reply().reauthorize_interval_ms_ > 0 ) {
      if ( reauthorize_callback_ == NULL ) {
        reauthorize_callback_ =
            NewPermanentCallback(
                this, &StreamRequest::Reauthorize);

      }
      selector_->RegisterAlarm(reauthorize_callback_,
                               auth_helper_->reply().reauthorize_interval_ms_);
    }
    if ( auth_helper_->reply().time_limit_ms_ > 0 ) {
      if ( terminate_callback_ == NULL ) {
        terminate_callback_ =
            NewPermanentCallback(this, &StreamRequest::TerminateRequest);
      }
      selector_->ReregisterAlarm(terminate_callback_,
                                auth_helper_->reply().time_limit_ms_);
    } else  if ( terminate_callback_ != NULL ) {
      selector_->UnregisterAlarm(terminate_callback_);
    }
  }
}

void StreamRequest::Reauthorize() {
  DCHECK(auth_helper_ != NULL);
  auth_helper_->mutable_req()->action_performed_ms_ =
      (selector_->now() - start_time_);
  if ( auth_helper_->is_started() ) {
    return;
  }
  auth_helper_->Start(NewCallback(this, &StreamRequest::ReauthorizeCallback));
}

void StreamRequest::TerminateRequest() {
  LOG_INFO << " Authorization time ended for "
           << media_request_->ToString();
  ClearRequest(false);
}

StreamRequest::~StreamRequest() {
  if ( reauthorize_callback_ != NULL ) {
    selector_->UnregisterAlarm(reauthorize_callback_);
    delete reauthorize_callback_;
  }
  if ( terminate_callback_ != NULL ) {
    selector_->UnregisterAlarm(terminate_callback_);
    delete terminate_callback_;
  }
  if ( auth_helper_ != NULL ) {
    if ( auth_helper_->is_started() ) {
      auth_helper_->Cancel();
    } else {
      delete auth_helper_;
    }
  }
  // end media stats
  CloseMediaStats();

  const int64 now = timer::Date::Now();
  // end stream stats
  stream_end_stats_.timestamp_utc_ms_ = now;
  if ( stream_end_stats_.result_.Get().result_.Get().StdStr() == "" ) {
    stream_end_stats_.result_ = StreamResult("END");
  }
  stats_collector_->EndStats(&stream_end_stats_);

  // end connection stats
  connection_end_stats_.timestamp_utc_ms_ = now;
  connection_end_stats_.result_ = ConnectionResult("END");
  stats_collector_->EndStats(&connection_end_stats_);

  // TODO[cpopescu] : record stats..
  delete serializer_;
  if ( media_request_ != NULL ) {
    LOG_INFO << " Deleting Request for: " << media_request_->ToString();
    mapper_->RemoveRequest(media_request_, processing_callback_);
    media_request_ = NULL;
  }
  delete processing_callback_;
  processing_callback_ = NULL;
}

bool StreamRequest::StartRequest() {
  const char* media_name = media_request_->serving_info().media_name_.c_str();
  *media_request_->mutable_caps() = mapper_->HasMedia(media_name);
  switch ( media_request_->caps().tag_type_ ) {
    case streaming::TYPE_MP3:
      serializer_ = new streaming::Mp3TagSerializer();
      break;
    case streaming::TYPE_FLV:
      serializer_ = new streaming::FlvTagSerializer();
      break;
    case streaming::TYPE_AAC:
      serializer_ = new streaming::AacTagSerializer();
      break;
    case streaming::TYPE_INTERNAL_MEDIA:
      serializer_ = new streaming::InternalTagSerializer();
      break;
    default:
      serializer_ = new streaming::RawTagSerializer();
      break;

      LOG_ERROR << " Invalid serializer for element: " << media_name
                << " Tag type: "
                << TagTypeName(media_request_->caps().tag_type_);
      serializer_ = NULL;
  }
  if ( serializer_ ) {
    // Bootstrap the request
    if ( mapper_->AddRequest(media_name,
                             media_request_,
                             processing_callback_) ) {
      tag_type_ = media_request_->caps().tag_type_;
      return true;
    }
    tag_type_ = streaming::INVALID_STREAM_TYPE;
  }
  delete media_request_;
  media_request_ = NULL;
  return false;
}

void StreamRequest::ClearRequest(bool from_server) {
  LOG_INFO << " Clearing request: " << from_server;
  CHECK(http_request_ != NULL);
  http::ServerRequest* req = http_request_;
  http_request_ = NULL;

  stream_end_stats_.result_ = StreamResult(
      ResultName(req->request()->server_header()->status_code()));
  LOG_INFO << "Client ended: "
           << "is_orphaned:" << req->is_orphaned();
  // << " stats: " << stats_.ToString();

  if ( !from_server ) {
    if ( is_first_tag_ ) {
      req->Reply();  // basically nothing was sent ..
      selector_->DeleteInSelectLoop(this);
    } else {
      serializer_->Finalize(req->request()->server_data());
      req->EndStreamingData();
    }
  }
}
void StreamRequest::EndOfStream() {
  if ( http_request_ != NULL ) {
    // If was closed before any returned tag
    ClearRequest(true);
  }
  selector_->DeleteInSelectLoop(this);
}

void StreamRequest::ProcessTag(const streaming::Tag* tag,
                               streaming::ElementController* controller) {
  if ( tag->tag_type()  == streaming::TYPE_STREAM_EOS
       || http_request_ == NULL
       || http_request_->is_orphaned() ) {
    if ( media_request_ != NULL ) {
      mapper_->RemoveRequest(media_request_, processing_callback_);
      media_request_ = NULL;
    }
    if ( http_request_ != NULL ) {
      ClearRequest(false);
    }
    return;
  }
  if ( tag->tag_type() == streaming::TYPE_SOURCE_CHANGED ) {
    CloseMediaStats();
    OpenMediaStats(tag->data()->ToString());
    media_name_ = tag->data()->ToString();
    element_seq_id_++;
  }
  if ( tag->tag_type() == streaming::TYPE_SOURCE_CHANGED ||
       tag->tag_type() == streaming::TYPE_SEEK_PERFORMED ||
       tag->tag_type() == streaming::TYPE_SOURCE_RESTARTED ) {
    is_first_tag_ = true;
  }
  // First tag to send - set content type and prepare the header..
  const int64 now = timer::TicksMsec();
  if ( is_first_tag_ ) {
    first_tag_time_ = now;
    is_first_tag_ = false;
  }
  if ( !http_request_->is_server_streaming() ) {
    if ( media_request_->serving_info().enable_buffer_flow_control_ &&
         streaming::IsRawTagType(media_request_->caps().tag_type_) &&
         media_request_->serving_info().size_ > 0 ) {
      http_request_->request()->server_header()->AddField(
          http::kHeaderContentLength,
          strutil::StringPrintf("%lld",
                                media_request_->serving_info().size_ -
                                media_request_->serving_info().offset_),
          true);
    }
    http_request_->BeginStreamingData(
        http::OK, NULL,
        NewCallback(this, &StreamRequest::EndOfStream));
    serializer_->Initialize(http_request_->request()->server_data());
  }

  // We may pause if we run over buffer size
  if ( media_request_->serving_info().enable_buffer_flow_control_
       && ( http_request_->free_output_bytes() - tag->size() <
            (FLAGS_http_connection_max_media_outbuf_size >> 2) )
       && unpause_buffer_callback_ == NULL
       && controller != NULL
       && controller->SupportsPause() ) {
    controller->Pause(true);
    unpause_buffer_callback_ = NewCallback(
        this, &StreamRequest::UnpauseElementBuffer,
        element_seq_id_, controller);
  }

  bool drop_tag = false;
  // Update the stats..
  if ( tag->is_stream_tag() ) {
    if ( tag->is_audio_tag() ) {
      // Possible resync and re-start syncing
      // TODO: differentiate video / audio
      if ( is_audio_dropping_
           && http_request_->free_output_bytes() > tag->size() &&
           tag->can_resync() ) {
        is_audio_dropping_ = false;
      } else if ( http_request_->free_output_bytes() < tag->size() &&
                  tag->is_droppable() ) {
        is_audio_dropping_ = true;
      }

      drop_tag |= is_audio_dropping_;
      media_end_stats_.audio_frames_ = media_end_stats_.audio_frames_.Get() + 1;
      if ( is_audio_dropping_ ) {
        media_end_stats_.audio_frames_dropped_ =
            media_end_stats_.audio_frames_dropped_.Get() + 1;
      }
    }
    if ( tag->is_video_tag() ) {
      // Possible resync and re-start sending for video
      // TODO: differentiate video / audio
      if ( is_video_dropping_
           && http_request_->free_output_bytes() > tag->size() &&
           tag->can_resync() ) {
        is_video_dropping_ = false;
      } else if ( http_request_->free_output_bytes() < tag->size() &&
                  tag->is_droppable() ) {
        is_video_dropping_ = true;
      }
      drop_tag |= is_video_dropping_;
      media_end_stats_.video_frames_ =
          media_end_stats_.video_frames_.Get() + 1;
      if ( is_video_dropping_ ) {
        media_end_stats_.video_frames_dropped_ =
            media_end_stats_.video_frames_dropped_.Get() + 1;
      }
    }
  }
  // Send the data actually..
  if ( !drop_tag ) {
    const int32 initial_size = http_request_->request()->server_data()->Size();
    last_tag_stream_time_ = tag->time_offset_ms();
    const bool is_flv = tag->tag_type() == streaming::TYPE_FLV;
    if ( is_flv ) {
      const streaming::FlvTag* flv_tag =
          static_cast<const streaming::FlvTag*>(tag->data());
      if ( flv_tag->data_type() == streaming::FLV_FRAMETYPE_METADATA ) {
        SerializeFlvMetadataTag(tag, flv_tag);
      } else {
        serializer_->Serialize(tag, http_request_->request()->server_data());
      }
    } else {
      serializer_->Serialize(tag, http_request_->request()->server_data());
    }
    const int32 final_size = http_request_->request()->server_data()->Size();
    connection_end_stats_.bytes_up_ = connection_end_stats_.bytes_up_.Get() +
                                      final_size - initial_size;
    media_end_stats_.bytes_up_ = media_end_stats_.bytes_up_.Get() +
                                 final_size - initial_size;
  }

  // TODO: We may choose to do network flow control here..
  http_request_->ContinueStreamingData(unpause_buffer_callback_);
}

void StreamRequest::UnpauseElementBuffer(
    int64 element_seq_id,
    streaming::ElementController* controller) {
  DCHECK(unpause_buffer_callback_ != NULL);
  DCHECK(controller->SupportsPause());
  unpause_buffer_callback_ = NULL;
  if ( element_seq_id != element_seq_id_ )
    return;

  controller->Pause(false);
}

void StreamRequest::SerializeFlvMetadataTag(
    const streaming::Tag* tag,
    const streaming::FlvTag* flv_tag) {
  rtmp::CString name;
  rtmp::CMixedMap values;
  io::MemoryStream tag_content(common::BIGENDIAN);
  tag_content.AppendStreamNonDestructive(flv_tag->data());
  if ( (name.ReadFromMemoryStream(&tag_content,
                                  rtmp::AmfUtil::AMF0_VERSION) !=
        rtmp::AmfUtil::READ_OK)
       || (values.ReadFromMemoryStream(&tag_content,
                                       rtmp::AmfUtil::AMF0_VERSION) !=
           rtmp::AmfUtil::READ_OK) ) {
    serializer_->Serialize(tag, http_request_->request()->server_data());
    return;
  }
  if ( name.value() == streaming::kOnMetaData ) {
    if ( values.data().find("wc_mn") == values.data().end() ) {
      values.mutable_data()["wc_mn"] =
          new rtmp::CString(media_name_);
    }
    const rtmp::CMixedMap::Map::iterator
        it_cue = values.mutable_data().find("cuePoints");
    if ( it_cue != values.mutable_data().end() ) {
      delete it_cue->second;
      values.mutable_data().erase(it_cue);
    }
  } else if ( name.value() == streaming::kOnCuePoint ) {
    if ( values.data().find("wc_mn") == values.data().end() ) {
      values.mutable_data()["wc_mn"] = new rtmp::CString(media_name_);
    }
  } else {
    serializer_->Serialize(tag, http_request_->request()->server_data());
    return;
  }
  streaming::FlvTag* const copy_flv_tag =
      reinterpret_cast<streaming::FlvTag*>(flv_tag->CopyClone());
  copy_flv_tag->mutable_data()->Clear();
  name.WriteToMemoryStream(copy_flv_tag->mutable_data(),
                           rtmp::AmfUtil::AMF0_VERSION);
  values.WriteToMemoryStream(copy_flv_tag->mutable_data(),
                             rtmp::AmfUtil::AMF0_VERSION);
  streaming::Tag* copy_tag = new streaming::Tag(
      copy_flv_tag, copy_flv_tag->data()->Size(),
      tag->tag_type(),
      tag->data_attributes(),
      tag->flavour_mask(),
      tag->time_offset_ms());
  serializer_->Serialize(copy_tag, http_request_->request()->server_data());
  copy_tag->Release();

}
void StreamRequest::OpenMediaStats(const string& content_id) {
  media_begin_stats_.media_id_ = strutil::StringPrintf("%s.%lld",
          stream_begin_stats_.stream_id_.Get().CStr(),
          static_cast<long long int>(media_id_));
  ++media_id_;

  const int64 now = timer::Date::Now();
  media_begin_stats_.timestamp_utc_ms_ = now;
  media_begin_stats_.stream_ = stream_begin_stats_;
  media_begin_stats_.content_id_ = content_id;
  const int64 time_delta = selector_->now() - start_time_;
  media_begin_stats_.stream_time_ms_ = time_delta;
  media_begin_stats_.content_time_ms_ = 0;

  media_end_stats_.media_id_ = media_begin_stats_.media_id_;
  media_end_stats_.timestamp_utc_ms_ = now;
  media_end_stats_.bytes_up_ = 0;
  media_end_stats_.bytes_down_ = 0;
  media_end_stats_.video_frames_ = 0;
  media_end_stats_.video_frames_dropped_ = 0;
  media_end_stats_.audio_frames_ = 0;
  media_end_stats_.audio_frames_dropped_ = 0;
  media_end_stats_.result_ = MediaResult("Active");

  LOG_INFO << "Seding media stats: " << media_begin_stats_;
  stats_collector_->StartStats(&media_begin_stats_, &media_end_stats_);
}

void StreamRequest::CloseMediaStats() {
  const int64 now = timer::Date::Now();
  media_end_stats_.timestamp_utc_ms_ = now;
  // TODO(cosmin): duration_in_ms_ should be content time! not physical time
  const int64 time_delta = now -
                           media_begin_stats_.stream_time_ms_.Get();
  media_end_stats_.duration_in_ms_ = time_delta;
  media_end_stats_.result_ = MediaResult("END");
  LOG_INFO << "Ending media stats: " << media_end_stats_;
  stats_collector_->EndStats(&media_end_stats_);
}