| 2574 | } |
| 2575 | |
| 2576 | bool RtmpChunkStream::OnCreateStream(const RtmpMessageHeader& mh, |
| 2577 | AMFInputStream* istream, |
| 2578 | Socket* socket) { |
| 2579 | RtmpService* service = connection_context()->service(); |
| 2580 | if (service == NULL) { |
| 2581 | RTMP_ERROR(socket, mh) << "Client should not receive `createStream'"; |
| 2582 | return false; |
| 2583 | } |
| 2584 | double transaction_id = 0; |
| 2585 | if (!ReadAMFNumber(&transaction_id, istream)) { |
| 2586 | RTMP_ERROR(socket, mh) << "Fail to read createStream.TransactionId"; |
| 2587 | return false; |
| 2588 | } |
| 2589 | bool is_publish = false; |
| 2590 | RtmpPublishType publish_type = RTMP_PUBLISH_LIVE; |
| 2591 | std::string stream_name; |
| 2592 | AMFObject cmd_obj; |
| 2593 | if (!ReadAMFObject(&cmd_obj, istream)) { |
| 2594 | RTMP_ERROR(socket, mh) << "Fail to read createStream.CommandObject"; |
| 2595 | return false; |
| 2596 | } |
| 2597 | const AMFField* cmd_name_field = cmd_obj.Find("CommandName"); |
| 2598 | if (cmd_name_field != NULL && cmd_name_field->IsString()) { |
| 2599 | is_publish = (cmd_name_field->AsString() == "publish"); |
| 2600 | } |
| 2601 | const AMFField* stream_name_field = cmd_obj.Find("StreamName"); |
| 2602 | if (stream_name_field != NULL && stream_name_field->IsString()) { |
| 2603 | stream_name_field->AsString().CopyToString(&stream_name); |
| 2604 | } |
| 2605 | if (is_publish) { |
| 2606 | const AMFField* publish_type_field = cmd_obj.Find("PublishType"); |
| 2607 | if (publish_type_field != NULL && publish_type_field->IsString()) { |
| 2608 | Str2RtmpPublishType(publish_type_field->AsString(), &publish_type); |
| 2609 | } |
| 2610 | } |
| 2611 | RPC_VLOG << socket->remote_side() << "[" << mh.stream_id |
| 2612 | << "] createStream{transaction_id=" << transaction_id << '}'; |
| 2613 | std::string error_text; |
| 2614 | butil::intrusive_ptr<RtmpServerStream> stream( |
| 2615 | service->NewStream(connection_context()->_connect_req)); |
| 2616 | if (connection_context()->_connect_req.stream_multiplexing() && |
| 2617 | stream != NULL) { |
| 2618 | stream->_client_supports_stream_multiplexing = true; |
| 2619 | } |
| 2620 | if (NULL == stream) { |
| 2621 | error_text = "Fail to create stream"; |
| 2622 | LOG(ERROR) << error_text; |
| 2623 | } else { |
| 2624 | socket->ReAddress(&stream->_rtmpsock); |
| 2625 | if (!connection_context()->AddServerStream(stream.get())) { |
| 2626 | error_text = "Fail to add stream"; |
| 2627 | LOG(ERROR) << error_text; |
| 2628 | } else { |
| 2629 | const int rc = bthread_id_create(&stream->_onfail_id, stream.get(), |
| 2630 | RtmpServerStream::RunOnFailed); |
| 2631 | if (rc) { |
| 2632 | LOG(ERROR) << "Fail to create RtmpServerStream._onfail_id: " |
| 2633 | << berror(rc); |
nothing calls this directly
no test coverage detected