| 2196 | } |
| 2197 | |
| 2198 | bool RtmpChunkStream::OnDataMessageAMF0( |
| 2199 | const RtmpMessageHeader& mh, butil::IOBuf* msg_body, Socket* socket) { |
| 2200 | butil::IOBufAsZeroCopyInputStream zc_stream(*msg_body); |
| 2201 | AMFInputStream istream(&zc_stream); |
| 2202 | std::string name; |
| 2203 | if (!ReadAMFString(&name, &istream)) { |
| 2204 | RTMP_ERROR(socket, mh) << "Fail to read name of DataMessage"; |
| 2205 | return false; |
| 2206 | } |
| 2207 | if (name == RTMP_AMF0_SET_DATAFRAME) { |
| 2208 | if (!ReadAMFString(&name, &istream)) { |
| 2209 | RTMP_ERROR(socket, mh) << "Fail to read name of DataMessage"; |
| 2210 | return false; |
| 2211 | } |
| 2212 | } |
| 2213 | RPC_VLOG << socket->remote_side() << "[" << mh.stream_id |
| 2214 | << "] DataMessage{timestamp=" << mh.timestamp |
| 2215 | << " name=" << name << '}'; |
| 2216 | |
| 2217 | if (name == RTMP_AMF0_ON_META_DATA || name == FLAGS_user_defined_data_message) { |
| 2218 | if (istream.check_emptiness()) { |
| 2219 | // Ignore empty metadata (seen in pulling streams from quanmin) |
| 2220 | return false; |
| 2221 | } |
| 2222 | RtmpMetaData metadata; |
| 2223 | metadata.timestamp = mh.timestamp; |
| 2224 | if (!ReadAMFObject(&metadata.data, &istream)) { |
| 2225 | RTMP_ERROR(socket, mh) << "Fail to read metadata"; |
| 2226 | return false; |
| 2227 | } |
| 2228 | // TODO: execq? |
| 2229 | butil::intrusive_ptr<RtmpStreamBase> stream; |
| 2230 | if (!connection_context()->FindMessageStream(mh.stream_id, &stream)) { |
| 2231 | LOG_EVERY_SECOND(WARNING) << socket->remote_side() |
| 2232 | << ": Fail to find stream_id=" << mh.stream_id; |
| 2233 | return false; |
| 2234 | } |
| 2235 | stream->CallOnMetaData(&metadata, name); |
| 2236 | return true; |
| 2237 | } else if (name == RTMP_AMF0_ON_CUE_POINT) { |
| 2238 | if (istream.check_emptiness()) { |
| 2239 | return false; |
| 2240 | } |
| 2241 | RtmpCuePoint cuepoint; |
| 2242 | cuepoint.timestamp = mh.timestamp; |
| 2243 | if (!ReadAMFObject(&cuepoint.data, &istream)) { |
| 2244 | RTMP_ERROR(socket, mh) << "Fail to read cuepoint"; |
| 2245 | return false; |
| 2246 | } |
| 2247 | // TODO: execq? |
| 2248 | butil::intrusive_ptr<RtmpStreamBase> stream; |
| 2249 | if (!connection_context()->FindMessageStream(mh.stream_id, &stream)) { |
| 2250 | LOG_EVERY_SECOND(WARNING) << socket->remote_side() |
| 2251 | << ": Fail to find stream_id=" << mh.stream_id; |
| 2252 | return false; |
| 2253 | } |
| 2254 | stream->CallOnCuePoint(&cuepoint); |
| 2255 | return true; |
nothing calls this directly
no test coverage detected