| 1691 | } |
| 1692 | |
| 1693 | StreamUserData* RtmpClientStream::OnCreatingStream( |
| 1694 | SocketUniquePtr* inout, Controller* cntl) { |
| 1695 | { |
| 1696 | std::unique_lock<butil::Mutex> mu(_state_mutex); |
| 1697 | if (_state == STATE_ERROR || _state == STATE_DESTROYING) { |
| 1698 | cntl->SetFailed(EINVAL, "Fail to replace socket for stream, _state is error or destroying"); |
| 1699 | return NULL; |
| 1700 | } |
| 1701 | } |
| 1702 | SocketId esid; |
| 1703 | if (cntl->connection_type() == CONNECTION_TYPE_SHORT) { |
| 1704 | if (_client_impl->CreateSocket((*inout)->remote_side(), &esid) != 0) { |
| 1705 | cntl->SetFailed(EINVAL, "Fail to create RTMP socket"); |
| 1706 | return NULL; |
| 1707 | } |
| 1708 | } else { |
| 1709 | if (_client_impl->socket_map().Insert( |
| 1710 | SocketMapKey((*inout)->remote_side()), &esid) != 0) { |
| 1711 | cntl->SetFailed(EINVAL, "Fail to get the RTMP socket"); |
| 1712 | return NULL; |
| 1713 | } |
| 1714 | } |
| 1715 | SocketUniquePtr tmp_ptr; |
| 1716 | if (Socket::Address(esid, &tmp_ptr) != 0) { |
| 1717 | cntl->SetFailed(EFAILEDSOCKET, "Fail to address RTMP SocketId=%" PRIu64 |
| 1718 | " from SocketMap of RtmpClient=%p", |
| 1719 | esid, _client_impl.get()); |
| 1720 | return NULL; |
| 1721 | } |
| 1722 | RPC_VLOG << "Replace Socket For Stream, RTMP socketId=" << esid |
| 1723 | << ", main socketId=" << (*inout)->id(); |
| 1724 | tmp_ptr->ShareStats(inout->get()); |
| 1725 | inout->reset(tmp_ptr.release()); |
| 1726 | return this; |
| 1727 | } |
| 1728 | |
| 1729 | int RtmpClientStream::RunOnFailed(bthread_id_t id, void* data, int) { |
| 1730 | butil::intrusive_ptr<RtmpClientStream> stream( |
no test coverage detected