libbase/RTMP.cpp

   1 //
   2 //   Copyright (C) 2007, 2008, 2009, 2010 Free Software Foundation, Inc.
   3 //
   4 // This program is free software; you can redistribute it and/or modify
   5 // it under the terms of the GNU General Public License as published by
   6 // the Free Software Foundation; either version 3 of the License, or
   7 // (at your option) any later version.
   8 //
   9 // This program is distributed in the hope that it will be useful,
  10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 // GNU General Public License for more details.
  13 //
  14 // You should have received a copy of the GNU General Public License
  15 // along with this program; if not, write to the Free Software
  16 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  17
  18 #include <cstdlib>
  19 #include <cstring>
  20 #include <cassert>
  21 #include <cstdio>
  22 #include <boost/lexical_cast.hpp>
  23
  24 #include "GnashSystemNetHeaders.h"
  25
  26 // Replace!!
  27 #ifndef _WIN32
  28 # include <sys/times.h>
  29 #else
  30 // TODO: use uptime properly on win32.
  31 # include <ctime>
  32 #endif
  33
  34 #include "RTMP.h"
  35 #include "log.h"
  36 #include "AMF.h"
  37 #include "GnashAlgorithm.h"
  38 #include "URL.h"
  39 #include "ClockTime.h"
  40
  41 namespace gnash {
  42 namespace rtmp {
  43
  44 namespace {
  45
  46     bool sendBytesReceived(RTMP* r);
  47
  48     void handleMetadata(RTMP& r, const boost::uint8_t *payload,
  49             unsigned int len);
  50     void handleChangeChunkSize(RTMP& r, const RTMPPacket& packet);
  51     void handleControl(RTMP& r, const RTMPPacket& packet);
  52     void handleServerBW(RTMP& r, const RTMPPacket& packet);
  53     void handleClientBW(RTMP& r, const RTMPPacket& packet);
  54
  55     void setupInvokePacket(RTMPPacket& packet);
  56     boost::uint32_t getUptime();
  57
  58     boost::int32_t decodeInt32LE(const boost::uint8_t* c);
  59     int encodeInt32LE(boost::uint8_t *output, int nVal);
  60     unsigned int decodeInt24(const boost::uint8_t* c);
  61     boost::uint8_t* encodeInt16(boost::uint8_t *output, boost::uint8_t *outend,
  62             short nVal);
  63     boost::uint8_t* encodeInt24(boost::uint8_t *output, boost::uint8_t *outend,
  64             int nVal);
  65     boost::uint8_t* encodeInt32(boost::uint8_t *output, boost::uint8_t *outend,
  66             int nVal);
  67
  68     static const int packetSize[] = { 12, 8, 4, 1 };
  69
  70 }
  71
  72 namespace {
  73
  74 /// A random generator for generating the signature.
  75 //
  76 /// TODO: do this properly (it's currently not very random).
  77 struct RandomByte
  78 {
  79     bool operator()() const {
  80         return std::rand() % 256;
  81     }
  82 };
  83
  84 }
  85
  86 /// A utility functor for carrying out the handshake.
  87 class HandShaker
  88 {
  89 public:
  90
  91     static const int sigSize = 1536;
  92
  93     HandShaker(Socket& s);
  94
  95     /// Calls the next stage in the handshake process.
  96     void call();
  97
  98     bool success() const {
  99         return _complete;
 100     }
 101
 102     bool error() const {
 103         return _error || _socket.bad();
 104     }
 105
 106 private:
 107
 108     /// These are the stages of the handshake.
 109     //
 110     /// If the socket is not ready, they will return false. If the socket
 111     /// is in error, they will set _error.
 112     bool stage0();
 113     bool stage1();
 114     bool stage2();
 115     bool stage3();
 116
 117     Socket _socket;
 118     std::vector<boost::uint8_t> _sendBuf;
 119     std::vector<boost::uint8_t> _recvBuf;
 120     bool _error;
 121     bool _complete;
 122     size_t _stage;
 123 };
 124
 125 RTMPPacket::RTMPPacket(size_t reserve)
 126     :
 127     header(),
 128     buffer(new SimpleBuffer(reserve + RTMPHeader::headerSize)),
 129     bytesRead(0)
 130 {
 131     // This is space for the header be filled in later.
 132     buffer->resize(RTMPHeader::headerSize);
 133 }
 134
 135 RTMPPacket::RTMPPacket(const RTMPPacket& other)
 136     :
 137     header(other.header),
 138     buffer(other.buffer)
 139 {}
 140
 141 const size_t RTMPHeader::headerSize;
 142
 143 RTMP::RTMP()
 144     :
 145     _inChunkSize(RTMP_DEFAULT_CHUNKSIZE),
 146     m_mediaChannel(0),
 147     m_nClientBW2(2),
 148     _bytesIn(0),
 149     _bytesInSent(0),
 150     _serverBandwidth(2500000),
 151     _bandwidth(2500000),
 152     _outChunkSize(RTMP_DEFAULT_CHUNKSIZE),
 153     _connected(false),
 154     _error(false)
 155 {
 156 }
 157
 158 RTMP::~RTMP()
 159 {
 160 }
 161
 162 bool
 163 RTMP::hasPacket(ChannelType t, size_t channel) const
 164 {
 165     const ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 166     return set.find(channel) != set.end();
 167 }
 168
 169 RTMPPacket&
 170 RTMP::getPacket(ChannelType t, size_t channel)
 171 {
 172     ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 173     return set[channel];
 174 }
 175
 176 RTMPPacket&
 177 RTMP::storePacket(ChannelType t, size_t channel, const RTMPPacket& p)
 178 {
 179     ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 180     RTMPPacket& stored = set[channel];
 181     stored = p;
 182     return stored;
 183 }
 184
 185 void
 186 RTMP::setBufferTime(size_t size, int streamID)
 187 {
 188     sendCtrl(*this, CONTROL_BUFFER_TIME, streamID, size);
 189 }
 190
 191 void
 192 RTMP::call(const SimpleBuffer& amf)
 193 {
 194     RTMPPacket p(amf.size());
 195     setupInvokePacket(p);
 196
 197     // Copy the data.
 198     p.buffer->append(amf.data(), amf.size());
 199     sendPacket(p);
 200 }
 201
 202 bool
 203 RTMP::connect(const URL& url)
 204 {
 205     log_debug("Connecting to %s", url.str());
 206
 207     const std::string& hostname = url.hostname();
 208     const std::string& p = url.port();
 209
 210     // Default port.
 211     boost::uint16_t port = 1935;
 212     if (!p.empty()) {
 213         try {
 214             port = boost::lexical_cast<boost::uint16_t>(p);
 215         }
 216         catch (const boost::bad_lexical_cast&) {}
 217     }
 218
 219     // Basic connection attempt.
 220     if (!_socket.connect(hostname, port)) {
 221         log_error("Initial connection failed");
 222         return false;
 223     }
 224
 225     _handShaker.reset(new HandShaker(_socket));
 226
 227     // Start handshake attempt immediately.
 228     _handShaker->call();
 229
 230     return true;
 231 }
 232
 233 void
 234 RTMP::update()
 235 {
 236     if (!connected()) {
 237         _handShaker->call();
 238         if (_handShaker->error()) {
 239             _error = true;
 240         }
 241         if (!_handShaker->success()) return;
 242         _connected = true;
 243     }
 244
 245     const size_t reads = 10;
 246
 247     for (size_t i = 0; i < reads; ++i) {
 248
 249         /// No need to continue reading (though it should do no harm).
 250         if (error()) return;
 251
 252         RTMPPacket p;
 253
 254         // If we haven't finished reading a packet, retrieve it; otherwise
 255         // use an empty one.
 256         if (_incompletePacket.get()) {
 257             log_debug("Doing incomplete packet");
 258             p = *_incompletePacket;
 259             _incompletePacket.reset();
 260         }
 261         else {
 262             if (!readPacketHeader(p)) continue;
 263         }
 264
 265         // Get the payload if possible.
 266         if (hasPayload(p) && !readPacketPayload(p)) {
 267             // If the payload is not completely readable, store it and
 268             // continue.
 269             _incompletePacket.reset(new RTMPPacket(p));
 270             continue;
 271         }
 272
 273         // Store a copy of the packet for later additions and as a reference for
 274         // future sends.
 275         RTMPPacket& stored = storePacket(CHANNELS_IN, p.header.channel, p);
 276
 277         // If the packet is complete, the stored packet no longer needs to
 278         // keep the data alive.
 279         if (isReady(p)) {
 280             clearPayload(stored);
 281             handlePacket(p);
 282             return;
 283         }
 284     }
 285 }
 286
 287 void
 288 RTMP::handlePacket(const RTMPPacket& packet)
 289 {
 290     const PacketType t = packet.header.packetType;
 291
 292     log_debug("Received %s", t);
 293
 294     switch (t) {
 295
 296         case PACKET_TYPE_CHUNK_SIZE:
 297             handleChangeChunkSize(*this, packet);
 298             break;
 299
 300         case PACKET_TYPE_BYTES_READ:
 301             break;
 302
 303         case PACKET_TYPE_CONTROL:
 304             handleControl(*this, packet);
 305             break;
 306
 307         case PACKET_TYPE_SERVERBW:
 308             handleServerBW(*this, packet);
 309             break;
 310
 311         case PACKET_TYPE_CLIENTBW:
 312             handleClientBW(*this, packet);
 313             break;
 314
 315         case PACKET_TYPE_AUDIO:
 316             if (!m_mediaChannel) m_mediaChannel = packet.header.channel;
 317             break;
 318
 319         case PACKET_TYPE_VIDEO:
 320             if (!m_mediaChannel) m_mediaChannel = packet.header.channel;
 321             break;
 322
 323         case PACKET_TYPE_FLEX_STREAM_SEND:
 324             LOG_ONCE(log_unimpl("unsupported packet %s received"));
 325             break;
 326
 327         case PACKET_TYPE_FLEX_SHARED_OBJECT:
 328             LOG_ONCE(log_unimpl("unsupported packet %s received"));
 329             break;
 330
 331         case PACKET_TYPE_FLEX_MESSAGE:
 332         {
 333             LOG_ONCE(log_unimpl("partially supported packet %s received"));
 334             _messageQueue.push_back(packet.buffer);
 335             break;
 336         }
 337
 338         case PACKET_TYPE_METADATA:
 339             handleMetadata(*this, payloadData(packet), payloadSize(packet));
 340             break;
 341
 342         case PACKET_TYPE_SHARED_OBJECT:
 343             LOG_ONCE(log_unimpl("packet %s received"));
 344             break;
 345
 346         case PACKET_TYPE_INVOKE:
 347             _messageQueue.push_back(packet.buffer);
 348             break;
 349
 350         case PACKET_TYPE_FLV:
 351             _flvQueue.push_back(packet.buffer);
 352             break;
 353
 354         default:
 355             log_error("Unknown packet %s received", t);
 356
 357     }
 358
 359 }
 360
 361 int
 362 RTMP::readSocket(boost::uint8_t* buffer, int n)
 363 {
 364
 365     assert(n >= 0);
 366
 367     const std::streamsize bytesRead = _socket.read(buffer, n);
 368
 369     if (_socket.bad() || _socket.eof() || !_socket.connected()) {
 370         _error = true;
 371         return 0;
 372     }
 373
 374     if (!bytesRead) return 0;
 375
 376     _bytesIn += bytesRead;
 377
 378     // Report bytes recieved every time we reach half the bandwidth.
 379     // Doesn't seem very likely to be the way the pp does it.
 380     if (_bytesIn > _bytesInSent + _bandwidth / 2) {
 381         sendBytesReceived(this);
 382         log_debug("Sent bytes received");
 383     }
 384
 385     buffer += bytesRead;
 386     return bytesRead;
 387 }
 388
 389 void
 390 RTMP::play(const SimpleBuffer& buf, int streamID)
 391 {
 392     RTMPPacket packet(buf.size());
 393
 394     packet.header.channel = CHANNEL_VIDEO;
 395     packet.header.packetType = PACKET_TYPE_INVOKE;
 396
 397     packet.header._streamID = streamID;
 398
 399     packet.buffer->append(buf.data(), buf.size());
 400     sendPacket(packet);
 401 }
 402
 403 /// Send the server bandwidth.
 404 //
 405 /// Why would we want to send this?
 406 bool
 407 sendServerBW(RTMP& r)
 408 {
 409     RTMPPacket packet(4);
 410
 411     packet.header.channel = CHANNEL_CONTROL1;
 412     packet.header.packetType = PACKET_TYPE_SERVERBW;
 413
 414     SimpleBuffer& buf = *packet.buffer;
 415
 416     buf.appendNetworkLong(r.serverBandwidth());
 417     return r.sendPacket(packet);
 418 }
 419
 420
 421 /// Fills a pre-existent RTMPPacket with information.
 422 //
 423 /// This is either read entirely from incoming data, or copied from a
 424 /// previous packet in the same channel. This happens when the header type
 425 /// is less than RTMP_PACKET_SIZE_LARGE.
 426 //
 427 /// It seems as if new packets can add to the data of old ones if they have
 428 /// a minimal, small header.
 429 bool
 430 RTMP::readPacketHeader(RTMPPacket& packet)
 431 {
 432
 433     RTMPHeader& hr = packet.header;
 434
 435     boost::uint8_t hbuf[RTMPHeader::headerSize] = { 0 };
 436     boost::uint8_t* header = hbuf;
 437
 438     // The first read may fail, but otherwise we expect a complete header.
 439     if (readSocket(hbuf, 1) == 0) {
 440         return false;
 441     }
 442
 443     //log_debug("Packet is %s", boost::io::group(std::hex, (unsigned)hbuf[0]));
 444
 445     const int htype = ((hbuf[0] & 0xc0) >> 6);
 446     //log_debug("Thingy whatsit (packet size type): %s", htype);
 447
 448     const int channel = (hbuf[0] & 0x3f);
 449     //log_debug("Channel: %s", channel);
 450
 451     hr.headerType = static_cast<PacketSize>(htype);
 452     hr.channel = channel;
 453     ++header;
 454
 455     if (hr.channel == 0) {
 456         if (readSocket(&hbuf[1], 1) != 1) {
 457           log_error("failed to read RTMP packet header 2nd byte");
 458           return false;
 459         }
 460         hr.channel = hbuf[1] + 64;
 461         ++header;
 462     }
 463     else if (hr.channel == 1) {
 464         if (readSocket(&hbuf[1], 2) != 2) {
 465             log_error("Failed to read RTMP packet header 3nd byte");
 466              return false;
 467         }
 468
 469         const boost::uint32_t tmp = (hbuf[2] << 8) + hbuf[1];
 470         hr.channel = tmp + 64;
 471         log_debug( "%s, channel: %0x", __FUNCTION__, hr.channel);
 472         header += 2;
 473     }
 474
 475     // This is the size in bytes of the packet header according to the
 476     // type.
 477     int nSize = packetSize[htype];
 478
 479     /// If we didn't receive a large header, the timestamp is relative
 480     if (htype != RTMP_PACKET_SIZE_LARGE) {
 481
 482         if (!hasPacket(CHANNELS_IN, hr.channel)) {
 483             log_error("Incomplete packet received on channel %s", channel);
 484             return false;
 485         }
 486
 487         // For all other header types, copy values from the last message of
 488         // this channel. This includes any payload data from incomplete
 489         // messages.
 490         packet = getPacket(CHANNELS_IN, hr.channel);
 491     }
 492
 493     --nSize;
 494
 495     if (nSize > 0 && readSocket(header, nSize) != nSize) {
 496         log_error( "Failed to read RTMP packet header. type: %s",
 497                 static_cast<unsigned>(hbuf[0]));
 498         return false;
 499     }
 500
 501     // nSize is predicted size - 1. Add what we've read already.
 502     int hSize = nSize + (header - hbuf);
 503
 504     if (nSize >= 3) {
 505
 506         const boost::uint32_t timestamp = decodeInt24(header);
 507
 508         // Make our packet timestamp absolute. If the value is 0xffffff,
 509         // the absolute value comes later.
 510         if (timestamp != 0xffffff) {
 511             if (htype != RTMP_PACKET_SIZE_LARGE) {
 512                 packet.header._timestamp += timestamp;
 513             }
 514             else {
 515                 packet.header._timestamp = timestamp;
 516             }
 517         }
 518
 519         // Have at least a different size payload from the last packet.
 520         if (nSize >= 6) {
 521
 522             // We do this in case there was an incomplete packet in the
 523             // channel already.
 524             clearPayload(packet);
 525             hr.dataSize = decodeInt24(header + 3);
 526
 527             // More than six: read packet type
 528             if (nSize > 6) {
 529                 hr.packetType = static_cast<PacketType>(header[6]);
 530
 531                 // Large packets have a streamID.
 532                 if (nSize == 11) {
 533                     hr._streamID = decodeInt32LE(header + 7);
 534                 }
 535             }
 536         }
 537     }
 538
 539     if (hr._timestamp == 0xffffff) {
 540       if (readSocket(header+nSize, 4) != 4) {
 541               log_error( "%s, failed to read extended timestamp",
 542               __FUNCTION__);
 543               return false;
 544             }
 545           hr._timestamp = amf::readNetworkLong(header+nSize);
 546           hSize += 4;
 547
 548     }
 549
 550     const size_t bufSize = hr.dataSize + RTMPHeader::headerSize;
 551
 552     // If the packet does not have a payload, it was a complete packet stored in
 553     // the channel for reference. This is the only case when a packet should
 554     // exist but have no payload. We re-allocate in this case.
 555     if (!hasPayload(packet)) {
 556         packet.buffer.reset(new SimpleBuffer(bufSize));
 557
 558         // Why do this again? In case it was copied from the old packet?
 559         hr.headerType = static_cast<PacketSize>(htype);
 560     }
 561
 562     // Resize anyway. If it's different from what it was before, we should
 563     // already have cleared it.
 564     packet.buffer->resize(bufSize);
 565     return true;
 566 }
 567
 568 bool
 569 RTMP::readPacketPayload(RTMPPacket& packet)
 570 {
 571     RTMPHeader& hr = packet.header;
 572
 573     const size_t bytesRead = packet.bytesRead;
 574
 575     const int nToRead = hr.dataSize - bytesRead;
 576
 577     const int nChunk = std::min<int>(nToRead, _inChunkSize);
 578     assert(nChunk >= 0);
 579
 580     // This is fine. We'll keep trying to read this payload until there
 581     // is enough data.
 582     if (readSocket(payloadData(packet) + bytesRead, nChunk) != nChunk) {
 583         return false;
 584     }
 585
 586     packet.bytesRead += nChunk;
 587
 588     return true;
 589 }
 590
 591 bool
 592 RTMP::sendPacket(RTMPPacket& packet)
 593 {
 594     // Set the data size of the packet to send.
 595     RTMPHeader& hr = packet.header;
 596
 597     hr.dataSize = payloadSize(packet);
 598
 599     // This is the timestamp for our message.
 600     const boost::uint32_t uptime = getUptime();
 601
 602     // Look at the previous packet on the channel.
 603     bool prev = hasPacket(CHANNELS_OUT, hr.channel);
 604
 605     // The packet shall be large if it contains an absolute timestamp.
 606     //      * This is necessary if there is no previous packet, or if the
 607     //        timestamp is smaller than the last packet.
 608     // Else it shall be medium if data size and packet type are the same
 609     // It shall be small if ...
 610     // It shall be minimal if it is exactly the same as its predecessor.
 611
 612     // All packets should start off as large. They will stay large if there
 613     // is no previous packet.
 614     assert(hr.headerType == RTMP_PACKET_SIZE_LARGE);
 615
 616     if (!prev) {
 617         hr._timestamp = uptime;
 618     }
 619     else {
 620
 621         const RTMPPacket& prevPacket = getPacket(CHANNELS_OUT, hr.channel);
 622         const RTMPHeader& oldh = prevPacket.header;
 623         const boost::uint32_t prevTimestamp = oldh._timestamp;
 624
 625         // If this timestamp is later than the other and the difference fits
 626         // in 3 bytes, encode a relative one.
 627         if (uptime >= oldh._timestamp && uptime - prevTimestamp < 0xffffff) {
 628             //log_debug("Shrinking to medium");
 629             hr.headerType = RTMP_PACKET_SIZE_MEDIUM;
 630             hr._timestamp = uptime - prevTimestamp;
 631
 632             // It can be still smaller if the data size is the same.
 633             if (oldh.dataSize == hr.dataSize &&
 634                     oldh.packetType == hr.packetType) {
 635                 //log_debug("Shrinking to small");
 636                 hr.headerType = RTMP_PACKET_SIZE_SMALL;
 637                 // If there is no timestamp difference, the minimum size
 638                 // is possible.
 639                 if (hr._timestamp == 0) {
 640                     //log_debug("Shrinking to minimum");
 641                     hr.headerType = RTMP_PACKET_SIZE_MINIMUM;
 642                 }
 643             }
 644         }
 645         else {
 646             // Otherwise we need an absolute one, so a large header.
 647             hr.headerType = RTMP_PACKET_SIZE_LARGE;
 648             hr._timestamp = uptime;
 649         }
 650     }
 651
 652     assert (hr.headerType < 4);
 653
 654     int nSize = packetSize[hr.headerType];
 655
 656     int hSize = nSize;
 657     boost::uint8_t* header;
 658     boost::uint8_t* hptr;
 659     boost::uint8_t* hend;
 660     boost::uint8_t c;
 661
 662     // If there is a payload, the same buffer is used to write the header.
 663     // Otherwise a separate buffer is used. But as we write them separately
 664     // anyway, why do we do that?
 665
 666     // Work out where the beginning of the header is.
 667     header = payloadData(packet) - nSize;
 668     hend = payloadData(packet);
 669
 670     // The header size includes only a single channel/type. If we need more,
 671     // they have to be added on.
 672     const int channelSize = hr.channel > 319 ? 3 : hr.channel > 63 ? 1 : 0;
 673     header -= channelSize;
 674     hSize += channelSize;
 675
 676     /// Add space for absolute timestamp if necessary.
 677     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 678         header -= 4;
 679         hSize += 4;
 680     }
 681
 682     hptr = header;
 683     c = hr.headerType << 6;
 684     switch (channelSize) {
 685         case 0:
 686             c |= hr.channel;
 687             break;
 688         case 1:
 689             break;
 690         case 2:
 691             c |= 1;
 692             break;
 693     }
 694     *hptr++ = c;
 695
 696     if (channelSize) {
 697         const int tmp = hr.channel - 64;
 698         *hptr++ = tmp & 0xff;
 699         if (channelSize == 2) *hptr++ = tmp >> 8;
 700     }
 701
 702     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 703         // Signify that the extended timestamp field is present.
 704         const boost::uint32_t t = 0xffffff;
 705         hptr = encodeInt24(hptr, hend, t);
 706     }
 707     else if (hr.headerType != RTMP_PACKET_SIZE_MINIMUM) {
 708         // Write absolute or relative timestamp. Only minimal packets have
 709         // no timestamp.
 710         hptr = encodeInt24(hptr, hend, hr._timestamp);
 711     }
 712
 713     /// Encode dataSize and packet type for medium packets.
 714     if (nSize > 4) {
 715         hptr = encodeInt24(hptr, hend, hr.dataSize);
 716         *hptr++ = hr.packetType;
 717     }
 718
 719     /// Encode streamID for large packets.
 720     if (hr.headerType == RTMP_PACKET_SIZE_LARGE) {
 721         hptr += encodeInt32LE(hptr, hr._streamID);
 722     }
 723
 724     // Encode extended absolute timestamp if needed.
 725     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 726         hptr += encodeInt32LE(hptr, hr._timestamp);
 727     }
 728
 729     nSize = hr.dataSize;
 730     boost::uint8_t *buffer = payloadData(packet);
 731     int nChunkSize = _outChunkSize;
 732
 733     std::string hx = hexify(header, payloadEnd(packet) - header, false);
 734
 735     while (nSize + hSize) {
 736
 737         if (nSize < nChunkSize) nChunkSize = nSize;
 738
 739         // First write header.
 740         if (header) {
 741             const int chunk = nChunkSize + hSize;
 742             if (_socket.write(header, chunk) != chunk) {
 743                 return false;
 744             }
 745             header = NULL;
 746             hSize = 0;
 747         }
 748
 749         else {
 750             // Then write data.
 751             if (_socket.write(buffer, nChunkSize) != nChunkSize) {
 752                 return false;
 753           }
 754
 755         }
 756
 757         nSize -= nChunkSize;
 758         buffer += nChunkSize;
 759
 760         if (nSize > 0) {
 761             header = buffer - 1;
 762             hSize = 1;
 763             if (channelSize) {
 764                 header -= channelSize;
 765                 hSize += channelSize;
 766             }
 767
 768             *header = (0xc0 | c);
 769             if (channelSize) {
 770                 int tmp = hr.channel - 64;
 771                 header[1] = tmp & 0xff;
 772                 if (channelSize == 2) header[2] = tmp >> 8;
 773             }
 774         }
 775     }
 776
 777     /* we invoked a remote method */
 778     if (hr.packetType == PACKET_TYPE_INVOKE) {
 779         assert(payloadData(packet)[0] == amf::STRING_AMF0);
 780         const boost::uint8_t* pos = payloadData(packet) + 1;
 781         const boost::uint8_t* end = payloadEnd(packet);
 782         const std::string& s = amf::readString(pos, end);
 783         log_debug( "Calling remote method %s", s);
 784     }
 785
 786     RTMPPacket& storedpacket = storePacket(CHANNELS_OUT, hr.channel, packet);
 787
 788     // Make it absolute for the next delta.
 789     storedpacket.header._timestamp = uptime;
 790
 791     return true;
 792 }
 793
 794 void
 795 RTMP::close()
 796 {
 797     _socket.close();
 798     _inChannels.clear();
 799     _outChannels.clear();
 800     _inChunkSize = RTMP_DEFAULT_CHUNKSIZE;
 801     _outChunkSize = RTMP_DEFAULT_CHUNKSIZE;
 802     _bytesIn = 0;
 803     _bytesInSent = 0;
 804     _bandwidth = 2500000;
 805     m_nClientBW2 = 2;
 806     _serverBandwidth = 2500000;
 807 }
 808
 809
 810 /////////////////////////////////////
 811 /// HandShaker implementation
 812 /////////////////////////////////////
 813
 814 HandShaker::HandShaker(Socket& s)
 815     :
 816     _socket(s),
 817     _sendBuf(sigSize + 1),
 818     _recvBuf(sigSize + 1),
 819     _error(false),
 820     _complete(false),
 821     _stage(0)
 822 {
 823     // Not encrypted
 824     _sendBuf[0] = 0x03;
 825
 826     // TODO: do this properly.
 827     boost::uint32_t uptime = htonl(getUptime());
 828
 829     boost::uint8_t* ourSig = &_sendBuf.front() + 1;
 830     std::memcpy(ourSig, &uptime, 4);
 831     std::fill_n(ourSig + 4, 4, 0);
 832
 833     // Generate 1536 random bytes.
 834     std::generate(ourSig + 8, ourSig + sigSize, RandomByte());
 835
 836 }
 837
 838
 839 /// Calls the next stage in the handshake process.
 840 void
 841 HandShaker::call()
 842 {
 843     if (error() || !_socket.connected()) return;
 844
 845     switch (_stage) {
 846         case 0:
 847             if (!stage0()) return;
 848             _stage = 1;
 849         case 1:
 850             if (!stage1()) return;
 851             _stage = 2;
 852         case 2:
 853             if (!stage2()) return;
 854             _stage = 3;
 855         case 3:
 856             if (!stage3()) return;
 857             log_debug("Handshake completed");
 858             _complete = true;
 859     }
 860 }
 861
 862 bool
 863 HandShaker::stage0()
 864 {
 865     std::streamsize sent = _socket.write(&_sendBuf.front(), sigSize + 1);
 866
 867     // This should probably not happen, but we can try again. An error will
 868     // be signalled later if the socket is no longer usable.
 869     if (!sent) {
 870         log_error("Stage 1 socket not ready. This should not happen.");
 871         return false;
 872     }
 873
 874     /// If we sent the wrong amount of data, we can't recover.
 875     if (sent != sigSize + 1) {
 876         log_error("Could not send stage 1 data");
 877         _error = true;
 878         return false;
 879     }
 880     return true;
 881 }
 882
 883 bool
 884 HandShaker::stage1()
 885 {
 886
 887     std::streamsize read = _socket.read(&_recvBuf.front(), sigSize + 1);
 888
 889     if (!read) {
 890         // If we receive nothing, wait until the next try.
 891         return false;
 892     }
 893
 894     // The read should never return anything but 0 or what we asked for.
 895     assert (read == sigSize + 1);
 896
 897     if (_recvBuf[0] != _sendBuf[0]) {
 898         log_error( "Type mismatch: client sent %d, server answered %d",
 899                 _recvBuf[0], _sendBuf[0]);
 900     }
 901
 902     const boost::uint8_t* serverSig = &_recvBuf.front() + 1;
 903
 904     // decode server response
 905     boost::uint32_t suptime;
 906     std::memcpy(&suptime, serverSig, 4);
 907     suptime = ntohl(suptime);
 908
 909     log_debug("Server Uptime : %d", suptime);
 910     log_debug("FMS Version   : %d.%d.%d.%d",
 911             +serverSig[4], +serverSig[5], +serverSig[6], +serverSig[7]);
 912
 913     return true;
 914 }
 915
 916 bool
 917 HandShaker::stage2()
 918 {
 919
 920     std::streamsize sent = _socket.write(&_recvBuf.front() + 1, sigSize);
 921
 922     // This should probably not happen.
 923     if (!sent) return false;
 924
 925     if (sent != sigSize) {
 926         log_error("Could not send complete signature.");
 927         _error = true;
 928         return false;
 929     }
 930
 931     return true;
 932 }
 933
 934 bool
 935 HandShaker::stage3()
 936 {
 937
 938     // Expect it back again.
 939     std::streamsize got = _socket.read(&_recvBuf.front(), sigSize);
 940
 941     if (!got) return false;
 942
 943     assert(got == sigSize);
 944
 945     const boost::uint8_t* serverSig = &_recvBuf.front();
 946     const boost::uint8_t* ourSig = &_sendBuf.front() + 1;
 947
 948     const bool match = std::equal(serverSig, serverSig + sigSize, ourSig);
 949
 950     // Should we set an error here?
 951     if (!match) {
 952         log_error( "Signatures do not match during handshake!");
 953     }
 954     return true;
 955 }
 956
 957 /// The type of Ping packet is 0x4 and contains two mandatory parameters
 958 /// and two optional parameters. The first parameter is
 959 /// the type of Ping and in short integer. The second parameter is the
 960 /// target of the ping. As Ping is always sent in Channel 2
 961 /// (control channel) and the target object in RTMP header is always 0 whicj
 962 /// means the Connection object, it's necessary to put an extra parameter
 963 /// to indicate the exact target object the Ping is sent to. The second
 964 /// parameter takes this responsibility. The value has the same meaning
 965 /// as the target object field in RTMP header. (The second value could also
 966 /// be used as other purposes, like RTT Ping/Pong. It is used as the
 967 /// timestamp.) The third and fourth parameters are optional and could be
 968 /// looked upon as the parameter of the Ping packet.
 969 bool
 970 sendCtrl(RTMP& r, ControlType t, unsigned int nObject, unsigned int nTime)
 971 {
 972     log_debug( "Sending control type %s %s", +t, t);
 973
 974     RTMPPacket packet(256);
 975
 976     packet.header.channel = CHANNEL_CONTROL1;
 977     packet.header.headerType = RTMP_PACKET_SIZE_LARGE;
 978     packet.header.packetType = PACKET_TYPE_CONTROL;
 979
 980     // type 3 is the buffer time and requires all 3 parameters.
 981     // all in all 10 bytes.
 982     int nSize = (t == CONTROL_BUFFER_TIME ? 10 : 6);
 983     if (t == CONTROL_RESPOND_VERIFY) nSize = 44;
 984
 985     SimpleBuffer& buf = *packet.buffer;
 986
 987     buf.appendNetworkShort(t);
 988
 989     if (t == CONTROL_RESPOND_VERIFY) { }
 990     else {
 991         if (nSize > 2) buf.appendNetworkLong(nObject);
 992         if (nSize > 6) buf.appendNetworkLong(nTime);
 993     }
 994     return r.sendPacket(packet);
 995 }
 996
 997 namespace {
 998
 999
1000 bool
1001 sendBytesReceived(RTMP* r)
1002 {
1003     RTMPPacket packet(4);
1004
1005     packet.header.channel = CHANNEL_CONTROL1;
1006     packet.header.packetType = PACKET_TYPE_BYTES_READ;
1007
1008     SimpleBuffer& buf = *packet.buffer;
1009
1010     buf.appendNetworkLong(r->_bytesIn);
1011     r->_bytesInSent = r->_bytesIn;
1012
1013     return r->sendPacket(packet);
1014 }
1015
1016
1017 void
1018 handleMetadata(RTMP& /*r*/, const boost::uint8_t* /* payload*/,
1019         unsigned int /*len*/)
1020 {
1021     return;
1022 }
1023
1024 void
1025 handleChangeChunkSize(RTMP& r, const RTMPPacket& packet)
1026 {
1027     if (payloadSize(packet) >= 4) {
1028         r._inChunkSize = amf::readNetworkLong(payloadData(packet));
1029         log_debug( "Changed chunk size to %d", r._inChunkSize);
1030     }
1031 }
1032
1033 void
1034 handleControl(RTMP& r, const RTMPPacket& packet)
1035 {
1036
1037     const size_t size = payloadSize(packet);
1038
1039     if (size < 2) {
1040         log_error("Control packet too short");
1041         return;
1042     }
1043
1044     const ControlType t =
1045         static_cast<ControlType>(amf::readNetworkShort(payloadData(packet)));
1046
1047     if (size < 6) {
1048         log_error("Control packet (%s) data too short", t);
1049         return;
1050     }
1051
1052     const int arg = amf::readNetworkLong(payloadData(packet) + 2);
1053     log_debug( "Received control packet %s with argument %s", t, arg);
1054
1055     switch (t)
1056     {
1057
1058         case CONTROL_CLEAR_STREAM:
1059             // TODO: handle this.
1060             break;
1061
1062         case CONTROL_CLEAR_BUFFER:
1063             // TODO: handle this.
1064             break;
1065
1066         case CONTROL_STREAM_DRY:
1067             break;
1068
1069         case CONTROL_RESET_STREAM:
1070             log_debug("Stream is recorded: %s", arg);
1071             break;
1072
1073         case CONTROL_PING:
1074             sendCtrl(r, CONTROL_PONG, arg, 0);
1075             break;
1076
1077         case CONTROL_BUFFER_EMPTY:
1078             // TODO: handle.
1079             break;
1080
1081         case CONTROL_BUFFER_READY:
1082             // TODO: handle
1083             break;
1084
1085         default:
1086             log_error("Received unknown or unhandled control %s", t);
1087             break;
1088     }
1089
1090 }
1091
1092 void
1093 handleServerBW(RTMP& r, const RTMPPacket& packet)
1094 {
1095     const boost::uint32_t bw = amf::readNetworkLong(payloadData(packet));
1096     log_debug( "Server bandwidth is %s", bw);
1097     r.setServerBandwidth(bw);
1098 }
1099
1100 void
1101 handleClientBW(RTMP& r, const RTMPPacket& packet)
1102 {
1103     const boost::uint32_t bw = amf::readNetworkLong(payloadData(packet));
1104
1105     r.setBandwidth(bw);
1106
1107     if (payloadSize(packet) > 4) r.m_nClientBW2 = payloadData(packet)[4];
1108     else r.m_nClientBW2 = -1;
1109
1110     log_debug( "Client bandwidth is %d %d", r.bandwidth(), +r.m_nClientBW2);
1111 }
1112
1113
1114
1115 boost::int32_t
1116 decodeInt32LE(const boost::uint8_t* c)
1117 {
1118     return (c[3] << 24) | (c[2] << 16) | (c[1] << 8) | c[0];
1119 }
1120
1121 int
1122 encodeInt32LE(boost::uint8_t *output, int nVal)
1123 {
1124     output[0] = nVal;
1125     nVal >>= 8;
1126     output[1] = nVal;
1127     nVal >>= 8;
1128     output[2] = nVal;
1129     nVal >>= 8;
1130     output[3] = nVal;
1131     return 4;
1132 }
1133
1134 void
1135 setupInvokePacket(RTMPPacket& packet)
1136 {
1137     RTMPHeader& hr = packet.header;
1138     // Control channel
1139     hr.channel = CHANNEL_CONTROL2;
1140     // Invoke
1141     hr.packetType = PACKET_TYPE_INVOKE;
1142 }
1143
1144 unsigned int
1145 decodeInt24(const boost::uint8_t *c)
1146 {
1147     unsigned int val;
1148     val = (c[0] << 16) | (c[1] << 8) | c[2];
1149     return val;
1150 }
1151
1152 boost::uint8_t*
1153 encodeInt16(boost::uint8_t *output, boost::uint8_t *outend, short nVal)
1154 {
1155     if (output+2 > outend) return NULL;
1156
1157     output[1] = nVal & 0xff;
1158     output[0] = nVal >> 8;
1159     return output + 2;
1160 }
1161
1162 boost::uint8_t*
1163 encodeInt24(boost::uint8_t *output, boost::uint8_t *outend, int nVal)
1164 {
1165     if (output + 3 > outend) return NULL;
1166
1167     output[2] = nVal & 0xff;
1168     output[1] = nVal >> 8;
1169     output[0] = nVal >> 16;
1170     return output+3;
1171 }
1172
1173 boost::uint8_t*
1174 encodeInt32(boost::uint8_t *output, boost::uint8_t *outend, int nVal)
1175 {
1176     if (output+4 > outend) return NULL;
1177
1178     output[3] = nVal & 0xff;
1179     output[2] = nVal >> 8;
1180     output[1] = nVal >> 16;
1181     output[0] = nVal >> 24;
1182     return output + 4;
1183 }
1184
1185 boost::uint32_t
1186 getUptime()
1187 {
1188 #if !defined(_WIN32) && !defined(__amigaos4__)
1189     struct tms t;
1190     return times(&t) * 1000 / sysconf(_SC_CLK_TCK);
1191 #elif defined(__amigaos4__)
1192     struct tms t;
1193     return times(&t) * 1000 / 50;
1194 #else
1195     return std::clock() * 100 / CLOCKS_PER_SEC;
1196 #endif
1197 }
1198
1199 } // anonymous namespace
1200
1201 std::ostream&
1202 operator<<(std::ostream& o, PacketType p)
1203 {
1204     switch(p) {
1205         case PACKET_TYPE_CHUNK_SIZE:
1206             return o << "<chunk size packet>";
1207         case PACKET_TYPE_BYTES_READ:
1208             return o << "<bytes read packet>";
1209         case PACKET_TYPE_CONTROL:
1210             return o << "<control packet>";
1211         case PACKET_TYPE_SERVERBW:
1212             return o << "<server bw packet>";
1213         case PACKET_TYPE_CLIENTBW:
1214             return o << "<client bw packet>";
1215         case PACKET_TYPE_AUDIO:
1216             return o << "<audio packet>";
1217         case PACKET_TYPE_VIDEO:
1218             return o << "<video packet>";
1219         case PACKET_TYPE_FLEX_STREAM_SEND:
1220             return o << "<flex stream send packet>";
1221         case PACKET_TYPE_FLEX_SHARED_OBJECT:
1222             return o << "<flex sharedobject packet>";
1223         case PACKET_TYPE_FLEX_MESSAGE:
1224             return o << "<flex message packet>";
1225         case PACKET_TYPE_METADATA:
1226             return o << "<metadata packet>";
1227         case PACKET_TYPE_SHARED_OBJECT:
1228             return o << "<sharedobject packet>";
1229         case PACKET_TYPE_INVOKE:
1230             return o << "<invoke packet>";
1231         case PACKET_TYPE_FLV:
1232             return o << "<flv packet>";
1233         default:
1234             return o << "<unknown packet type " << +p << ">";
1235     }
1236 }
1237
1238 std::ostream&
1239 operator<<(std::ostream& o, ControlType t)
1240 {
1241     switch (t) {
1242
1243         case CONTROL_CLEAR_STREAM:
1244             return o << "<clear stream>";
1245         case CONTROL_CLEAR_BUFFER:
1246             return o << "<clear buffer>";
1247         case CONTROL_STREAM_DRY:
1248             return o << "<stream dry>";
1249         case CONTROL_BUFFER_TIME:
1250             return o << "<buffer time>";
1251         case CONTROL_RESET_STREAM:
1252             return o << "<reset stream>";
1253         case CONTROL_PING:
1254             return o << "<ping>";
1255         case CONTROL_PONG:
1256             return o << "<pong>";
1257         case CONTROL_REQUEST_VERIFY:
1258             return o << "<verify request>";
1259         case CONTROL_RESPOND_VERIFY:
1260             return o << "<verify response>";
1261         case CONTROL_BUFFER_EMPTY:
1262             return o << "<buffer empty>";
1263         case CONTROL_BUFFER_READY:
1264             return o << "<buffer ready>";
1265         default:
1266             return o << "<unknown control " << +t << ">";
1267     }
1268 }
1269
1270 } // namespace rtmp
1271 } // namespace gnash