| 1783 | |
| 1784 | |
| 1785 | void RtmpClientStream::DestroyStreamCreator(Controller* cntl) { |
| 1786 | if (cntl->Failed()) { |
| 1787 | if (_rtmpsock != NULL && |
| 1788 | // ^ If sending_sock is NULL, the RPC fails before _pack_request |
| 1789 | // which calls AddTransaction, in another word, RemoveTransaction |
| 1790 | // is not needed. |
| 1791 | cntl->ErrorCode() != ERTMPCREATESTREAM) { |
| 1792 | // ^ ERTMPCREATESTREAM is triggered by receiving "_error" command, |
| 1793 | // RemoveTransaction should already be called. |
| 1794 | CHECK_LT(cntl->log_id(), (uint64_t)std::numeric_limits<uint32_t>::max()); |
| 1795 | const uint32_t transaction_id = cntl->log_id(); |
| 1796 | policy::RtmpContext* rtmp_ctx = |
| 1797 | static_cast<policy::RtmpContext*>(_rtmpsock->parsing_context()); |
| 1798 | if (rtmp_ctx == NULL) { |
| 1799 | LOG(FATAL) << "RtmpContext must be created"; |
| 1800 | } else { |
| 1801 | policy::RtmpTransactionHandler* handler = |
| 1802 | rtmp_ctx->RemoveTransaction(transaction_id); |
| 1803 | if (handler) { |
| 1804 | handler->Cancel(); |
| 1805 | } |
| 1806 | } |
| 1807 | } |
| 1808 | return OnFailedToCreateStream(); |
| 1809 | } |
| 1810 | |
| 1811 | int rc = 0; |
| 1812 | bthread_id_t onfail_id = INVALID_BTHREAD_ID; |
| 1813 | { |
| 1814 | std::unique_lock<butil::Mutex> mu(_state_mutex); |
| 1815 | switch (_state) { |
| 1816 | case STATE_CREATING: |
| 1817 | CHECK(_rtmpsock); |
| 1818 | rc = bthread_id_create(&onfail_id, this, RunOnFailed); |
| 1819 | if (rc) { |
| 1820 | cntl->SetFailed(ENOMEM, "Fail to create _onfail_id: %s", berror(rc)); |
| 1821 | mu.unlock(); |
| 1822 | return OnFailedToCreateStream(); |
| 1823 | } |
| 1824 | // Add a ref for RunOnFailed. |
| 1825 | butil::intrusive_ptr<RtmpClientStream>(this).detach(); |
| 1826 | _state = STATE_CREATED; |
| 1827 | _onfail_id = onfail_id; |
| 1828 | break; |
| 1829 | case STATE_UNINITIALIZED: |
| 1830 | case STATE_CREATED: |
| 1831 | _state = STATE_ERROR; |
| 1832 | mu.unlock(); |
| 1833 | CHECK(false) << "Impossible"; |
| 1834 | return OnStopInternal(); |
| 1835 | case STATE_ERROR: |
| 1836 | case STATE_DESTROYING: |
| 1837 | mu.unlock(); |
| 1838 | return OnStopInternal(); |
| 1839 | } |
| 1840 | } |
| 1841 | if (onfail_id != INVALID_BTHREAD_ID) { |
| 1842 | _rtmpsock->NotifyOnFailed(onfail_id); |
no test coverage detected