| 291 | } |
| 292 | |
| 293 | int mqtt_aclient::handle_data(char* data, int len) { |
| 294 | int left; |
| 295 | |
| 296 | if (!header_->finished()) { |
| 297 | left = header_->update(data, len); |
| 298 | } else { |
| 299 | left = len; |
| 300 | } |
| 301 | |
| 302 | if (left < 0) { |
| 303 | logger_error("header update failed"); |
| 304 | return -1; |
| 305 | } |
| 306 | |
| 307 | if (!header_->finished()) { |
| 308 | assert(left == 0); |
| 309 | return 0; |
| 310 | } |
| 311 | |
| 312 | data += len - left; |
| 313 | len = left; |
| 314 | |
| 315 | if (body_ == NULL) { |
| 316 | body_ = mqtt_message::create_message(*header_); |
| 317 | if (body_ == NULL) { |
| 318 | logger_error("create mqtt_message failed"); |
| 319 | return -1; |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | if (len > 0) { |
| 324 | left = body_->update(data, len); |
| 325 | if (left < 0) { |
| 326 | logger_error("message update failed"); |
| 327 | return -1; |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | if (body_->finished()) { |
| 332 | bool ret = this->on_body(*body_); |
| 333 | header_->reset(); |
| 334 | delete body_; |
| 335 | body_ = NULL; |
| 336 | |
| 337 | if (!ret) { |
| 338 | logger_error("subclass return false"); |
| 339 | return -1; |
| 340 | } |
| 341 | } |
| 342 | |
| 343 | return left; |
| 344 | } |
| 345 | |
| 346 | } // namespace acl |