| 680 | } |
| 681 | |
| 682 | void redis_client_pipeline::channel_closed(redis_pipeline_channel* channel) |
| 683 | { |
| 684 | if (channel == NULL) { |
| 685 | logger_error("The channel null!"); |
| 686 | return; |
| 687 | } |
| 688 | |
| 689 | const char* addr = channel->get_addr(); |
| 690 | const token_node* node = channels_->find(addr); |
| 691 | |
| 692 | // Clear all slots' addrs same as the dead node |
| 693 | for (size_t i = 0; i < max_slot_; i++) { |
| 694 | if (slot_addrs_[i] && strcmp(addr, slot_addrs_[i]) == 0) { |
| 695 | slot_addrs_[i] = NULL; |
| 696 | } |
| 697 | } |
| 698 | |
| 699 | // Reset the default addr which different from the dead node |
| 700 | if (addr_ == addr) { |
| 701 | for (std::vector<char*>::const_iterator it = addrs_.begin(); |
| 702 | it != addrs_.end(); ++it) { |
| 703 | if (strcmp(addr, *it) != 0) { |
| 704 | addr_ = *it; |
| 705 | break; |
| 706 | } |
| 707 | } |
| 708 | } |
| 709 | |
| 710 | if (node == NULL) { |
| 711 | logger("The channel not found, addr=%s", addr); |
| 712 | channel->wait(); // Wait the thread to exit. |
| 713 | delete channel; |
| 714 | return; |
| 715 | } |
| 716 | |
| 717 | redis_pipeline_channel* chan = |
| 718 | static_cast<redis_pipeline_channel*>(node->get_ctx()); |
| 719 | if (chan == NULL || chan != channel) { |
| 720 | logger_warn("The channel=%p not mine=%p", channel, chan); |
| 721 | } |
| 722 | |
| 723 | logger("The channel closed, addr=%s", addr); |
| 724 | channels_->remove(addr); |
| 725 | channel->wait(); |
| 726 | delete channel; |
| 727 | } |
| 728 | |
| 729 | void redis_client_pipeline::redirect(const redis_pipeline_message &msg, size_t slot) |
| 730 | { |