PeerStreamer

/*

  Copyright 2003 John Knottenbelt

  This program is free software; you can redistribute it and/or modify

  it under the terms of the GNU General Public License as published by

  the Free Software Foundation; either version 2 of the License, or

  (at your option) any later version.

  This program is distributed in the hope that it will be useful,

  but WITHOUT ANY WARRANTY; without even the implied warranty of

  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License

  along with this program; if not, write to the Free Software

  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111, USA.

*/

#include "config.h"

#include "connection.h"

#include "demuxer.h"

#include "utils.h"

#include "stringutil.h"

#include "inputbuffer.h"

#include "outputbuffer.h"

#include "scheduler.h"

#include <iostream>

#include <cstring>

#include <vector>

#include <unistd.h>

#include <cerrno>

#include <ctime>

#include <algorithm>

#include <cassert>

Connection::Connection(TunerManager *tm, int dataFD)

  : dataFD(dataFD), priority(0), 

    tuned(false), retuning(false), 

    tm(tm), removeMe(false),

    stopTime(0),

    interrupted(false), 

    lostPacket(false),

    convertToPS(false),

    lastTime(0)

{

  if (debugLevel > 1)

    std::cerr << "New client connection" << std::endl;

  dataOutput = new OutputBuffer(dataFD, 16384, 1048576);

}

Connection::~Connection()

{

  if (debugLevel > 1)

    std::cerr << "Client connection closed" << std::endl;

  retuning=true;

  doUnsubscribe();

  delete dataOutput;

  close(dataFD);

}

void Connection::doUnsubscribe()

{

  SourceList copy = demuxers;

  for (SourceList::iterator i = copy.begin(); i != copy.end(); i++)

    (*i)->unsubscribe(this);

}

bool Connection::doSubscribe()

{

  bool result = true;

  for (SourceList::iterator i = demuxers.begin(); i != demuxers.end(); i++) 

    result = (*i)->subscribe(this) && result;

  if (!result) {

    doUnsubscribe();

    return false;

  }

  return true;

}

void Connection::notifyUnsubscribe(Source *s)

{

  Demuxer *d = (Demuxer *) s;

  demuxers.remove(d);

  interrupted = true;

}

time_t Connection::getCurrentTime() const

{

  return time(NULL);

}

void Connection::addSelectFDs(Select &s) const

{

  dataOutput->addSelectFDs(s);

  if (stopTime != 0)

    s.addAlarm(std::max(0L, stopTime - getCurrentTime()), 0);

}

bool Connection::isReady(const Select &s) const

{

  if (removeMe)

    return true;

  return dataOutput->isReady(s);

}

int Connection::getPriority() const

{

  return priority;

}

void Connection::setPriority(int p)

{

  priority = p;

}

void Connection::setStopTime(time_t newStopTime)

{

  stopTime = newStopTime;

}

time_t Connection::getStopTime() const

{

  return stopTime;

}

bool Connection::isValidType(const std::string &type ) const

{

  return tm->validType(type);

}

bool Connection::isValidChannel(const std::string &type, const std::string &channel) const

{

  return tm->validChannel(type, channel);

}

bool Connection::canTune(const std::string &newType, const std::string &newChannel, int p) const

{

  SavePriority save(const_cast<Connection *>(this), p - 1);

  return tm->canGetDemuxers(newType, newChannel, p);

}

void Connection::getTunableChannels(const std::string &type, int priority, StringList &channels) const

{

  tm->getTunableChannels(type, priority, channels);

}

bool Connection::isRetuning() const {

  return retuning;

}

bool Connection::numDemuxers() const

{

  return demuxers.empty();

}

bool Connection::getInterrupted() const {

  return interrupted;

}

void Connection::setRemoveMe(bool status)

{

  removeMe = status;

}

bool Connection::tune(const std::string &newType, const std::string &newChannel, bool convertToPS, int p)

{

  if (tuned) {

    if (type == newType && channel == newChannel) {

      setPriority(p);

      return true;

    }

  }

  if (debugLevel > 1) {

    std::cerr << "Tuning to " << newType << " " << newChannel 

              << " priority " << p << std::endl;

  }

  // The retuning variable prevents the client from being notified

  // that he has lost a subscription. The loss of subscription is to

  // be expected since the request is to change the channel!

  retuning = true;

  SourceList newDemuxers;

  // Set the priority of the current channel

  // to be less than the specified priority

  {

    SavePriority save(this, p - 1);

    // Try to allocate the new demuxers

    tm->getDemuxers(newType, newChannel, convertToPS, newDemuxers, p);

    retuning = false;

    if (newDemuxers.empty()) 

      return false;      // We remain subscribed to the old channel.

  }

  // Unsubscribe (if necessary) from the old and subscribe to the new

  doUnsubscribe();

  demuxers = newDemuxers;

  type = newType;

  channel = newChannel;

  bool result = doSubscribe();

  setPriority(p);

  tuned = true;

  interrupted = false;

  lostPacket = false;

  this->convertToPS = convertToPS;

  return result;

}

bool Connection::processReady(const Select &s)

{

  if (stopTime != 0 && getCurrentTime() > stopTime) {

    if (debugLevel > 0) {

      std::cerr << "Stop time reached (stopTime = " << stopTime 

                << ", now = " << getCurrentTime() << ")\n";

    }

    return false;

  }

  if (removeMe)

    return false;

  if (dataOutput->isReady(s)) {

    if (!dataOutput->process()) 

      return false;

  }

  return true;

}

void Connection::receiveData(unsigned char *data, unsigned size)

{

  if (debugLevel > 0) {

    time_t now = time(NULL);

    if (now - lastTime >= 2) {

      std::cerr << "Buffer usage:" << dataOutput->getUsed() << "/" << dataOutput->getMaxSize() 

                << std::endl;

      lastTime = now;

    }

  }

  // Receive data from the Demuxers and forward it to the clients

  if (!dataOutput->write((char *) data, size)) {

    if (!lostPacket) {

      std::cerr << "Warning: client not reading data connection fast enough (dropping packets)" << std::endl;

      lostPacket = true;

    }

    if (!convertToPS) {

      // Drop a TS packet from the beginning

      dataOutput->discard(188 + dataOutput->getUsed() % 188);

      dataOutput->write((char *) data, size);

    }

  }

}

std::ostream& Connection::printOn(std::ostream &o) const

{

  if (!tuned) 

    return o << "UNTUNED";

  else

    return o << priority << " " << type << " " << escapeWS(channel);

}