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