source: trunk/sources/thelib/src/protocols/rtmp/rtmpprotocolserializer.cpp @ 765

Revision 765, 26.9 KB checked in by shiretu, 13 days ago (diff)

-- fixes from EvoStream?
Line 
1/*
2 *  Copyright (c) 2010,
3 *  Gavriloaie Eugen-Andrei (shiretu@gmail.com)
4 *
5 *  This file is part of crtmpserver.
6 *  crtmpserver is free software: you can redistribute it and/or modify
7 *  it under the terms of the GNU General Public License as published by
8 *  the Free Software Foundation, either version 3 of the License, or
9 *  (at your option) any later version.
10 *
11 *  crtmpserver is distributed in the hope that it will be useful,
12 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
13 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 *  GNU General Public License for more details.
15 *
16 *  You should have received a copy of the GNU General Public License
17 *  along with crtmpserver.  If not, see <http://www.gnu.org/licenses/>.
18 */
19
20#ifdef HAS_PROTOCOL_RTMP
21#include "protocols/rtmp/rtmpprotocolserializer.h"
22#include "protocols/rtmp/messagefactories/messagefactories.h"
23#include "protocols/rtmp/amftypes.h"
24
25RTMPProtocolSerializer::RTMPProtocolSerializer() {
26}
27
28RTMPProtocolSerializer::~RTMPProtocolSerializer() {
29}
30
31string RTMPProtocolSerializer::GetUserCtrlTypeString(uint16_t type) {
32        switch (type) {
33                case RM_USRCTRL_TYPE_STREAM_BEGIN:
34                        return "RM_USRCTRL_TYPE_STREAM_BEGIN";
35                case RM_USRCTRL_TYPE_STREAM_EOF:
36                        return "RM_USRCTRL_TYPE_STREAM_EOF";
37                case RM_USRCTRL_TYPE_STREAM_DRY:
38                        return "RM_USRCTRL_TYPE_STREAM_DRY";
39                case RM_USRCTRL_TYPE_STREAM_IS_RECORDED:
40                        return "RM_USRCTRL_TYPE_STREAM_IS_RECORDED";
41                case RM_USRCTRL_TYPE_STREAM_SET_BUFFER_LENGTH:
42                        return "RM_USRCTRL_TYPE_STREAM_SET_BUFFER_LENGTH";
43                case RM_USRCTRL_TYPE_PING_REQUEST:
44                        return "RM_USRCTRL_TYPE_PING_REQUEST";
45                case RM_USRCTRL_TYPE_PING_RESPONSE:
46                        return "RM_USRCTRL_TYPE_PING_RESPONSE";
47                case RM_USRCTRL_TYPE_UNKNOWN1:
48                        return "RM_USRCTRL_TYPE_UNKNOWN1";
49                case RM_USRCTRL_TYPE_UNKNOWN2:
50                        return "RM_USRCTRL_TYPE_UNKNOWN2";
51                default:
52                        return format("#unknownUCT(%hu)", type);
53        }
54}
55
56string RTMPProtocolSerializer::GetSOPrimitiveString(uint8_t type) {
57        switch (type) {
58                case SOT_CS_CONNECT:
59                        return "SOT_CS_CONNECT";
60                case SOT_CS_DISCONNECT:
61                        return "SOT_CS_DISCONNECT";
62                case SOT_CS_UPDATE_FIELD_REQUEST:
63                        return "SOT_CS_UPDATE_FIELD_REQUEST";
64                case SOT_CS_UPDATE_FIELD:
65                        return "SOT_CS_UPDATE_FIELD";
66                case SOT_CS_UPDATE_FIELD_ACK:
67                        return "SOT_CS_UPDATE_FIELD_ACK";
68                case SOT_BW_SEND_MESSAGE:
69                        return "SOT_BW_SEND_MESSAGE";
70                case SOT_SC_STATUS:
71                        return "SOT_SC_STATUS";
72                case SOT_SC_CLEAR_DATA:
73                        return "SOT_SC_CLEAR_DATA";
74                case SOT_SC_DELETE_FIELD:
75                        return "SOT_SC_DELETE_FIELD";
76                case SOT_CS_DELETE_FIELD_REQUEST:
77                        return "SOT_CS_DELETE_FIELD_REQUEST";
78                case SOT_SC_INITIAL_DATA:
79                        return "SOT_SC_INITIAL_DATA";
80                default:
81                        return format("#unknownSOP(%hhu)", type);
82        }
83}
84
85bool RTMPProtocolSerializer::Deserialize(Header &header, IOBuffer &buffer,
86                Variant &message) {
87        message.Reset();
88        message[RM_HEADER] = header.GetVariant();
89        switch (H_MT(header)) {
90                case RM_HEADER_MESSAGETYPE_NOTIFY:
91                {
92                        return DeserializeNotify(buffer, message[RM_NOTIFY]);
93                }
94                case RM_HEADER_MESSAGETYPE_FLEXSTREAMSEND:
95                {
96                        return DeserializeFlexStreamSend(buffer, message[RM_FLEXSTREAMSEND]);
97                }
98                case RM_HEADER_MESSAGETYPE_INVOKE:
99                case RM_HEADER_MESSAGETYPE_FLEX:
100                {
101                        message[RM_INVOKE][RM_INVOKE_IS_FLEX] =
102                                        (bool)(H_MT(header) == RM_HEADER_MESSAGETYPE_FLEX);
103                        return DeserializeInvoke(buffer, message[RM_INVOKE]);
104                }
105                case RM_HEADER_MESSAGETYPE_FLEXSHAREDOBJECT:
106                {
107                        //TODO: This is a hack. We store the data on RM_SHAREDOBJECT
108                        //instead of RM_FLEXSHAREDOBJECT
109                        //return DeserializeFlexSharedObject(buffer, message[RM_FLEXSHAREDOBJECT]);
110                        return DeserializeFlexSharedObject(buffer, message[RM_SHAREDOBJECT]);
111                }
112                case RM_HEADER_MESSAGETYPE_SHAREDOBJECT:
113                {
114                        return DeserializeSharedObject(buffer, message[RM_SHAREDOBJECT]);
115                }
116                case RM_HEADER_MESSAGETYPE_USRCTRL:
117                {
118                        return DeserializeUsrCtrl(buffer, message[RM_USRCTRL]);
119                }
120                case RM_HEADER_MESSAGETYPE_CHUNKSIZE:
121                {
122                        return DeserializeChunkSize(buffer, message[RM_CHUNKSIZE]);
123                }
124                case RM_HEADER_MESSAGETYPE_ACK:
125                {
126                        return DeserializeAck(buffer, message[RM_ACK]);
127                }
128                case RM_HEADER_MESSAGETYPE_WINACKSIZE:
129                {
130                        return DeserializeWinAckSize(buffer, message[RM_WINACKSIZE]);
131                }
132                case RM_HEADER_MESSAGETYPE_PEERBW:
133                {
134                        return DeserializePeerBW(buffer, message[RM_PEERBW]);
135                }
136                case RM_HEADER_MESSAGETYPE_ABORTMESSAGE:
137                {
138                        return DeserializeAbortMessage(buffer, message[RM_ABORTMESSAGE]);
139                }
140                default:
141                {
142                        FATAL("Invalid message type: %u %s", H_MT(header), STR(buffer));
143                        return false;
144                }
145        }
146}
147
148bool RTMPProtocolSerializer::Serialize(Channel &channel,
149                Variant &message, IOBuffer &buffer, uint32_t chunkSize) {
150        bool result = false;
151        _internalBuffer.Ignore(GETAVAILABLEBYTESCOUNT(_internalBuffer));
152
153        switch ((uint32_t) VH_MT(message)) {
154                case RM_HEADER_MESSAGETYPE_INVOKE:
155                {
156                        result = SerializeInvoke(_internalBuffer, message[RM_INVOKE]);
157                        break;
158                }
159                case RM_HEADER_MESSAGETYPE_NOTIFY:
160                {
161                        result = SerializeNotify(_internalBuffer, message[RM_NOTIFY]);
162                        break;
163                }
164                case RM_HEADER_MESSAGETYPE_FLEXSTREAMSEND:
165                {
166                        result = SerializeFlexStreamSend(_internalBuffer, message[RM_FLEXSTREAMSEND]);
167                        break;
168                }
169                case RM_HEADER_MESSAGETYPE_FLEXSHAREDOBJECT:
170                {
171                        result = SerializeFlexSharedObject(_internalBuffer, message[RM_SHAREDOBJECT]);
172                        break;
173                }
174                case RM_HEADER_MESSAGETYPE_SHAREDOBJECT:
175                {
176                        result = SerializeSharedObject(_internalBuffer, message[RM_SHAREDOBJECT]);
177                        break;
178                }
179                case RM_HEADER_MESSAGETYPE_ACK:
180                {
181                        result = SerializeAck(_internalBuffer, message[RM_ACK]);
182                        break;
183                }
184                case RM_HEADER_MESSAGETYPE_USRCTRL:
185                {
186                        result = SerializeUsrCtrl(_internalBuffer, message[RM_USRCTRL]);
187                        break;
188                }
189                case RM_HEADER_MESSAGETYPE_CHUNKSIZE:
190                {
191                        result = SerializeChunkSize(_internalBuffer, message[RM_CHUNKSIZE]);
192                        break;
193                }
194                case RM_HEADER_MESSAGETYPE_WINACKSIZE:
195                {
196                        result = SerializeWinAckSize(_internalBuffer, message[RM_WINACKSIZE]);
197                        break;
198                }
199                case RM_HEADER_MESSAGETYPE_PEERBW:
200                {
201                        result = SerializeClientBW(_internalBuffer, message[RM_PEERBW]);
202                        break;
203                }
204                case RM_HEADER_MESSAGETYPE_ABORTMESSAGE:
205                {
206                        result = SerializeAbortMessage(_internalBuffer, message[RM_ABORTMESSAGE]);
207                        break;
208                }
209                default:
210                {
211                        FATAL("Invalid message type:\n%s", STR(message.ToString()));
212                }
213        }
214
215        //2. Check out the result
216        if (!result) {
217                FATAL("Unable to serialize message body");
218                return false;
219        }
220
221        //3. Update the message length
222        VH_ML(message) = GETAVAILABLEBYTESCOUNT(_internalBuffer);
223
224        //4. Extract the header
225        Header header;
226        if (!Header::GetFromVariant(header, message[RM_HEADER])) {
227                FATAL("Unable to read RTMP header: %s", STR(message.ToString()));
228                return false;
229        }
230
231        //5. Chunk and send the data
232        uint32_t available = 0;
233        while ((available = GETAVAILABLEBYTESCOUNT(_internalBuffer)) != 0) {
234                if (!header.Write(channel, buffer)) {
235                        FATAL("Unable to serialize message header");
236                        result = false;
237                }
238                if (available >= chunkSize) {
239                        buffer.ReadFromInputBuffer(&_internalBuffer, 0, chunkSize);
240                        channel.lastOutProcBytes += chunkSize;
241                        _internalBuffer.Ignore(chunkSize);
242                } else {
243                        buffer.ReadFromInputBuffer(&_internalBuffer, 0, available);
244                        channel.lastOutProcBytes += available;
245                        _internalBuffer.Ignore(available);
246                }
247        }
248        channel.lastOutProcBytes = 0;
249
250        //6. Done
251        return result;
252}
253
254bool RTMPProtocolSerializer::SerializeInvoke(IOBuffer &buffer,
255                Variant &message) {
256
257        string functionName = message[RM_INVOKE_FUNCTION];
258        if (!_amf0.WriteShortString(buffer, functionName)) {
259                FATAL("Unable to write %s", STR(RM_INVOKE_FUNCTION));
260                return false;
261        }
262
263        if (!_amf0.WriteDouble(buffer, message[RM_INVOKE_ID])) {
264                FATAL("Unable to write %s", STR(RM_INVOKE_ID));
265                return false;
266        }
267
268        FOR_MAP(message[RM_INVOKE_PARAMS], string, Variant, i) {
269                if (!_amf0.Write(buffer, MAP_VAL(i))) {
270                        FATAL("Unable to serialize invoke parameter %s: %s",
271                                        STR(MAP_KEY(i)),
272                                        STR(message.ToString()));
273                        return false;
274                }
275        }
276
277        return true;
278}
279
280bool RTMPProtocolSerializer::SerializeNotify(IOBuffer &buffer, Variant &message) {
281
282        FOR_MAP(message[RM_NOTIFY_PARAMS], string, Variant, i) {
283                if (!_amf0.Write(buffer, MAP_VAL(i))) {
284                        FATAL("Unable to serialize invoke parameter %s: %s",
285                                        STR(MAP_KEY(i)),
286                                        STR(message.ToString()));
287                        return false;
288                }
289        }
290        return true;
291}
292
293bool RTMPProtocolSerializer::SerializeFlexStreamSend(IOBuffer &buffer, Variant &message) {
294        buffer.ReadFromByte((uint8_t) (message[RM_FLEXSTREAMSEND_UNKNOWNBYTE]));
295
296        FOR_MAP(message[RM_FLEXSTREAMSEND_PARAMS], string, Variant, i) {
297                if (!_amf0.Write(buffer, MAP_VAL(i))) {
298                        FATAL("Unable to serialize invoke parameter %s: %s",
299                                        STR(MAP_KEY(i)),
300                                        STR(message.ToString()));
301                        return false;
302                }
303        }
304        return true;
305}
306
307bool RTMPProtocolSerializer::SerializeAck(IOBuffer &buffer, uint32_t value) {
308        if (!_amf0.WriteUInt32(buffer, value, false)) {
309                FATAL("Unable to write uint32_t value: %u", value);
310                return false;
311        }
312        return true;
313}
314
315bool RTMPProtocolSerializer::SerializeUsrCtrl(IOBuffer &buffer, Variant message) {
316        if (!_amf0.WriteInt16(buffer, message[RM_USRCTRL_TYPE], false)) {
317                FATAL("Unable to write user control message type value");
318                return false;
319        }
320
321        switch ((uint16_t) message[RM_USRCTRL_TYPE]) {
322                case RM_USRCTRL_TYPE_STREAM_BEGIN:
323                case RM_USRCTRL_TYPE_STREAM_EOF:
324                case RM_USRCTRL_TYPE_STREAM_DRY:
325                case RM_USRCTRL_TYPE_STREAM_IS_RECORDED:
326                {
327                        if (!_amf0.WriteInt32(buffer, message[RM_USRCTRL_STREAMID], false)) {
328                                FATAL("Unable to write stream id from user control message");
329                                return false;
330                        }
331                        return true;
332                }
333                case RM_USRCTRL_TYPE_PING_RESPONSE:
334                {
335                        if (!_amf0.WriteInt32(buffer, message[RM_USRCTRL_PONG], false)) {
336                                FATAL("Unable to write timestamp from ping response user control message");
337                                return false;
338                        }
339                        return true;
340                }
341                default:
342                {
343                        FATAL("Invalid user control message:\n%s", STR(message.ToString()));
344                        return false;
345                }
346        }
347}
348
349bool RTMPProtocolSerializer::SerializeChunkSize(IOBuffer &buffer, uint32_t value) {
350        if (!_amf0.WriteUInt32(buffer, value, false)) {
351                FATAL("Unable to write uint32_t value: %u", value);
352                return false;
353        }
354        return true;
355}
356
357bool RTMPProtocolSerializer::SerializeWinAckSize(IOBuffer &buffer, uint32_t value) {
358        if (!_amf0.WriteUInt32(buffer, value, false)) {
359                FATAL("Unable to write uint32_t value: %u", value);
360                return false;
361        }
362        return true;
363}
364
365bool RTMPProtocolSerializer::SerializeAbortMessage(IOBuffer &buffer, uint32_t value) {
366        if (!_amf0.WriteUInt32(buffer, value, false)) {
367                FATAL("Unable to write uint32_t value: %u", value);
368                return false;
369        }
370        return true;
371}
372
373bool RTMPProtocolSerializer::SerializeClientBW(IOBuffer &buffer, Variant value) {
374        if (!_amf0.WriteUInt32(buffer, value[RM_PEERBW_VALUE], false)) {
375                FATAL("Unable to write uint32_t value: %u",
376                                (uint32_t) value[RM_PEERBW_VALUE]);
377                return false;
378        }
379        if (!_amf0.WriteUInt8(buffer, value[RM_PEERBW_TYPE], false)) {
380                FATAL("Unable to write uint8_t value: %hhu",
381                                (uint8_t) value[RM_PEERBW_TYPE]);
382                return false;
383        }
384        return true;
385}
386
387bool RTMPProtocolSerializer::SerializeFlexSharedObject(IOBuffer &buffer, Variant &message) {
388        buffer.ReadFromByte(0);
389
390        return SerializeSharedObject(buffer, message);
391}
392
393bool RTMPProtocolSerializer::SerializeSharedObject(IOBuffer &buffer,
394                Variant &message) {
395        string name = message[RM_SHAREDOBJECT_NAME];
396
397        //name
398        if (!_amf0.WriteShortString(buffer, name, false)) {
399                FATAL("Unable to write %s", STR(RM_SHAREDOBJECT_NAME));
400                return false;
401        }
402
403        //version
404        if (!_amf0.WriteUInt32(buffer, message[RM_SHAREDOBJECT_VERSION], false)) {
405                FATAL("Unable to write %s", STR(RM_SHAREDOBJECT_VERSION));
406                return false;
407        }
408
409        //persistance
410        uint32_t persistence = message[RM_SHAREDOBJECT_PERSISTENCE] ? 2 : 0;
411        if (!_amf0.WriteUInt32(buffer, persistence, false)) {
412                FATAL("Unable to write %s", STR(RM_SHAREDOBJECT_PERSISTENCE));
413                return false;
414        }
415
416        //unknown bytes
417        if (!_amf0.WriteUInt32(buffer, 0, false)) {
418                FATAL("Unable to write a 0 value");
419                return false;
420        }
421
422        //primitives
423        for (uint32_t i = 0; i < message[RM_SHAREDOBJECT_PRIMITIVES].MapSize(); i++) {
424
425                Variant primitive = message[RM_SHAREDOBJECT_PRIMITIVES][i];
426
427                //type
428                if (!_amf0.WriteUInt8(buffer, primitive[RM_SHAREDOBJECTPRIMITIVE_TYPE],
429                                false)) {
430                        FATAL("Unable to write %s", STR(RM_SHAREDOBJECTPRIMITIVE_TYPE));
431                        return false;
432                }
433
434                switch ((uint8_t) primitive[RM_SHAREDOBJECTPRIMITIVE_TYPE]) {
435                        case SOT_CS_UPDATE_FIELD:
436                        case SOT_SC_INITIAL_DATA:
437                        case SOT_CS_UPDATE_FIELD_REQUEST:
438                        {
439                                uint32_t rawLengthPosition = buffer.GetCurrentWritePosition();
440                                //length
441                                if (!_amf0.WriteUInt32(buffer, 0, false)) {
442                                        FATAL("Unable to write data");
443                                        return false;
444                                }
445
446                                if (!primitive.HasKey(RM_SHAREDOBJECTPRIMITIVE_PAYLOAD))
447                                        break;
448
449                                FOR_MAP(primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD],
450                                                string, Variant, i) {
451                                        string key = MAP_KEY(i);
452                                        Variant value = MAP_VAL(i);
453                                        if (!_amf0.WriteShortString(buffer, key, false)) {
454                                                FATAL("Unable to write data: %s", STR(key));
455                                                return false;
456                                        }
457                                        if (!_amf0.Write(buffer, value)) {
458                                                FATAL("Unable to write data: %s", STR(value.ToString()));
459                                                return false;
460                                        }
461                                }
462
463                                uint32_t length = buffer.GetCurrentWritePosition()
464                                                - rawLengthPosition - 4;
465                                EHTONLP(buffer.GetPointer() + rawLengthPosition, length);
466                                break;
467                        }
468                        case SOT_SC_CLEAR_DATA:
469                        {
470                                if (!_amf0.WriteUInt32(buffer, 0, false)) {
471                                        FATAL("Unable to write data");
472                                        return false;
473                                }
474                                break;
475                        }
476                        case SOT_SC_DELETE_FIELD:
477                        case SOT_CS_UPDATE_FIELD_ACK:
478                        case SOT_CS_DELETE_FIELD_REQUEST:
479                        {
480                                uint32_t rawLengthPosition = buffer.GetCurrentWritePosition();
481                                //length
482                                if (!_amf0.WriteUInt32(buffer, 0, false)) {
483                                        FATAL("Unable to write data");
484                                        return false;
485                                }
486
487                                Variant &payload = primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD];
488                                for (uint32_t i = 0; i < payload.MapSize(); i++) {
489                                        string name = payload[i];
490                                        if (!_amf0.WriteShortString(buffer, name, false)) {
491                                                FATAL("Unable to write data: %s", STR(name));
492                                                return false;
493                                        }
494                                }
495
496                                uint32_t length = buffer.GetCurrentWritePosition()
497                                                - rawLengthPosition - 4;
498                                EHTONLP(buffer.GetPointer() + rawLengthPosition, length);
499                                break;
500                        }
501                        case SOT_BW_SEND_MESSAGE:
502                        {
503                                uint32_t rawLengthPosition = buffer.GetCurrentWritePosition();
504                                if (!_amf0.WriteUInt32(buffer, 0, false)) {
505                                        FATAL("Unable to write data");
506                                        return false;
507                                }
508
509                                Variant &payload = primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD];
510
511                                FOR_MAP(payload, string, Variant, i) {
512                                        if (!_amf0.Write(buffer, MAP_VAL(i))) {
513                                                FATAL("Unable to write data: %s", STR(MAP_VAL(i).ToString()));
514                                                return false;
515                                        }
516                                }
517                                uint32_t length = buffer.GetCurrentWritePosition()
518                                                - rawLengthPosition - 4;
519                                EHTONLP(buffer.GetPointer() + rawLengthPosition, length);
520                                break;
521                        }
522                        case SOT_CS_CONNECT:
523                        {
524                                //uint32_t rawLengthPosition = buffer.GetCurrentWritePosition();
525                                if (!_amf0.WriteUInt32(buffer, 0, false)) {
526                                        FATAL("Unable to write data");
527                                        return false;
528                                }
529                                break;
530                        }
531                        default:
532                        {
533                                FATAL("Unable to serialize primitive:\n%s",
534                                                STR(primitive.ToString()));
535                                return false;
536                        }
537                }
538        }
539        return true;
540}
541
542bool RTMPProtocolSerializer::DeserializeNotify(IOBuffer &buffer, Variant &message) {
543        for (uint32_t i = 0; GETAVAILABLEBYTESCOUNT(buffer) > 0; i++) {
544                if (!_amf0.Read(buffer, message[RM_NOTIFY_PARAMS][i])) {
545                        FATAL("Unable to de-serialize invoke parameter %u", i);
546                        return false;
547                }
548        }
549        return true;
550}
551
552bool RTMPProtocolSerializer::DeserializeFlexStreamSend(IOBuffer &buffer, Variant &message) {
553        AMF_CHECK_BOUNDARIES(buffer, 1);
554        message[RM_FLEXSTREAMSEND_UNKNOWNBYTE] = (uint8_t) GETIBPOINTER(buffer)[0];
555        if (!buffer.Ignore(1)) {
556                FATAL("Unable to ignore 1 byte");
557                return false;
558        }
559
560        for (uint32_t i = 0; GETAVAILABLEBYTESCOUNT(buffer) > 0; i++) {
561                if (!_amf0.Read(buffer, message[RM_FLEXSTREAMSEND_PARAMS][i])) {
562                        FATAL("Unable to de-serialize invoke parameter %u", i);
563                        return false;
564                }
565        }
566
567        return true;
568}
569
570bool RTMPProtocolSerializer::DeserializeInvoke(IOBuffer &buffer, Variant &message) {
571        if (message[RM_INVOKE_IS_FLEX]) {
572                if (!buffer.Ignore(1)) {
573                        FATAL("Unable to ignore 1 byte");
574                        return false;
575                }
576        }
577
578        if (!_amf0.ReadShortString(buffer, message[RM_INVOKE_FUNCTION])) {
579                FATAL("Unable to read %s", STR(RM_INVOKE_FUNCTION));
580                return false;
581        }
582
583        if (!_amf0.ReadDouble(buffer, message[RM_INVOKE_ID])) {
584                FATAL("Unable to read %s", STR(RM_INVOKE_ID));
585                return false;
586        }
587
588        for (uint32_t i = 0; GETAVAILABLEBYTESCOUNT(buffer) > 0; i++) {
589                if (!_amf0.Read(buffer, message[RM_INVOKE_PARAMS][i])) {
590                        FATAL("Unable to de-serialize invoke parameter %u", i);
591                        return false;
592                }
593        }
594
595        message[RM_INVOKE_PARAMS].IsArray(true);
596
597        return true;
598}
599
600bool RTMPProtocolSerializer::DeserializeAck(IOBuffer &buffer,
601                Variant &message) {
602        AMF_CHECK_BOUNDARIES(buffer, 4);
603        message = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
604        return buffer.Ignore(4);
605}
606
607bool RTMPProtocolSerializer::DeserializeUsrCtrl(IOBuffer &buffer, Variant &message) {
608        AMF_CHECK_BOUNDARIES(buffer, 2);
609        message[RM_USRCTRL_TYPE] = ENTOHSP(GETIBPOINTER(buffer)); //----MARKED-SHORT----
610        message[RM_USRCTRL_TYPE_STRING] = GetUserCtrlTypeString(message[RM_USRCTRL_TYPE]);
611        if (!buffer.Ignore(2)) {
612                FATAL("Unable to ignore 2 bytes");
613                return false;
614        }
615
616        switch ((uint16_t) message[RM_USRCTRL_TYPE]) {
617                case RM_USRCTRL_TYPE_STREAM_BEGIN:
618                case RM_USRCTRL_TYPE_STREAM_EOF:
619                case RM_USRCTRL_TYPE_STREAM_DRY:
620                case RM_USRCTRL_TYPE_STREAM_IS_RECORDED:
621                {
622                        AMF_CHECK_BOUNDARIES(buffer, 4);
623                        message[RM_USRCTRL_STREAMID] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
624                        if (!buffer.Ignore(4)) {
625                                FATAL("Unable to ignore 4 bytes");
626                                return false;
627                        }
628                        return true;
629                }
630                case RM_USRCTRL_TYPE_STREAM_SET_BUFFER_LENGTH:
631                {
632                        AMF_CHECK_BOUNDARIES(buffer, 8);
633                        message[RM_USRCTRL_STREAMID] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
634                        if (!buffer.Ignore(4)) {
635                                FATAL("Unable to ignore 4 bytes");
636                                return false;
637                        }
638                        message[RM_USRCTRL_BUFFLEN] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
639                        if (!buffer.Ignore(4)) {
640                                FATAL("Unable to ignore 4 bytes");
641                                return false;
642                        }
643                        return true;
644                }
645                case RM_USRCTRL_TYPE_PING_REQUEST:
646                {
647                        AMF_CHECK_BOUNDARIES(buffer, 4);
648                        message[RM_USRCTRL_PING] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
649                        if (!buffer.Ignore(4)) {
650                                FATAL("Unable to ignore 4 bytes");
651                                return false;
652                        }
653                        return true;
654                }
655                case RM_USRCTRL_TYPE_PING_RESPONSE:
656                {
657                        AMF_CHECK_BOUNDARIES(buffer, 4);
658                        message[RM_USRCTRL_PONG] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
659                        if (!buffer.Ignore(4)) {
660                                FATAL("Unable to ignore 4 bytes");
661                                return false;
662                        }
663                        return true;
664                }
665                case RM_USRCTRL_TYPE_UNKNOWN1:
666                case RM_USRCTRL_TYPE_UNKNOWN2:
667                {
668                        AMF_CHECK_BOUNDARIES(buffer, 4);
669                        message[RM_USRCTRL_UNKNOWN_U32] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
670                        if (!buffer.Ignore(4)) {
671                                FATAL("Unable to ignore 4 bytes");
672                                return false;
673                        }
674                        return true;
675                }
676                default:
677                {
678                        FATAL("Invalid user control message:\n%s;Buffer:\n%s",
679                                        STR(message.ToString()), STR(buffer));
680                        return false;
681                }
682        }
683}
684
685bool RTMPProtocolSerializer::DeserializeChunkSize(IOBuffer &buffer,
686                Variant &message) {
687        AMF_CHECK_BOUNDARIES(buffer, 4);
688        message = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
689        return buffer.Ignore(4);
690}
691
692bool RTMPProtocolSerializer::DeserializeWinAckSize(IOBuffer &buffer, Variant &message) {
693        AMF_CHECK_BOUNDARIES(buffer, 4);
694        message = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
695        return buffer.Ignore(4);
696}
697
698bool RTMPProtocolSerializer::DeserializePeerBW(IOBuffer &buffer, Variant &message) {
699        AMF_CHECK_BOUNDARIES(buffer, 4);
700        message[RM_PEERBW_VALUE] = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
701        if (!buffer.Ignore(4)) {
702                FATAL("Unable to ignore 4 bytes");
703                return false;
704        }
705        message[RM_PEERBW_TYPE] = *(uint8_t *) GETIBPOINTER(buffer);
706        return buffer.Ignore(1);
707}
708
709bool RTMPProtocolSerializer::DeserializeAbortMessage(IOBuffer &buffer, Variant &message) {
710        AMF_CHECK_BOUNDARIES(buffer, 4);
711        message = (uint32_t) ENTOHLP(GETIBPOINTER(buffer)); //----MARKED-LONG---
712        if (!buffer.Ignore(4)) {
713                FATAL("Unable to ignore 4 bytes");
714                return false;
715        }
716        return true;
717}
718
719bool RTMPProtocolSerializer::DeserializeFlexSharedObject(IOBuffer &buffer,
720                Variant &message) {
721        AMF_CHECK_BOUNDARIES(buffer, 1);
722        if (GETIBPOINTER(buffer)[0] != 0) {
723                FATAL("Encoding %"PRIu8" not supported yet", GETIBPOINTER(buffer)[0]);
724                return false;
725        }
726
727        if (!buffer.Ignore(1)) {
728                FATAL("Unable to ignore 1 bytes");
729                return false;
730        }
731
732        return DeserializeSharedObject(buffer, message);
733}
734
735bool RTMPProtocolSerializer::DeserializeSharedObject(IOBuffer &buffer, Variant &message) {
736        //1. Read the name
737        if (!_amf0.ReadShortString(buffer, message[RM_SHAREDOBJECT_NAME], false)) {
738                FATAL("Unable to read %s", STR(RM_SHAREDOBJECT_NAME));
739                return false;
740        }
741
742        //2. Read the current version
743        if (!_amf0.ReadUInt32(buffer, message[RM_SHAREDOBJECT_VERSION], false)) {
744                FATAL("Unable to read %s", STR(RM_SHAREDOBJECT_VERSION));
745                return false;
746        }
747
748        //3. Read the flags
749        Variant persistence;
750        if (!_amf0.ReadUInt32(buffer, persistence, false)) {
751                FATAL("Unable to read %s", STR(RM_SHAREDOBJECT_PERSISTENCE));
752                return false;
753        }
754        message[RM_SHAREDOBJECT_PERSISTENCE] = (uint32_t) persistence == 2;
755
756        //4. Skip 4 unknown bytes
757        AMF_CHECK_BOUNDARIES(buffer, 4);
758        if (!buffer.Ignore(4)) {
759                FATAL("Unable to ignore 4 bytes");
760                return false;
761        }
762
763        //5. Read the primitives as long as we have data in the buffer
764        Variant primitive;
765        uint32_t primitiveIndex = 0;
766        while (GETAVAILABLEBYTESCOUNT(buffer) > 0) {
767                primitive.Reset();
768
769                //6. Read the type
770                if (!_amf0.ReadUInt8(buffer, primitive[RM_SHAREDOBJECTPRIMITIVE_TYPE], false)) {
771                        FATAL("Unable to read %s", STR(RM_SHAREDOBJECTPRIMITIVE_TYPE));
772                        return false;
773                }
774                primitive[RM_SHAREDOBJECTPRIMITIVE_STRTYPE] =
775                                GetSOPrimitiveString(primitive[RM_SHAREDOBJECTPRIMITIVE_TYPE]);
776
777                //7. Read the length in bytes
778                if (!_amf0.ReadUInt32(buffer, primitive[RM_SHAREDOBJECTPRIMITIVE_RAWLENGTH], false)) {
779                        FATAL("Unable to read %s", STR(RM_SHAREDOBJECTPRIMITIVE_RAWLENGTH));
780                        return false;
781                }
782                uint32_t rawLength = primitive[RM_SHAREDOBJECTPRIMITIVE_RAWLENGTH];
783                AMF_CHECK_BOUNDARIES(buffer, rawLength);
784
785                //8. Read the rest of the primitive based on it's type
786                switch ((uint8_t) primitive[RM_SHAREDOBJECTPRIMITIVE_TYPE]) {
787                        case SOT_CS_CONNECT:
788                        {
789                                break;
790                        }
791                        case SOT_CS_DISCONNECT:
792                        {
793                                break;
794                        }
795                        case SOT_CS_UPDATE_FIELD_REQUEST:
796                        {
797                                uint32_t read = 0;
798                                uint32_t beforeRead = 0;
799                                uint32_t afterRead = 0;
800                                while (read < rawLength) {
801                                        Variant key;
802                                        Variant value;
803                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
804                                        if (!_amf0.ReadShortString(buffer, key, false)) {
805                                                FATAL("Unable to read key");
806                                                return false;
807                                        }
808                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
809                                        read += beforeRead - afterRead;
810
811                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
812                                        if (!_amf0.Read(buffer, value)) {
813                                                FATAL("Unable to read value");
814                                                return false;
815                                        }
816                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
817                                        read += beforeRead - afterRead;
818                                        primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD][key] = value;
819                                }
820                                if (read != rawLength) {
821                                        FATAL("length mismatch");
822                                        return false;
823                                }
824                                break;
825                        }
826                        case SOT_BW_SEND_MESSAGE:
827                        {
828                                uint32_t read = 0;
829                                uint32_t beforeRead = 0;
830                                uint32_t afterRead = 0;
831                                while (read < rawLength) {
832                                        Variant value;
833                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
834                                        if (!_amf0.Read(buffer, value)) {
835                                                FATAL("Unable to read value");
836                                                return false;
837                                        }
838                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
839                                        read += beforeRead - afterRead;
840                                        primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD].PushToArray(value);
841                                }
842                                if (read != rawLength) {
843                                        FATAL("length mismatch");
844                                        return false;
845                                }
846                                break;
847                        }
848                        case SOT_SC_STATUS:
849                        {
850                                NYIR;
851                                break;
852                        }
853                        case SOT_SC_CLEAR_DATA:
854                        {
855                                break;
856                        }
857                        case SOT_CS_UPDATE_FIELD_ACK:
858                        case SOT_SC_DELETE_FIELD:
859                        {
860                                uint32_t read = 0;
861                                uint32_t beforeRead = 0;
862                                uint32_t afterRead = 0;
863                                while (read < rawLength) {
864                                        Variant key;
865                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
866                                        if (!_amf0.ReadShortString(buffer, key, false)) {
867                                                FATAL("Unable to read key");
868                                                return false;
869                                        }
870                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
871                                        read += beforeRead - afterRead;
872                                        primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD].PushToArray(key);
873                                }
874                                if (read != rawLength) {
875                                        FATAL("length mismatch");
876                                        return false;
877                                }
878                                break;
879                        }
880                        case SOT_CS_DELETE_FIELD_REQUEST:
881                        {
882                                Variant value;
883                                uint32_t beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
884                                if (!_amf0.ReadShortString(buffer, value, false)) {
885                                        FATAL("Unable to read value");
886                                        return false;
887                                }
888                                uint32_t afterRead = GETAVAILABLEBYTESCOUNT(buffer);
889                                primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD].PushToArray(value);
890                                if ((beforeRead - afterRead) != rawLength) {
891                                        FATAL("length mismatch");
892                                        return false;
893                                }
894                                break;
895                        }
896                        case SOT_CS_UPDATE_FIELD:
897                        case SOT_SC_INITIAL_DATA:
898                        {
899                                primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD].IsArray(false);
900                                uint32_t read = 0;
901                                uint32_t beforeRead = 0;
902                                uint32_t afterRead = 0;
903                                while (read < rawLength) {
904                                        Variant key;
905                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
906                                        if (!_amf0.ReadShortString(buffer, key, false)) {
907                                                FATAL("Unable to read key");
908                                                return false;
909                                        }
910                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
911                                        read += beforeRead - afterRead;
912                                        if (read >= rawLength) {
913                                                FATAL("No more data");
914                                                return false;
915                                        }
916                                        Variant value;
917                                        beforeRead = GETAVAILABLEBYTESCOUNT(buffer);
918                                        if (!_amf0.Read(buffer, value)) {
919                                                FATAL("Unable to read value");
920                                                return false;
921                                        }
922                                        afterRead = GETAVAILABLEBYTESCOUNT(buffer);
923                                        read += beforeRead - afterRead;
924
925                                        primitive[RM_SHAREDOBJECTPRIMITIVE_PAYLOAD][key] = value;
926                                }
927                                if (read != rawLength) {
928                                        FATAL("length mismatch");
929                                        return false;
930                                }
931                                break;
932                        }
933                        default:
934                        {
935                                FATAL("Invalid SO primitive type. Partial result:\n%s",
936                                                STR(message.ToString()));
937                                return false;
938                        }
939                }
940
941                message[RM_SHAREDOBJECT_PRIMITIVES][(uint32_t) primitiveIndex] = primitive;
942                primitiveIndex++;
943        }
944
945        return true;
946}
947
948void RTMPProtocolSerializer::ChunkBuffer(IOBuffer &destination,
949                IOBuffer &source, uint32_t chunkSize, Channel &channel) {
950        uint32_t length = GETAVAILABLEBYTESCOUNT(source);
951        uint32_t chunksCount = length / chunkSize;
952        chunksCount += (length % chunkSize) == 0 ? 0 : 1;
953        uint32_t copySizeChunk = 0;
954
955        for (uint32_t i = 0; i < chunksCount - 1; i++) {
956                copySizeChunk = length > chunkSize ? chunkSize : length;
957                destination.ReadFromInputBuffer(&source, 0, copySizeChunk);
958                if (channel.id <= 63) {
959                        destination.ReadFromRepeat((uint8_t) (channel.id | 0xc0), 1);
960                } else {
961                        NYIA;
962                }
963                source.Ignore(copySizeChunk);
964                length -= copySizeChunk;
965                channel.lastOutProcBytes += length;
966        }
967
968        destination.ReadFromInputBuffer(&source, 0, length);
969}
970#endif /* HAS_PROTOCOL_RTMP */
971
Note: See TracBrowser for help on using the repository browser.