diff --git a/trunk/src/protocol/srs_kafka_stack.cpp b/trunk/src/protocol/srs_kafka_stack.cpp index 220dd595c..ed1c33228 100644 --- a/trunk/src/protocol/srs_kafka_stack.cpp +++ b/trunk/src/protocol/srs_kafka_stack.cpp @@ -95,56 +95,46 @@ int SrsKafkaString::nb_bytes() return _size == -1? 2 : 2 + _size; } -int SrsKafkaString::encode(SrsBuffer* buf) +srs_error_t SrsKafkaString::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(2)) { - ret = ERROR_KAFKA_CODEC_STRING; - srs_error("kafka encode string failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_STRING, "requires 2 only %d bytes", buf->left()); } buf->write_2bytes(_size); if (_size <= 0) { - return ret; + return err; } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_STRING; - srs_error("kafka encode string data failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_STRING, "requires %d only %d bytes", _size, buf->left()); } buf->write_bytes(data, _size); - return ret; + return err; } -int SrsKafkaString::decode(SrsBuffer* buf) +srs_error_t SrsKafkaString::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(2)) { - ret = ERROR_KAFKA_CODEC_STRING; - srs_error("kafka decode string failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_STRING, "requires 2 only %d bytes", buf->left()); } _size = buf->read_2bytes(); if (_size != -1 && _size < 0) { - ret = ERROR_KAFKA_CODEC_STRING; - srs_error("kafka string must be -1 or >=0, actual is %d. ret=%d", _size, ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_STRING, "invalid size=%d", _size); } if (_size <= 0) { - return ret; + return err; } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_STRING; - srs_error("kafka decode string data failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_STRING, "requires %d only %d bytes", _size, buf->left()); } srs_freepa(data); @@ -152,7 +142,7 @@ int SrsKafkaString::decode(SrsBuffer* buf) buf->read_bytes(data, _size); - return ret; + return err; } SrsKafkaBytes::SrsKafkaBytes() @@ -232,59 +222,49 @@ int SrsKafkaBytes::nb_bytes() return 4 + (_size == -1? 0 : _size); } -int SrsKafkaBytes::encode(SrsBuffer* buf) +srs_error_t SrsKafkaBytes::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_BYTES; - srs_error("kafka encode bytes failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_BYTES, "requires 4 only %d bytes", buf->left()); } buf->write_4bytes(_size); if (_size <= 0) { - return ret; + return err; } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_BYTES; - srs_error("kafka encode bytes data failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_BYTES, "requires %d only %d bytes", _size, buf->left()); } buf->write_bytes(_data, _size); - return ret; + return err; } -int SrsKafkaBytes::decode(SrsBuffer* buf) +srs_error_t SrsKafkaBytes::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_BYTES; - srs_error("kafka decode bytes failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_BYTES, "requires 4 only %d bytes", buf->left()); } _size = buf->read_4bytes(); if (_size != -1 && _size < 0) { - ret = ERROR_KAFKA_CODEC_BYTES; - srs_error("kafka bytes must be -1 or >=0, actual is %d. ret=%d", _size, ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_BYTES, "invalid size=%d", _size); } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_BYTES; - srs_error("kafka decode bytes data failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_BYTES, "requires %d only %d bytes", _size, buf->left()); } srs_freepa(_data); _data = new char[_size]; buf->read_bytes(_data, _size); - return ret; + return err; } SrsKafkaRequestHeader::SrsKafkaRequestHeader() @@ -380,14 +360,12 @@ int SrsKafkaRequestHeader::nb_bytes() return 4 + header_size(); } -int SrsKafkaRequestHeader::encode(SrsBuffer* buf) +srs_error_t SrsKafkaRequestHeader::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4 + _size)) { - ret = ERROR_KAFKA_CODEC_REQUEST; - srs_error("kafka encode request failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_REQUEST, "requires %d only %d bytes", 4 + _size, buf->left()); } buf->write_4bytes(_size); @@ -395,45 +373,39 @@ int SrsKafkaRequestHeader::encode(SrsBuffer* buf) buf->write_2bytes(api_version); buf->write_4bytes(_correlation_id); - if ((ret = client_id->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode request client_id failed. ret=%d", ret); - return ret; + if ((err = client_id->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode client_id"); } - return ret; + return err; } -int SrsKafkaRequestHeader::decode(SrsBuffer* buf) +srs_error_t SrsKafkaRequestHeader::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_REQUEST; - srs_error("kafka decode request size failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_REQUEST, "requires %d only %d bytes", 4, buf->left()); } _size = buf->read_4bytes(); if (_size <= 0) { srs_warn("kafka got empty request"); - return ret; + return err; } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_REQUEST; - srs_error("kafka decode request message failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_REQUEST, "requires %d only %d bytes", _size, buf->left()); } _api_key = buf->read_2bytes(); api_version = buf->read_2bytes(); _correlation_id = buf->read_4bytes(); - if ((ret = client_id->decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode request client_id failed. ret=%d", ret); - return ret; + if ((err = client_id->decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode client_id"); } - return ret; + return err; } SrsKafkaResponseHeader::SrsKafkaResponseHeader() @@ -476,46 +448,40 @@ int SrsKafkaResponseHeader::nb_bytes() return 4 + header_size(); } -int SrsKafkaResponseHeader::encode(SrsBuffer* buf) +srs_error_t SrsKafkaResponseHeader::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4 + _size)) { - ret = ERROR_KAFKA_CODEC_RESPONSE; - srs_error("kafka encode response failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_RESPONSE, "requires %d only %d bytes", 4 + _size, buf->left()); } buf->write_4bytes(_size); buf->write_4bytes(_correlation_id); - return ret; + return err; } -int SrsKafkaResponseHeader::decode(SrsBuffer* buf) +srs_error_t SrsKafkaResponseHeader::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_RESPONSE; - srs_error("kafka decode response size failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_RESPONSE, "requires %d only %d bytes", 4, buf->left()); } _size = buf->read_4bytes(); if (_size <= 0) { srs_warn("kafka got empty response"); - return ret; + return err; } if (!buf->require(_size)) { - ret = ERROR_KAFKA_CODEC_RESPONSE; - srs_error("kafka decode response message failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_RESPONSE, "requires %d only %d bytes", _size, buf->left()); } _correlation_id = buf->read_4bytes(); - return ret; + return err; } SrsKafkaRawMessage::SrsKafkaRawMessage() @@ -535,9 +501,9 @@ SrsKafkaRawMessage::~SrsKafkaRawMessage() srs_freep(value); } -int SrsKafkaRawMessage::create(SrsJsonObject* obj) +srs_error_t SrsKafkaRawMessage::create(SrsJsonObject* obj) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // current must be 0. magic_byte = 0; @@ -558,7 +524,7 @@ int SrsKafkaRawMessage::create(SrsJsonObject* obj) message_size = raw_message_size(); - return ret; + return err; } int SrsKafkaRawMessage::raw_message_size() @@ -571,14 +537,13 @@ int SrsKafkaRawMessage::nb_bytes() return 8 + 4 + 4 + 1 + 1 + key->nb_bytes() + value->nb_bytes(); } -int SrsKafkaRawMessage::encode(SrsBuffer* buf) +srs_error_t SrsKafkaRawMessage::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(8 + 4 + 4 + 1 + 1)) { - ret = ERROR_KAFKA_CODEC_MESSAGE; - srs_error("kafka encode message failed. ret=%d", ret); - return ret; + int nb_required = 8 + 4 + 4 + 1 + 1; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_MESSAGE, "requires %d only %d bytes", nb_required, buf->left()); } buf->write_8bytes(offset); buf->write_4bytes(message_size); @@ -586,27 +551,24 @@ int SrsKafkaRawMessage::encode(SrsBuffer* buf) buf->write_1bytes(magic_byte); buf->write_1bytes(attributes); - if ((ret = key->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode message key failed. ret=%d", ret); - return ret; + if ((err = key->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode key"); } - if ((ret = value->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode message value failed. ret=%d", ret); - return ret; + if ((err = value->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode value"); } - return ret; + return err; } -int SrsKafkaRawMessage::decode(SrsBuffer* buf) +srs_error_t SrsKafkaRawMessage::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(8 + 4 + 4 + 1 + 1)) { - ret = ERROR_KAFKA_CODEC_MESSAGE; - srs_error("kafka decode message failed. ret=%d", ret); - return ret; + int nb_required = 8 + 4 + 4 + 1 + 1; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_MESSAGE, "requires %d only %d bytes", nb_required, buf->left()); } offset = buf->read_8bytes(); message_size = buf->read_4bytes(); @@ -614,17 +576,15 @@ int SrsKafkaRawMessage::decode(SrsBuffer* buf) magic_byte = buf->read_1bytes(); attributes = buf->read_1bytes(); - if ((ret = key->decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode message key failed. ret=%d", ret); - return ret; + if ((err = key->decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode key"); } - if ((ret = value->decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode message value failed. ret=%d", ret); - return ret; + if ((err = value->decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode value"); } - return ret; + return err; } SrsKafkaRawMessageSet::SrsKafkaRawMessageSet() @@ -659,39 +619,37 @@ int SrsKafkaRawMessageSet::nb_bytes() return s; } -int SrsKafkaRawMessageSet::encode(SrsBuffer* buf) +srs_error_t SrsKafkaRawMessageSet::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; vector::iterator it; for (it = messages.begin(); it != messages.end(); ++it) { SrsKafkaRawMessage* message = *it; - if ((ret = message->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode message set failed. ret=%d", ret); - return ret; + if ((err = message->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode message"); } } - return ret; + return err; } -int SrsKafkaRawMessageSet::decode(SrsBuffer* buf) +srs_error_t SrsKafkaRawMessageSet::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; while (!buf->empty()) { SrsKafkaRawMessage* message = new SrsKafkaRawMessage(); - if ((ret = message->decode(buf)) != ERROR_SUCCESS) { + if ((err = message->decode(buf)) != srs_success) { srs_freep(message); - srs_error("kafka decode message set elem failed. ret=%d", ret); - return ret; + return srs_error_wrap(err, "decode message"); } messages.push_back(message); } - return ret; + return err; } SrsKafkaRequest::SrsKafkaRequest() @@ -723,12 +681,12 @@ int SrsKafkaRequest::nb_bytes() return header.nb_bytes(); } -int SrsKafkaRequest::encode(SrsBuffer* buf) +srs_error_t SrsKafkaRequest::encode(SrsBuffer* buf) { return header.encode(buf); } -int SrsKafkaRequest::decode(SrsBuffer* buf) +srs_error_t SrsKafkaRequest::decode(SrsBuffer* buf) { return header.decode(buf); } @@ -751,12 +709,12 @@ int SrsKafkaResponse::nb_bytes() return header.nb_bytes(); } -int SrsKafkaResponse::encode(SrsBuffer* buf) +srs_error_t SrsKafkaResponse::encode(SrsBuffer* buf) { return header.encode(buf); } -int SrsKafkaResponse::decode(SrsBuffer* buf) +srs_error_t SrsKafkaResponse::decode(SrsBuffer* buf) { return header.decode(buf); } @@ -780,38 +738,34 @@ int SrsKafkaTopicMetadataRequest::nb_bytes() return SrsKafkaRequest::nb_bytes() + topics.nb_bytes(); } -int SrsKafkaTopicMetadataRequest::encode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadataRequest::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaRequest::encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode metadata request failed. ret=%d", ret); - return ret; + if ((err = SrsKafkaRequest::encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode request"); } - if ((ret = topics.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode metadata topics failed. ret=%d", ret); - return ret; + if ((err = topics.encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode topics"); } - return ret; + return err; } -int SrsKafkaTopicMetadataRequest::decode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadataRequest::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaRequest::decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode metadata request failed. ret=%d", ret); - return ret; + if ((err = SrsKafkaRequest::decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode request"); } - if ((ret = topics.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode metadata topics failed. ret=%d", ret); - return ret; + if ((err = topics.decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode topics"); } - return ret; + return err; } SrsKafkaBroker::SrsKafkaBroker() @@ -828,56 +782,46 @@ int SrsKafkaBroker::nb_bytes() return 4 + host.nb_bytes() + 4; } -int SrsKafkaBroker::encode(SrsBuffer* buf) +srs_error_t SrsKafkaBroker::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka encode broker node_id failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 4, buf->left()); } buf->write_4bytes(node_id); - if ((ret = host.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode broker host failed. ret=%d", ret); - return ret; + if ((err = host.encode(buf)) != srs_success) { + return srs_error_wrap(err, "host"); } if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka encode broker port failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 4, buf->left()); } buf->write_4bytes(port); - return ret; + return err; } -int SrsKafkaBroker::decode(SrsBuffer* buf) +srs_error_t SrsKafkaBroker::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka decode broker node_id failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 4, buf->left()); } node_id = buf->read_4bytes(); - if ((ret = host.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode broker host failed. ret=%d", ret); - return ret; + if ((err = host.decode(buf)) != srs_success) { + return srs_error_wrap(err, "host"); } if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka decode broker port failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 4, buf->left()); } port = buf->read_4bytes(); - return ret; + return err; } SrsKafkaPartitionMetadata::SrsKafkaPartitionMetadata() @@ -896,54 +840,48 @@ int SrsKafkaPartitionMetadata::nb_bytes() return 2 + 4 + 4 + replicas.nb_bytes() + isr.nb_bytes(); } -int SrsKafkaPartitionMetadata::encode(SrsBuffer* buf) +srs_error_t SrsKafkaPartitionMetadata::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(2 + 4 + 4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka encode partition metadata failed. ret=%d", ret); - return ret; + int nb_required = 2 + 4 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", nb_required, buf->left()); } buf->write_2bytes(error_code); buf->write_4bytes(partition_id); buf->write_4bytes(leader); - if ((ret = replicas.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode partition metadata replicas failed. ret=%d", ret); - return ret; + if ((err = replicas.encode(buf)) != srs_success) { + return srs_error_wrap(err, "replicas"); } - if ((ret = isr.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode partition metadata isr failed. ret=%d", ret); - return ret; + if ((err = isr.encode(buf)) != srs_success) { + return srs_error_wrap(err, "isr"); } - return ret; + return err; } -int SrsKafkaPartitionMetadata::decode(SrsBuffer* buf) +srs_error_t SrsKafkaPartitionMetadata::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(2 + 4 + 4)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka decode partition metadata failed. ret=%d", ret); - return ret; + int nb_required = 2 + 4 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", nb_required, buf->left()); } error_code = buf->read_2bytes(); partition_id = buf->read_4bytes(); leader = buf->read_4bytes(); - if ((ret = replicas.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode partition metadata replicas failed. ret=%d", ret); - return ret; + if ((err = replicas.decode(buf)) != srs_success) { + return srs_error_wrap(err, "replicas"); } - if ((ret = isr.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode partition metadata isr failed. ret=%d", ret); - return ret; + if ((err = isr.decode(buf)) != srs_success) { + return srs_error_wrap(err, "isr"); } - return ret; + return err; } SrsKafkaTopicMetadata::SrsKafkaTopicMetadata() @@ -960,52 +898,44 @@ int SrsKafkaTopicMetadata::nb_bytes() return 2 + name.nb_bytes() + metadatas.nb_bytes(); } -int SrsKafkaTopicMetadata::encode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadata::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(2)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka encode topic metadata failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 2, buf->left()); } buf->write_2bytes(error_code); - if ((ret = name.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode topic name failed. ret=%d", ret); - return ret; + if ((err = name.encode(buf)) != srs_success) { + return srs_error_wrap(err, "name"); } - if ((ret = metadatas.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode topic metadatas failed. ret=%d", ret); - return ret; + if ((err = metadatas.encode(buf)) != srs_success) { + return srs_error_wrap(err, "metadatas"); } - return ret; + return err; } -int SrsKafkaTopicMetadata::decode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadata::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(2)) { - ret = ERROR_KAFKA_CODEC_METADATA; - srs_error("kafka decode topic metadata failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_METADATA, "requires %d only %d bytes", 2, buf->left()); } error_code = buf->read_2bytes(); - if ((ret = name.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode topic name failed. ret=%d", ret); - return ret; + if ((err = name.decode(buf)) != srs_success) { + return srs_error_wrap(err, "name"); } - if ((ret = metadatas.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode topic metadatas failed. ret=%d", ret); - return ret; + if ((err = metadatas.decode(buf)) != srs_success) { + return srs_error_wrap(err, "metadatas"); } - return ret; + return err; } SrsKafkaTopicMetadataResponse::SrsKafkaTopicMetadataResponse() @@ -1021,48 +951,42 @@ int SrsKafkaTopicMetadataResponse::nb_bytes() return SrsKafkaResponse::nb_bytes() + brokers.nb_bytes() + metadatas.nb_bytes(); } -int SrsKafkaTopicMetadataResponse::encode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadataResponse::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaResponse::encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode metadata response failed. ret=%d", ret); - return ret; + if ((err = SrsKafkaResponse::encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode response"); } - if ((ret = brokers.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode metadata brokers failed. ret=%d", ret); - return ret; + if ((err = brokers.encode(buf)) != srs_success) { + return srs_error_wrap(err, "brokers"); } - if ((ret = metadatas.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode metadatas failed. ret=%d", ret); - return ret; + if ((err = metadatas.encode(buf)) != srs_success) { + return srs_error_wrap(err, "metadatas"); } - return ret; + return err; } -int SrsKafkaTopicMetadataResponse::decode(SrsBuffer* buf) +srs_error_t SrsKafkaTopicMetadataResponse::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaResponse::decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode metadata response failed. ret=%d", ret); - return ret; + if ((err = SrsKafkaResponse::decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode response"); } - if ((ret = brokers.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode metadata brokers failed. ret=%d", ret); - return ret; + if ((err = brokers.decode(buf)) != srs_success) { + return srs_error_wrap(err, "brokers"); } - if ((ret = metadatas.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode metadatas failed. ret=%d", ret); - return ret; + if ((err = metadatas.decode(buf)) != srs_success) { + return srs_error_wrap(err, "metadatas"); } - return ret; + return err; } int SrsKafkaProducerPartitionMessages::nb_bytes() @@ -1070,34 +994,31 @@ int SrsKafkaProducerPartitionMessages::nb_bytes() return 4 + 4 + messages.nb_bytes(); } -int SrsKafkaProducerPartitionMessages::encode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerPartitionMessages::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(4 + 4)) { - ret = ERROR_KAFKA_CODEC_PRODUCER; - srs_error("kafka encode producer failed. ret=%d", ret); - return ret; + int nb_required = 4 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_PRODUCER, "requires %d only %d bytes", nb_required, buf->left()); } buf->write_4bytes(partition); buf->write_4bytes(message_set_size); - if ((ret = messages.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode producer messages failed. ret=%d", ret); - return ret; + if ((err = messages.encode(buf)) != srs_success) { + return srs_error_wrap(err, "messages"); } - return ret; + return err; } -int SrsKafkaProducerPartitionMessages::decode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerPartitionMessages::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(4 + 4)) { - ret = ERROR_KAFKA_CODEC_PRODUCER; - srs_error("kafka decode producer failed. ret=%d", ret); - return ret; + int nb_required = 4 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_PRODUCER, "requires %d only %d bytes", nb_required, buf->left()); } partition = buf->read_4bytes(); message_set_size = buf->read_4bytes(); @@ -1105,25 +1026,19 @@ int SrsKafkaProducerPartitionMessages::decode(SrsBuffer* buf) // for the message set decode util empty, we must create a new buffer when // there exists other objects after message set. if (buf->left() != message_set_size) { - SrsBuffer* tbuf = new SrsBuffer(); + SrsBuffer* tbuf = new SrsBuffer(buf->data() + buf->pos(), message_set_size); SrsAutoFree(SrsBuffer, tbuf); - if ((ret = tbuf->initialize(buf->data() + buf->pos(), message_set_size)) != ERROR_SUCCESS) { - return ret; - } - - if ((ret = messages.decode(tbuf)) != ERROR_SUCCESS) { - srs_error("kafka decode procuder messages failed. ret=%d", ret); - return ret; + if ((err = messages.decode(buf)) != srs_success) { + return srs_error_wrap(err, "messages"); } } else { - if ((ret = messages.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode procuder messages failed. ret=%d", ret); - return ret; + if ((err = messages.decode(buf)) != srs_success) { + return srs_error_wrap(err, "messages"); } } - return ret; + return err; } int SrsKafkaProducerTopicMessages::nb_bytes() @@ -1131,34 +1046,34 @@ int SrsKafkaProducerTopicMessages::nb_bytes() return topic_name.nb_bytes() + partitions.nb_bytes(); } -int SrsKafkaProducerTopicMessages::encode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerTopicMessages::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = topic_name.encode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = topic_name.encode(buf)) != srs_success) { + return srs_error_wrap(err, "topic_name"); } - if ((ret = partitions.encode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = partitions.encode(buf)) != srs_success) { + return srs_error_wrap(err, "partitions"); } - return ret; + return err; } -int SrsKafkaProducerTopicMessages::decode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerTopicMessages::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = topic_name.decode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = topic_name.decode(buf)) != srs_success) { + return srs_error_wrap(err, "topic_name"); } - if ((ret = partitions.decode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = partitions.decode(buf)) != srs_success) { + return srs_error_wrap(err, "partitions"); } - return ret; + return err; } SrsKafkaProducerRequest::SrsKafkaProducerRequest() @@ -1176,50 +1091,48 @@ int SrsKafkaProducerRequest::nb_bytes() return SrsKafkaRequest::nb_bytes() + 2 + 4 + topics.nb_bytes(); } -int SrsKafkaProducerRequest::encode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerRequest::encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaRequest::encode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = SrsKafkaRequest::encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode request"); } - if (!buf->require(2 + 4)) { - ret = ERROR_KAFKA_CODEC_PRODUCER; - srs_error("kafka encode producer failed. ret=%d", ret); - return ret; + int nb_required = 2 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_PRODUCER, "requires %d only %d bytes", nb_required, buf->left()); } buf->write_2bytes(required_acks); buf->write_4bytes(timeout); - if ((ret = topics.encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode producer topics failed. ret=%d", ret); + if ((err = topics.encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode topics"); } - return ret; + return err; } -int SrsKafkaProducerRequest::decode(SrsBuffer* buf) +srs_error_t SrsKafkaProducerRequest::decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if ((ret = SrsKafkaRequest::decode(buf)) != ERROR_SUCCESS) { - return ret; + if ((err = SrsKafkaRequest::decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode request"); } - if (!buf->require(2 + 4)) { - ret = ERROR_KAFKA_CODEC_PRODUCER; - srs_error("kafka decode producer failed. ret=%d", ret); - return ret; + int nb_required = 2 + 4; + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_PRODUCER, "requires %d only %d bytes", nb_required, buf->left()); } required_acks = buf->read_2bytes(); timeout = buf->read_4bytes(); - if ((ret = topics.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode producer topics failed. ret=%d", ret); + if ((err = topics.decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode topics"); } - return ret; + return err; } SrsKafkaCorrelationPool* SrsKafkaCorrelationPool::_instance = new SrsKafkaCorrelationPool(); @@ -1291,16 +1204,16 @@ SrsKafkaProtocol::~SrsKafkaProtocol() srs_freep(reader); } -int SrsKafkaProtocol::send_and_free_message(SrsKafkaRequest* msg) +srs_error_t SrsKafkaProtocol::send_and_free_message(SrsKafkaRequest* msg) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // TODO: FIXME: refine for performance issue. SrsAutoFree(SrsKafkaRequest, msg); int size = msg->nb_bytes(); if (size <= 0) { - return ret; + return err; } // update the header of message. @@ -1315,52 +1228,40 @@ int SrsKafkaProtocol::send_and_free_message(SrsKafkaRequest* msg) SrsAutoFreeA(char, bytes); // TODO: FIXME: refine for performance issue. - SrsBuffer* buf = new SrsBuffer(); + SrsBuffer* buf = new SrsBuffer(bytes, size); SrsAutoFree(SrsBuffer, buf); - if ((ret = buf->initialize(bytes, size)) != ERROR_SUCCESS) { - srs_error("kafka create buffer failed. ret=%d", ret); - return ret; + if ((err = msg->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode msg"); } - if ((ret = msg->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode message failed. ret=%d", ret); - return ret; + if ((err = skt->write(bytes, size, NULL)) != srs_success) { + return srs_error_wrap(err, "write msg"); } - if ((ret = skt->write(bytes, size, NULL)) != ERROR_SUCCESS) { - srs_error("kafka send message failed. ret=%d", ret); - return ret; - } - - return ret; + return err; } -int SrsKafkaProtocol::recv_message(SrsKafkaResponse** pmsg) +srs_error_t SrsKafkaProtocol::recv_message(SrsKafkaResponse** pmsg) { *pmsg = NULL; - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; while (true) { SrsKafkaResponseHeader header; // ensure enough bytes for response header. - if ((ret = reader->grow(skt, header.nb_bytes())) != ERROR_SUCCESS) { - srs_error("kafka recv message failed. ret=%d", ret); - return ret; + if ((err = reader->grow(skt, header.nb_bytes())) != srs_success) { + return srs_error_wrap(err, "grow buffer"); } // decode response header. - SrsBuffer buffer; - if ((ret = buffer.initialize(reader->bytes(), reader->size())) != ERROR_SUCCESS) { - return ret; - } + SrsBuffer* buf = new SrsBuffer(reader->bytes(), reader->size()); + SrsAutoFree(SrsBuffer, buf); - SrsBuffer* buf = &buffer; - if ((ret = header.decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode response header failed. ret=%d", ret); - return ret; + if ((err = header.decode(buf)) != srs_success) { + return srs_error_wrap(err, "decode header"); } // skip the used buffer for header. @@ -1384,10 +1285,9 @@ int SrsKafkaProtocol::recv_message(SrsKafkaResponse** pmsg) } // ensure enough bytes to decode message. - if ((ret = reader->grow(skt, header.total_size())) != ERROR_SUCCESS) { + if ((err = reader->grow(skt, header.total_size())) != srs_success) { srs_freep(res); - srs_error("kafka recv message body failed. ret=%d", ret); - return ret; + return srs_error_wrap(err, "grow buffer"); } // dropped message, fetch next. @@ -1398,17 +1298,16 @@ int SrsKafkaProtocol::recv_message(SrsKafkaResponse** pmsg) } // parse the whole message. - if ((ret = res->decode(buf)) != ERROR_SUCCESS) { + if ((err = res->decode(buf)) != srs_success) { srs_freep(res); - srs_error("kafka decode message failed. ret=%d", ret); - return ret; + return srs_error_wrap(err, "decode response"); } *pmsg = res; break; } - return ret; + return err; } SrsKafkaClient::SrsKafkaClient(ISrsProtocolReaderWriter* io) @@ -1421,32 +1320,30 @@ SrsKafkaClient::~SrsKafkaClient() srs_freep(protocol); } -int SrsKafkaClient::fetch_metadata(string topic, SrsKafkaTopicMetadataResponse** pmsg) +srs_error_t SrsKafkaClient::fetch_metadata(string topic, SrsKafkaTopicMetadataResponse** pmsg) { *pmsg = NULL; - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsKafkaTopicMetadataRequest* req = new SrsKafkaTopicMetadataRequest(); req->add_topic(topic); - if ((ret = protocol->send_and_free_message(req)) != ERROR_SUCCESS) { - srs_error("kafka send message failed. ret=%d", ret); - return ret; + if ((err = protocol->send_and_free_message(req)) != srs_success) { + return srs_error_wrap(err, "send request"); } - if ((ret = protocol->expect_message(pmsg)) != ERROR_SUCCESS) { - srs_error("kafka recv response failed. ret=%d", ret); - return ret; + if ((err = protocol->expect_message(pmsg)) != srs_success) { + return srs_error_wrap(err, "expect message"); } - return ret; + return err; } -int SrsKafkaClient::write_messages(std::string topic, int32_t partition, vector& msgs) +srs_error_t SrsKafkaClient::write_messages(std::string topic, int32_t partition, vector& msgs) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsKafkaProducerRequest* req = new SrsKafkaProducerRequest(); @@ -1471,11 +1368,10 @@ int SrsKafkaClient::write_messages(std::string topic, int32_t partition, vector< SrsJsonObject* obj = *it; SrsKafkaRawMessage* msg = new SrsKafkaRawMessage(); - if ((ret = msg->create(obj)) != ERROR_SUCCESS) { + if ((err = msg->create(obj)) != srs_success) { srs_freep(msg); srs_freep(req); - srs_error("kafka write messages failed. ret=%d", ret); - return ret; + return srs_error_wrap(err, "create message"); } partitions->messages.append(msg); @@ -1484,12 +1380,11 @@ int SrsKafkaClient::write_messages(std::string topic, int32_t partition, vector< partitions->message_set_size = partitions->messages.nb_bytes(); // write to kafka cluster. - if ((ret = protocol->send_and_free_message(req)) != ERROR_SUCCESS) { - srs_error("kafka write producer message failed. ret=%d", ret); - return ret; + if ((err = protocol->send_and_free_message(req)) != srs_success) { + return srs_error_wrap(err, "send request"); } - return ret; + return err; } vector srs_kafka_array2vector(SrsKafkaArray* arr) diff --git a/trunk/src/protocol/srs_kafka_stack.hpp b/trunk/src/protocol/srs_kafka_stack.hpp index 4198665f2..70658e192 100644 --- a/trunk/src/protocol/srs_kafka_stack.hpp +++ b/trunk/src/protocol/srs_kafka_stack.hpp @@ -80,8 +80,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -109,8 +109,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -177,50 +177,44 @@ public: return s; } - virtual int encode(SrsBuffer* buf) + virtual srs_error_t encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_ARRAY; - srs_error("kafka encode array failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_ARRAY, "requires 4 only %d bytes", buf->left()); } buf->write_4bytes(length); for (SrsIterator it = elems.begin(); it != elems.end(); ++it) { T* elem = *it; - if ((ret = elem->encode(buf)) != ERROR_SUCCESS) { - srs_error("kafka encode array elem failed. ret=%d", ret); - return ret; + if ((err = elem->encode(buf)) != srs_success) { + return srs_error_wrap(err, "encode elem"); } } - return ret; + return err; } - virtual int decode(SrsBuffer* buf) + virtual srs_error_t decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_ARRAY; - srs_error("kafka decode array failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_ARRAY, "requires 4 only %d bytes", buf->left()); } length = buf->read_4bytes(); for (int i = 0; i < length; i++) { T* elem = new T(); - if ((ret = elem->decode(buf)) != ERROR_SUCCESS) { - srs_error("kafka decode array elem failed. ret=%d", ret); + if ((err = elem->decode(buf)) != srs_success) { srs_freep(elem); - return ret; + return srs_error_wrap(err, "decode elem"); } elems.push_back(elem); } - return ret; + return err; } }; template<> @@ -263,14 +257,13 @@ public: { return 4 + 4 * (int)elems.size(); } - virtual int encode(SrsBuffer* buf) + virtual srs_error_t encode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; - if (!buf->require(4 + sizeof(int32_t) * (int)elems.size())) { - ret = ERROR_KAFKA_CODEC_ARRAY; - srs_error("kafka encode array failed. ret=%d", ret); - return ret; + int nb_required = 4 + sizeof(int32_t) * (int)elems.size(); + if (!buf->require(nb_required)) { + return srs_error_new(ERROR_KAFKA_CODEC_ARRAY, "requires %d only %d bytes", nb_required, buf->left()); } buf->write_4bytes(length); @@ -279,24 +272,20 @@ public: buf->write_4bytes(elem); } - return ret; + return err; } - virtual int decode(SrsBuffer* buf) + virtual srs_error_t decode(SrsBuffer* buf) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (!buf->require(4)) { - ret = ERROR_KAFKA_CODEC_ARRAY; - srs_error("kafka decode array failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_ARRAY, "requires 4 only %d bytes", buf->left()); } length = buf->read_4bytes(); for (int i = 0; i < length; i++) { if (!buf->require(sizeof(int32_t))) { - ret = ERROR_KAFKA_CODEC_ARRAY; - srs_error("kafka decode array elem failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_KAFKA_CODEC_ARRAY, "requires %d only %d bytes", sizeof(int32_t), buf->left()); } @@ -304,7 +293,7 @@ public: elems.push_back(elem); } - return ret; + return err; } }; @@ -414,8 +403,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -480,8 +469,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -538,7 +527,7 @@ public: /** * create message from json object. */ - virtual int create(SrsJsonObject* obj); + virtual srs_error_t create(SrsJsonObject* obj); private: /** * get the raw message, bytes after the message_size. @@ -547,8 +536,8 @@ private: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -568,8 +557,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -599,8 +588,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -622,8 +611,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -652,8 +641,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -672,8 +661,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; struct SrsKafkaPartitionMetadata : public ISrsCodec { @@ -689,8 +678,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; struct SrsKafkaTopicMetadata : public ISrsCodec { @@ -704,8 +693,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -727,8 +716,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; @@ -754,8 +743,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; struct SrsKafkaProducerTopicMessages : public ISrsCodec { @@ -771,8 +760,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -815,8 +804,8 @@ public: // interface ISrsCodec public: virtual int nb_bytes(); - virtual int encode(SrsBuffer* buf); - virtual int decode(SrsBuffer* buf); + virtual srs_error_t encode(SrsBuffer* buf); + virtual srs_error_t decode(SrsBuffer* buf); }; /** @@ -875,26 +864,25 @@ public: * write the message to kafka server. * @param msg the msg to send. user must not free it again. */ - virtual int send_and_free_message(SrsKafkaRequest* msg); + virtual srs_error_t send_and_free_message(SrsKafkaRequest* msg); /** * read the message from kafka server. * @param pmsg output the received message. user must free it. */ - virtual int recv_message(SrsKafkaResponse** pmsg); + virtual srs_error_t recv_message(SrsKafkaResponse** pmsg); public: /** * expect specified message. */ template - int expect_message(T** pmsg) + srs_error_t expect_message(T** pmsg) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; while (true) { SrsKafkaResponse* res = NULL; - if ((ret = recv_message(&res)) != ERROR_SUCCESS) { - srs_error("recv response failed. ret=%d", ret); - return ret; + if ((err = recv_message(&res)) != srs_success) { + return srs_error_wrap(err, "recv message"); } // drop not matched. @@ -909,7 +897,7 @@ public: break; } - return ret; + return err; } }; @@ -927,11 +915,11 @@ public: /** * fetch the metadata from broker for topic. */ - virtual int fetch_metadata(std::string topic, SrsKafkaTopicMetadataResponse** pmsg); + virtual srs_error_t fetch_metadata(std::string topic, SrsKafkaTopicMetadataResponse** pmsg); /** * write the messages to partition of topic. */ - virtual int write_messages(std::string topic, int32_t partition, std::vector& msgs); + virtual srs_error_t write_messages(std::string topic, int32_t partition, std::vector& msgs); }; // convert kafka array[string] to vector[string] diff --git a/trunk/src/protocol/srs_raw_avc.cpp b/trunk/src/protocol/srs_raw_avc.cpp index d46b1d8c6..08ee4c584 100644 --- a/trunk/src/protocol/srs_raw_avc.cpp +++ b/trunk/src/protocol/srs_raw_avc.cpp @@ -41,9 +41,9 @@ SrsRawH264Stream::~SrsRawH264Stream() { } -int SrsRawH264Stream::annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_frame) +srs_error_t SrsRawH264Stream::annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_frame) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; *pframe = NULL; *pnb_frame = 0; @@ -53,7 +53,7 @@ int SrsRawH264Stream::annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_fr // about annexb, @see ISO_IEC_14496-10-AVC-2003.pdf, page 211. int pnb_start_code = 0; if (!srs_avc_startswith_annexb(stream, &pnb_start_code)) { - return ERROR_H264_API_NO_PREFIXED; + return srs_error_new(ERROR_H264_API_NO_PREFIXED, "annexb start code"); } int start = stream->pos() + pnb_start_code; @@ -72,7 +72,7 @@ int SrsRawH264Stream::annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_fr break; } - return ret; + return err; } bool SrsRawH264Stream::is_sps(char* frame, int nb_frame) @@ -99,13 +99,13 @@ bool SrsRawH264Stream::is_pps(char* frame, int nb_frame) return nal_unit_type == 8; } -int SrsRawH264Stream::sps_demux(char* frame, int nb_frame, string& sps) +srs_error_t SrsRawH264Stream::sps_demux(char* frame, int nb_frame, string& sps) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // atleast 1bytes for SPS to decode the type, profile, constrain and level. if (nb_frame < 4) { - return ret; + return err; } sps = ""; @@ -115,15 +115,15 @@ int SrsRawH264Stream::sps_demux(char* frame, int nb_frame, string& sps) // should never be empty. if (sps.empty()) { - return ERROR_STREAM_CASTER_AVC_SPS; + return srs_error_new(ERROR_STREAM_CASTER_AVC_SPS, "no sps"); } - return ret; + return err; } -int SrsRawH264Stream::pps_demux(char* frame, int nb_frame, string& pps) +srs_error_t SrsRawH264Stream::pps_demux(char* frame, int nb_frame, string& pps) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; pps = ""; if (nb_frame > 0) { @@ -132,15 +132,15 @@ int SrsRawH264Stream::pps_demux(char* frame, int nb_frame, string& pps) // should never be empty. if (pps.empty()) { - return ERROR_STREAM_CASTER_AVC_PPS; + return srs_error_new(ERROR_STREAM_CASTER_AVC_PPS, "no pps"); } - return ret; + return err; } -int SrsRawH264Stream::mux_sequence_header(string sps, string pps, uint32_t dts, uint32_t pts, string& sh) +srs_error_t SrsRawH264Stream::mux_sequence_header(string sps, string pps, uint32_t dts, uint32_t pts, string& sh) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // 5bytes sps/pps header: // configurationVersion, AVCProfileIndication, profile_compatibility, @@ -160,10 +160,7 @@ int SrsRawH264Stream::mux_sequence_header(string sps, string pps, uint32_t dts, SrsAutoFreeA(char, packet); // use stream to generate the h264 packet. - SrsBuffer stream; - if ((ret = stream.initialize(packet, nb_packet)) != ERROR_SUCCESS) { - return ret; - } + SrsBuffer stream(packet, nb_packet); // decode the SPS: // @see: 7.3.2.1.1, ISO_IEC_14496-10-AVC-2012.pdf, page 62 @@ -223,12 +220,12 @@ int SrsRawH264Stream::mux_sequence_header(string sps, string pps, uint32_t dts, sh = ""; sh.append(packet, nb_packet); - return ret; + return err; } -int SrsRawH264Stream::mux_ipb_frame(char* frame, int nb_frame, string& ibp) +srs_error_t SrsRawH264Stream::mux_ipb_frame(char* frame, int nb_frame, string& ibp) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // 4bytes size of nalu: // NALUnitLength @@ -239,10 +236,7 @@ int SrsRawH264Stream::mux_ipb_frame(char* frame, int nb_frame, string& ibp) SrsAutoFreeA(char, packet); // use stream to generate the h264 packet. - SrsBuffer stream; - if ((ret = stream.initialize(packet, nb_packet)) != ERROR_SUCCESS) { - return ret; - } + SrsBuffer stream(packet, nb_packet); // 5.3.4.2.1 Syntax, ISO_IEC_14496-15-AVC-format-2012.pdf, page 16 // lengthSizeMinusOne, or NAL_unit_length, always use 4bytes size @@ -258,12 +252,12 @@ int SrsRawH264Stream::mux_ipb_frame(char* frame, int nb_frame, string& ibp) ibp = ""; ibp.append(packet, nb_packet); - return ret; + return err; } -int SrsRawH264Stream::mux_avc2flv(string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char** flv, int* nb_flv) +srs_error_t SrsRawH264Stream::mux_avc2flv(string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char** flv, int* nb_flv) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // for h264 in RTMP video payload, there is 5bytes header: // 1bytes, FrameType | CodecID @@ -299,7 +293,7 @@ int SrsRawH264Stream::mux_avc2flv(string video, int8_t frame_type, int8_t avc_pa *flv = data; *nb_flv = size; - return ret; + return err; } SrsRawAacStream::SrsRawAacStream() @@ -310,9 +304,9 @@ SrsRawAacStream::~SrsRawAacStream() { } -int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame, SrsRawAacStreamCodec& codec) +srs_error_t SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame, SrsRawAacStreamCodec& codec) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; while (!stream->empty()) { int adts_header_start = stream->pos(); @@ -337,12 +331,12 @@ int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame // else // require(9bytes)=72bits if (!stream->require(7)) { - return ERROR_AAC_ADTS_HEADER; + return srs_error_new(ERROR_AAC_ADTS_HEADER, "requires 7 only %d bytes", stream->left()); } // for aac, the frame must be ADTS format. if (!srs_aac_startswith_adts(stream)) { - return ERROR_AAC_REQUIRED_ADTS; + return srs_error_new(ERROR_AAC_REQUIRED_ADTS, "not adts"); } // syncword 12 bslbf @@ -362,8 +356,6 @@ int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame * and set to '0' if the audio data are MPEG-4. See also ISO/IEC 11172-3, subclause 2.4.2.3. */ if (id != 0x01) { - srs_info("adts: id must be 1(aac), actual 0(mp4a). ret=%d", ret); - // well, some system always use 0, but actually is aac format. // for example, houjian vod ts always set the aac id to 0, actually 1. // we just ignore it, and alwyas use 1(aac) to demux. @@ -406,7 +398,7 @@ int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame // adts_error_check(), 1.A.2.2.3 Error detection if (!protection_absent) { if (!stream->require(2)) { - return ERROR_AAC_ADTS_HEADER; + return srs_error_new(ERROR_AAC_ADTS_HEADER, "requires 2 only %d bytes", stream->left()); } // crc_check 16 Rpchof /*int16_t crc_check = */stream->read_2bytes(); @@ -419,7 +411,7 @@ int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame int adts_header_size = stream->pos() - adts_header_start; int raw_data_size = frame_length - adts_header_size; if (!stream->require(raw_data_size)) { - return ERROR_AAC_ADTS_HEADER; + return srs_error_new(ERROR_AAC_ADTS_HEADER, "requires %d only %d bytes", raw_data_size, stream->left()); } // the codec info. @@ -456,16 +448,16 @@ int SrsRawAacStream::adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame break; } - return ret; + return err; } -int SrsRawAacStream::mux_sequence_header(SrsRawAacStreamCodec* codec, string& sh) +srs_error_t SrsRawAacStream::mux_sequence_header(SrsRawAacStreamCodec* codec, string& sh) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // only support aac profile 1-4. if (codec->aac_object == SrsAacObjectTypeReserved) { - return ERROR_AAC_DATA_INVALID; + return srs_error_new(ERROR_AAC_DATA_INVALID, "invalid aac object"); } SrsAacObjectType audioObjectType = codec->aac_object; @@ -500,7 +492,7 @@ int SrsRawAacStream::mux_sequence_header(SrsRawAacStreamCodec* codec, string& sh sh += ch; ch = (samplingFrequencyIndex << 7) & 0x80; if (samplingFrequencyIndex == 0x0f) { - return ERROR_AAC_DATA_INVALID; + return srs_error_new(ERROR_AAC_DATA_INVALID, "invalid sampling frequency index"); } // 7bits left. @@ -515,12 +507,12 @@ int SrsRawAacStream::mux_sequence_header(SrsRawAacStreamCodec* codec, string& sh // extensionFlag; 1 bslbf sh += ch; - return ret; + return err; } -int SrsRawAacStream::mux_aac2flv(char* frame, int nb_frame, SrsRawAacStreamCodec* codec, uint32_t dts, char** flv, int* nb_flv) +srs_error_t SrsRawAacStream::mux_aac2flv(char* frame, int nb_frame, SrsRawAacStreamCodec* codec, uint32_t dts, char** flv, int* nb_flv) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; char sound_format = codec->sound_format; char sound_type = codec->sound_type; @@ -554,6 +546,6 @@ int SrsRawAacStream::mux_aac2flv(char* frame, int nb_frame, SrsRawAacStreamCodec *flv = data; *nb_flv = size; - return ret; + return err; } diff --git a/trunk/src/protocol/srs_raw_avc.hpp b/trunk/src/protocol/srs_raw_avc.hpp index 6fbb7f838..f5279121a 100644 --- a/trunk/src/protocol/srs_raw_avc.hpp +++ b/trunk/src/protocol/srs_raw_avc.hpp @@ -47,7 +47,7 @@ public: * @param pframe the output h.264 frame in stream. user should never free it. * @param pnb_frame the output h.264 frame size. */ - virtual int annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_frame); + virtual srs_error_t annexb_demux(SrsBuffer* stream, char** pframe, int* pnb_frame); /** * whether the frame is sps or pps. */ @@ -57,22 +57,22 @@ public: * demux the sps or pps to string. * @param sps/pps output the sps/pps. */ - virtual int sps_demux(char* frame, int nb_frame, std::string& sps); - virtual int pps_demux(char* frame, int nb_frame, std::string& pps); + virtual srs_error_t sps_demux(char* frame, int nb_frame, std::string& sps); + virtual srs_error_t pps_demux(char* frame, int nb_frame, std::string& pps); public: /** * h264 raw data to h264 packet, without flv payload header. * mux the sps/pps to flv sequence header packet. * @param sh output the sequence header. */ - virtual int mux_sequence_header(std::string sps, std::string pps, uint32_t dts, uint32_t pts, std::string& sh); + virtual srs_error_t mux_sequence_header(std::string sps, std::string pps, uint32_t dts, uint32_t pts, std::string& sh); /** * h264 raw data to h264 packet, without flv payload header. * mux the ibp to flv ibp packet. * @param ibp output the packet. * @param frame_type output the frame type. */ - virtual int mux_ipb_frame(char* frame, int nb_frame, std::string& ibp); + virtual srs_error_t mux_ipb_frame(char* frame, int nb_frame, std::string& ibp); /** * mux the avc video packet to flv video packet. * @param frame_type, SrsVideoAvcFrameTypeKeyFrame or SrsVideoAvcFrameTypeInterFrame. @@ -81,7 +81,7 @@ public: * @param flv output the muxed flv packet. * @param nb_flv output the muxed flv size. */ - virtual int mux_avc2flv(std::string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char** flv, int* nb_flv); + virtual srs_error_t mux_avc2flv(std::string video, int8_t frame_type, int8_t avc_packet_type, uint32_t dts, uint32_t pts, char** flv, int* nb_flv); }; /** @@ -119,13 +119,13 @@ public: * @param pnb_frame the output aac frame size. * @param codec the output codec info. */ - virtual int adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame, SrsRawAacStreamCodec& codec); + virtual srs_error_t adts_demux(SrsBuffer* stream, char** pframe, int* pnb_frame, SrsRawAacStreamCodec& codec); /** * aac raw data to aac packet, without flv payload header. * mux the aac specific config to flv sequence header packet. * @param sh output the sequence header. */ - virtual int mux_sequence_header(SrsRawAacStreamCodec* codec, std::string& sh); + virtual srs_error_t mux_sequence_header(SrsRawAacStreamCodec* codec, std::string& sh); /** * mux the aac audio packet to flv audio packet. * @param frame the aac raw data. @@ -134,7 +134,7 @@ public: * @param flv output the muxed flv packet. * @param nb_flv output the muxed flv size. */ - virtual int mux_aac2flv(char* frame, int nb_frame, SrsRawAacStreamCodec* codec, uint32_t dts, char** flv, int* nb_flv); + virtual srs_error_t mux_aac2flv(char* frame, int nb_frame, SrsRawAacStreamCodec* codec, uint32_t dts, char** flv, int* nb_flv); }; #endif diff --git a/trunk/src/protocol/srs_rtsp_stack.cpp b/trunk/src/protocol/srs_rtsp_stack.cpp index 0b1a0d722..422233d00 100644 --- a/trunk/src/protocol/srs_rtsp_stack.cpp +++ b/trunk/src/protocol/srs_rtsp_stack.cpp @@ -179,15 +179,13 @@ void SrsRtpPacket::reap(SrsRtpPacket* src) src->audio = NULL; } -int SrsRtpPacket::decode(SrsBuffer* stream) +srs_error_t SrsRtpPacket::decode(SrsBuffer* stream) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // 12bytes header if (!stream->require(12)) { - ret = ERROR_RTP_HEADER_CORRUPT; - srs_error("rtsp: rtp header corrupt. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTP_HEADER_CORRUPT, "requires 12 only %d bytes", stream->left()); } int8_t vv = stream->read_1bytes(); @@ -213,19 +211,16 @@ int SrsRtpPacket::decode(SrsBuffer* stream) return decode_97(stream); } - return ret; + return err; } -int SrsRtpPacket::decode_97(SrsBuffer* stream) +srs_error_t SrsRtpPacket::decode_97(SrsBuffer* stream) { - int ret = ERROR_SUCCESS; srs_error_t err = srs_success; // atleast 2bytes content. if (!stream->require(2)) { - ret = ERROR_RTP_TYPE97_CORRUPT; - srs_error("rtsp: rtp type97 corrupt. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTP_TYPE97_CORRUPT, "requires 2 only %d bytes", stream->left()); } int8_t hasv = stream->read_1bytes(); @@ -233,9 +228,7 @@ int SrsRtpPacket::decode_97(SrsBuffer* stream) uint16_t au_size = ((hasv << 5) & 0xE0) | ((lasv >> 3) & 0x1f); if (!stream->require(au_size)) { - ret = ERROR_RTP_TYPE97_CORRUPT; - srs_error("rtsp: rtp type97 au_size corrupt. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTP_TYPE97_CORRUPT, "requires %d only %d bytes", au_size, stream->left()); } int required_size = 0; @@ -261,35 +254,28 @@ int SrsRtpPacket::decode_97(SrsBuffer* stream) required_size += sample_size; if (!stream->require(required_size)) { - ret = ERROR_RTP_TYPE97_CORRUPT; - srs_error("rtsp: rtp type97 samples corrupt. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTP_TYPE97_CORRUPT, "requires %d only %d bytes", required_size, stream->left()); } if ((err = audio->add_sample(sample, sample_size)) != srs_success) { - // TODO: FIXME: Use error - ret = srs_error_code(err); srs_freep(err); - srs_error("rtsp: rtp type97 add sample failed. ret=%d", ret); - return ret; + return srs_error_wrap(err, "add sample"); } } // parsed ok. completed = true; - return ret; + return err; } -int SrsRtpPacket::decode_96(SrsBuffer* stream) +srs_error_t SrsRtpPacket::decode_96(SrsBuffer* stream) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // atleast 2bytes content. if (!stream->require(2)) { - ret = ERROR_RTP_TYPE96_CORRUPT; - srs_error("rtsp: rtp type96 corrupt. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTP_TYPE96_CORRUPT, "requires 2 only %d bytes", stream->left()); } // frame type @@ -324,7 +310,7 @@ int SrsRtpPacket::decode_96(SrsBuffer* stream) } payload->append(stream->data() + stream->pos(), stream->size() - stream->pos()); - return ret; + return err; } // no chunked, append to payload. @@ -332,7 +318,7 @@ int SrsRtpPacket::decode_96(SrsBuffer* stream) payload->append(stream->data() + stream->pos(), stream->size() - stream->pos()); completed = true; - return ret; + return err; } SrsRtspSdp::SrsRtspSdp() @@ -344,13 +330,13 @@ SrsRtspSdp::~SrsRtspSdp() { } -int SrsRtspSdp::parse(string token) +srs_error_t SrsRtspSdp::parse(string token) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; if (token.empty()) { - srs_info("rtsp: ignore empty token."); - return ret; + // ignore empty token + return err; } size_t pos = string::npos; @@ -456,17 +442,15 @@ int SrsRtspSdp::parse(string token) } else if (desc_key == "fmtp") { for (int i = 1; i < (int)attrs.size(); i++) { std::string attr = attrs.at(i); - if ((ret = parse_fmtp_attribute(attr)) != ERROR_SUCCESS) { - srs_error("rtsp: parse fmtp failed, attr=%s. ret=%d", attr.c_str(), ret); - return ret; + if ((err = parse_fmtp_attribute(attr)) != srs_success) { + return srs_error_wrap(err, "parse fmtp attr=%s", attr.c_str()); } } } else if (desc_key == "control") { for (int i = 0; i < (int)attrs.size(); i++) { std::string attr = attrs.at(i); - if ((ret = parse_control_attribute(attr)) != ERROR_SUCCESS) { - srs_error("rtsp: parse control failed, attr=%s. ret=%d", attr.c_str(), ret); - return ret; + if ((err = parse_control_attribute(attr)) != srs_success) { + return srs_error_wrap(err, "parse control attr=%s", attr.c_str()); } } } @@ -497,12 +481,12 @@ int SrsRtspSdp::parse(string token) default: break; } - return ret; + return err; } -int SrsRtspSdp::parse_fmtp_attribute(string attr) +srs_error_t SrsRtspSdp::parse_fmtp_attribute(string attr) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; size_t pos = string::npos; std::string token = attr; @@ -548,9 +532,7 @@ int SrsRtspSdp::parse_fmtp_attribute(string attr) audio_index_delta_length = item_value; } else if (item_key == "config") { if (item_value.length() <= 0) { - ret = ERROR_RTSP_AUDIO_CONFIG; - srs_error("rtsp: audio config failed. ret=%d", ret); - return ret; + return srs_error_new(ERROR_RTSP_AUDIO_CONFIG, "audio config"); } char* tmp_sh = new char[item_value.length()]; @@ -562,12 +544,12 @@ int SrsRtspSdp::parse_fmtp_attribute(string attr) } } - return ret; + return err; } -int SrsRtspSdp::parse_control_attribute(string attr) +srs_error_t SrsRtspSdp::parse_control_attribute(string attr) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; size_t pos = string::npos; std::string token = attr; @@ -598,7 +580,7 @@ int SrsRtspSdp::parse_control_attribute(string attr) } } - return ret; + return err; } string SrsRtspSdp::base64_decode(string value) @@ -631,9 +613,9 @@ SrsRtspTransport::~SrsRtspTransport() { } -int SrsRtspTransport::parse(string attr) +srs_error_t SrsRtspTransport::parse(string attr) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; size_t pos = string::npos; std::string token = attr; @@ -681,7 +663,7 @@ int SrsRtspTransport::parse(string attr) } } - return ret; + return err; } SrsRtspRequest::SrsRtspRequest() @@ -729,9 +711,9 @@ SrsRtspResponse::~SrsRtspResponse() { } -int SrsRtspResponse::encode(stringstream& ss) +srs_error_t SrsRtspResponse::encode(stringstream& ss) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // status line ss << SRS_RTSP_VERSION << SRS_RTSP_SP @@ -751,20 +733,19 @@ int SrsRtspResponse::encode(stringstream& ss) ss << SRS_RTSP_TOKEN_SESSION << ":" << session << SRS_RTSP_CRLF; } - if ((ret = encode_header(ss)) != ERROR_SUCCESS) { - srs_error("rtsp: encode header failed. ret=%d", ret); - return ret; + if ((err = encode_header(ss)) != srs_success) { + return srs_error_wrap(err, "encode header"); }; // header EOF. ss << SRS_RTSP_CRLF; - return ret; + return err; } -int SrsRtspResponse::encode_header(std::stringstream& ss) +srs_error_t SrsRtspResponse::encode_header(std::stringstream& ss) { - return ERROR_SUCCESS; + return srs_success; } SrsRtspOptionsResponse::SrsRtspOptionsResponse(int cseq) : SrsRtspResponse(cseq) @@ -778,7 +759,7 @@ SrsRtspOptionsResponse::~SrsRtspOptionsResponse() { } -int SrsRtspOptionsResponse::encode_header(stringstream& ss) +srs_error_t SrsRtspOptionsResponse::encode_header(stringstream& ss) { SrsRtspMethod rtsp_methods[] = { SrsRtspMethodDescribe, @@ -812,7 +793,7 @@ int SrsRtspOptionsResponse::encode_header(stringstream& ss) } ss << SRS_RTSP_CRLF; - return ERROR_SUCCESS; + return srs_success; } SrsRtspSetupResponse::SrsRtspSetupResponse(int seq) : SrsRtspResponse(seq) @@ -825,14 +806,14 @@ SrsRtspSetupResponse::~SrsRtspSetupResponse() { } -int SrsRtspSetupResponse::encode_header(stringstream& ss) +srs_error_t SrsRtspSetupResponse::encode_header(stringstream& ss) { ss << SRS_RTSP_TOKEN_SESSION << ":" << SRS_RTSP_SP << session << SRS_RTSP_CRLF; ss << SRS_RTSP_TOKEN_TRANSPORT << ":" << SRS_RTSP_SP << "RTP/AVP;unicast;client_port=" << client_port_min << "-" << client_port_max << ";" << "server_port=" << local_port_min << "-" << local_port_max << SRS_RTSP_CRLF; - return ERROR_SUCCESS; + return srs_success; } SrsRtspStack::SrsRtspStack(ISrsProtocolReaderWriter* s) @@ -846,24 +827,24 @@ SrsRtspStack::~SrsRtspStack() srs_freep(buf); } -int SrsRtspStack::recv_message(SrsRtspRequest** preq) +srs_error_t SrsRtspStack::recv_message(SrsRtspRequest** preq) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsRtspRequest* req = new SrsRtspRequest(); - if ((ret = do_recv_message(req)) != ERROR_SUCCESS) { + if ((err = do_recv_message(req)) != srs_success) { srs_freep(req); - return ret; + return srs_error_wrap(err, "recv message"); } *preq = req; - return ret; + return err; } -int SrsRtspStack::send_message(SrsRtspResponse* res) +srs_error_t SrsRtspStack::send_message(SrsRtspResponse* res) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; std::stringstream ss; // encode the message to string. @@ -872,119 +853,83 @@ int SrsRtspStack::send_message(SrsRtspResponse* res) std::string str = ss.str(); srs_assert(!str.empty()); - if ((ret = skt->write((char*)str.c_str(), (int)str.length(), NULL)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: send response failed. ret=%d", ret); - } - return ret; + if ((err = skt->write((char*)str.c_str(), (int)str.length(), NULL)) != srs_success) { + return srs_error_wrap(err, "write message"); } - srs_info("rtsp: send response ok"); - return ret; + return err; } -int SrsRtspStack::do_recv_message(SrsRtspRequest* req) +srs_error_t SrsRtspStack::do_recv_message(SrsRtspRequest* req) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // parse request line. - if ((ret = recv_token_normal(req->method)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse method failed. ret=%d", ret); - } - return ret; + if ((err = recv_token_normal(req->method)) != srs_success) { + return srs_error_wrap(err, "method"); } - if ((ret = recv_token_normal(req->uri)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse uri failed. ret=%d", ret); - } - return ret; + if ((err = recv_token_normal(req->uri)) != srs_success) { + return srs_error_wrap(err, "uri"); } - if ((ret = recv_token_eof(req->version)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse version failed. ret=%d", ret); - } - return ret; + if ((err = recv_token_eof(req->version)) != srs_success) { + return srs_error_wrap(err, "version"); } // parse headers. for (;;) { // parse the header name std::string token; - if ((ret = recv_token_normal(token)) != ERROR_SUCCESS) { - if (ret == ERROR_RTSP_REQUEST_HEADER_EOF) { - ret = ERROR_SUCCESS; - srs_info("rtsp: message header parsed"); + if ((err = recv_token_normal(token)) != srs_success) { + if (srs_error_code(err) == ERROR_RTSP_REQUEST_HEADER_EOF) { + srs_error_reset(err); break; } - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse token failed. ret=%d", ret); - } - return ret; + return srs_error_wrap(err, "recv token"); } // parse the header value according by header name if (token == SRS_RTSP_TOKEN_CSEQ) { std::string seq; - if ((ret = recv_token_eof(seq)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse %s failed. ret=%d", SRS_RTSP_TOKEN_CSEQ, ret); - } - return ret; + if ((err = recv_token_eof(seq)) != srs_success) { + return srs_error_wrap(err, "seq"); } req->seq = ::atol(seq.c_str()); } else if (token == SRS_RTSP_TOKEN_CONTENT_TYPE) { std::string ct; - if ((ret = recv_token_eof(ct)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse %s failed. ret=%d", SRS_RTSP_TOKEN_CONTENT_TYPE, ret); - } - return ret; + if ((err = recv_token_eof(ct)) != srs_success) { + return srs_error_wrap(err, "ct"); } req->content_type = ct; } else if (token == SRS_RTSP_TOKEN_CONTENT_LENGTH) { std::string cl; - if ((ret = recv_token_eof(cl)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse %s failed. ret=%d", SRS_RTSP_TOKEN_CONTENT_LENGTH, ret); - } - return ret; + if ((err = recv_token_eof(cl)) != srs_success) { + return srs_error_wrap(err, "cl"); } req->content_length = ::atol(cl.c_str()); } else if (token == SRS_RTSP_TOKEN_TRANSPORT) { std::string transport; - if ((ret = recv_token_eof(transport)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse %s failed. ret=%d", SRS_RTSP_TOKEN_TRANSPORT, ret); - } - return ret; + if ((err = recv_token_eof(transport)) != srs_success) { + return srs_error_wrap(err, "transport"); } if (!req->transport) { req->transport = new SrsRtspTransport(); } - if ((ret = req->transport->parse(transport)) != ERROR_SUCCESS) { - srs_error("rtsp: parse transport failed, transport=%s. ret=%d", transport.c_str(), ret); - return ret; + if ((err = req->transport->parse(transport)) != srs_success) { + return srs_error_wrap(err, "parse transport=%s", transport.c_str()); } } else if (token == SRS_RTSP_TOKEN_SESSION) { - if ((ret = recv_token_eof(req->session)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse %s failed. ret=%d", SRS_RTSP_TOKEN_SESSION, ret); - } - return ret; + if ((err = recv_token_eof(req->session)) != srs_success) { + return srs_error_wrap(err, "session"); } } else { // unknown header name, parse util EOF. SrsRtspTokenState state = SrsRtspTokenStateNormal; while (state == SrsRtspTokenStateNormal) { std::string value; - if ((ret = recv_token(value, state)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse token failed. ret=%d", ret); - } - return ret; + if ((err = recv_token(value, state)) != srs_success) { + return srs_error_wrap(err, "state"); } srs_trace("rtsp: ignore header %s=%s", token.c_str(), value.c_str()); } @@ -1011,104 +956,83 @@ int SrsRtspStack::do_recv_message(SrsRtspRequest* req) int nb_token = 0; std::string token; - if ((ret = recv_token_util_eof(token, &nb_token)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse sdp token failed. ret=%d", ret); - } - return ret; + if ((err = recv_token_util_eof(token, &nb_token)) != srs_success) { + return srs_error_wrap(err, "recv token"); } consumed += nb_token; - if ((ret = req->sdp->parse(token)) != ERROR_SUCCESS) { - srs_error("rtsp: sdp parse token failed, token=%s. ret=%d", token.c_str(), ret); - return ret; + if ((err = req->sdp->parse(token)) != srs_success) { + return srs_error_wrap(err, "parse token"); } - srs_info("rtsp: %s", token.c_str()); } - srs_info("rtsp: sdp parsed, size=%d", consumed); - return ret; + return err; } -int SrsRtspStack::recv_token_normal(std::string& token) +srs_error_t SrsRtspStack::recv_token_normal(std::string& token) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsRtspTokenState state; - if ((ret = recv_token(token, state)) != ERROR_SUCCESS) { - if (ret == ERROR_RTSP_REQUEST_HEADER_EOF) { - return ret; + if ((err = recv_token(token, state)) != srs_success) { + if (srs_error_code(err) == ERROR_RTSP_REQUEST_HEADER_EOF) { + return srs_error_wrap(err, "EOF"); } - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse token failed. ret=%d", ret); - } - return ret; + return srs_error_wrap(err, "recv token"); } if (state != SrsRtspTokenStateNormal) { - ret = ERROR_RTSP_TOKEN_NOT_NORMAL; - srs_error("rtsp: parse normal token failed, state=%d. ret=%d", state, ret); - return ret; + return srs_error_new(ERROR_RTSP_TOKEN_NOT_NORMAL, "invalid state=%d", state); } - return ret; + return err; } -int SrsRtspStack::recv_token_eof(std::string& token) +srs_error_t SrsRtspStack::recv_token_eof(std::string& token) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsRtspTokenState state; - if ((ret = recv_token(token, state)) != ERROR_SUCCESS) { - if (ret == ERROR_RTSP_REQUEST_HEADER_EOF) { - return ret; + if ((err = recv_token(token, state)) != srs_success) { + if (srs_error_code(err) == ERROR_RTSP_REQUEST_HEADER_EOF) { + return srs_error_wrap(err, "EOF"); } - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse token failed. ret=%d", ret); - } - return ret; + return srs_error_wrap(err, "recv token"); } if (state != SrsRtspTokenStateEOF) { - ret = ERROR_RTSP_TOKEN_NOT_NORMAL; - srs_error("rtsp: parse eof token failed, state=%d. ret=%d", state, ret); - return ret; + return srs_error_new(ERROR_RTSP_TOKEN_NOT_NORMAL, "invalid state=%d", state); } - return ret; + return err; } -int SrsRtspStack::recv_token_util_eof(std::string& token, int* pconsumed) +srs_error_t SrsRtspStack::recv_token_util_eof(std::string& token, int* pconsumed) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; SrsRtspTokenState state; // use 0x00 as ignore the normal token flag. - if ((ret = recv_token(token, state, 0x00, pconsumed)) != ERROR_SUCCESS) { - if (ret == ERROR_RTSP_REQUEST_HEADER_EOF) { - return ret; + if ((err = recv_token(token, state, 0x00, pconsumed)) != srs_success) { + if (srs_error_code(err) == ERROR_RTSP_REQUEST_HEADER_EOF) { + return srs_error_wrap(err, "EOF"); } - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: parse token failed. ret=%d", ret); - } - return ret; + return srs_error_wrap(err, "recv token"); } if (state != SrsRtspTokenStateEOF) { - ret = ERROR_RTSP_TOKEN_NOT_NORMAL; - srs_error("rtsp: parse eof token failed, state=%d. ret=%d", state, ret); - return ret; + return srs_error_new(ERROR_RTSP_TOKEN_NOT_NORMAL, "invalid state=%d", state); } - return ret; + return err; } -int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char normal_ch, int* pconsumed) +srs_error_t SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char normal_ch, int* pconsumed) { - int ret = ERROR_SUCCESS; + srs_error_t err = srs_success; // whatever, default to error state. state = SrsRtspTokenStateError; @@ -1124,15 +1048,11 @@ int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char char buffer[SRS_RTSP_BUFFER]; ssize_t nb_read = 0; - if ((ret = skt->read(buffer, SRS_RTSP_BUFFER, &nb_read)) != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("rtsp: io read failed. ret=%d", ret); - } - return ret; + if ((err = skt->read(buffer, SRS_RTSP_BUFFER, &nb_read)) != srs_success) { + return srs_error_wrap(err, "recv data"); } - srs_info("rtsp: io read %d bytes", nb_read); - buf->append(buffer, nb_read); + buf->append(buffer, (int)nb_read); } // parse one by one. @@ -1154,7 +1074,7 @@ int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char } // got the token. - int nb_token = p - start; + int nb_token = (int)(p - start); // trim last ':' character. if (nb_token && p[-1] == ':') { nb_token--; @@ -1162,7 +1082,7 @@ int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char if (nb_token) { token.append(start, nb_token); } else { - ret = ERROR_RTSP_REQUEST_HEADER_EOF; + err = srs_error_new(ERROR_RTSP_REQUEST_HEADER_EOF, "EOF"); } // ignore SP/CR/LF @@ -1171,9 +1091,9 @@ int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char // consume the token bytes. srs_assert(p - start); - buf->erase(p - start); + buf->erase((int)(p - start)); if (pconsumed) { - *pconsumed = p - start; + *pconsumed = (int)(p - start); } break; } @@ -1182,7 +1102,7 @@ int SrsRtspStack::recv_token(std::string& token, SrsRtspTokenState& state, char append_bytes = true; } - return ret; + return err; } #endif diff --git a/trunk/src/protocol/srs_rtsp_stack.hpp b/trunk/src/protocol/srs_rtsp_stack.hpp index a13ccf0c9..0a77775c5 100644 --- a/trunk/src/protocol/srs_rtsp_stack.hpp +++ b/trunk/src/protocol/srs_rtsp_stack.hpp @@ -311,10 +311,10 @@ public: /** * decode rtp packet from stream. */ - virtual int decode(SrsBuffer* stream); + virtual srs_error_t decode(SrsBuffer* stream); private: - virtual int decode_97(SrsBuffer* stream); - virtual int decode_96(SrsBuffer* stream); + virtual srs_error_t decode_97(SrsBuffer* stream); + virtual srs_error_t decode_96(SrsBuffer* stream); }; /** @@ -394,16 +394,16 @@ public: /** * parse a line of token for sdp. */ - virtual int parse(std::string token); + virtual srs_error_t parse(std::string token); private: /** * generally, the fmtp is the sequence header for video or audio. */ - virtual int parse_fmtp_attribute(std::string attr); + virtual srs_error_t parse_fmtp_attribute(std::string attr); /** * generally, the control is the stream info for video or audio. */ - virtual int parse_control_attribute(std::string attr); + virtual srs_error_t parse_control_attribute(std::string attr); /** * decode the string by base64. */ @@ -454,7 +454,7 @@ public: /** * parse a line of token for transport. */ - virtual int parse(std::string attr); + virtual srs_error_t parse(std::string attr); }; /** @@ -588,12 +588,12 @@ public: /** * encode message to string. */ - virtual int encode(std::stringstream& ss); + virtual srs_error_t encode(std::stringstream& ss); protected: /** * sub classes override this to encode the headers. */ - virtual int encode_header(std::stringstream& ss); + virtual srs_error_t encode_header(std::stringstream& ss); }; /** @@ -613,7 +613,7 @@ public: SrsRtspOptionsResponse(int cseq); virtual ~SrsRtspOptionsResponse(); protected: - virtual int encode_header(std::stringstream& ss); + virtual srs_error_t encode_header(std::stringstream& ss); }; /** @@ -643,7 +643,7 @@ public: SrsRtspSetupResponse(int cseq); virtual ~SrsRtspSetupResponse(); protected: - virtual int encode_header(std::stringstream& ss); + virtual srs_error_t encode_header(std::stringstream& ss); }; /** @@ -670,31 +670,31 @@ public: * @return an int error code. * ERROR_RTSP_REQUEST_HEADER_EOF indicates request header EOF. */ - virtual int recv_message(SrsRtspRequest** preq); + virtual srs_error_t recv_message(SrsRtspRequest** preq); /** * send rtsp message over underlayer io. * @param res the rtsp response message, which user should never free it. * @return an int error code. */ - virtual int send_message(SrsRtspResponse* res); + virtual srs_error_t send_message(SrsRtspResponse* res); private: /** * recv the rtsp message. */ - virtual int do_recv_message(SrsRtspRequest* req); + virtual srs_error_t do_recv_message(SrsRtspRequest* req); /** * read a normal token from io, error when token state is not normal. */ - virtual int recv_token_normal(std::string& token); + virtual srs_error_t recv_token_normal(std::string& token); /** * read a normal token from io, error when token state is not eof. */ - virtual int recv_token_eof(std::string& token); + virtual srs_error_t recv_token_eof(std::string& token); /** * read the token util got eof, for example, to read the response status Reason-Phrase * @param pconsumed, output the token parsed length. NULL to ignore. */ - virtual int recv_token_util_eof(std::string& token, int* pconsumed = NULL); + virtual srs_error_t recv_token_util_eof(std::string& token, int* pconsumed = NULL); /** * read a token from io, split by SP, endswith CRLF: * token1 SP token2 SP ... tokenN CRLF @@ -705,7 +705,7 @@ private: * the 0x00 use to ignore normal token flag. @see recv_token_util_eof * @param pconsumed, output the token parsed length. NULL to ignore. */ - virtual int recv_token(std::string& token, SrsRtspTokenState& state, char normal_ch = SRS_RTSP_SP, int* pconsumed = NULL); + virtual srs_error_t recv_token(std::string& token, SrsRtspTokenState& state, char normal_ch = SRS_RTSP_SP, int* pconsumed = NULL); }; #endif